feat: audit full alpha catalog

apps/api/app/catalog_audit.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session
+
+from .models import CatchReport, ExternalObservation, Fish, SourceType, Spot, Waterbody
+
+
+def audit_catalog(db: Session) -> dict[str, int]:
+    count = lambda model: db.scalar(select(func.count()).select_from(model)) or 0
+    failures = {
+        "invalid_weights": db.scalar(select(func.count()).select_from(CatchReport).where(or_(CatchReport.weight_g <= 0, CatchReport.weight_g > 3_000_000))) or 0,
+        "invalid_coordinates": db.scalar(select(func.count()).select_from(Spot).where(or_(Spot.x < -10_000, Spot.x > 10_000, Spot.y < -10_000, Spot.y > 10_000))) or 0,
+        "incomplete_official_records": db.scalar(select(func.count()).select_from(CatchReport).where(CatchReport.source_type == SourceType.official_record, or_(CatchReport.caught_at.is_(None), CatchReport.source_url.is_(None)))) or 0,
+        "incomplete_published_staging": db.scalar(select(func.count()).select_from(ExternalObservation).where(ExternalObservation.status == "published", or_(ExternalObservation.fish_id.is_(None), ExternalObservation.waterbody_id.is_(None), ExternalObservation.x.is_(None), ExternalObservation.y.is_(None), ExternalObservation.weight_g.is_(None), ExternalObservation.catch_report_id.is_(None)))) or 0,
+    }
+    return {"fishes": count(Fish), "waterbodies": count(Waterbody), "reports": count(CatchReport), "staging": count(ExternalObservation), **failures, "failures": sum(failures.values())}

apps/api/app/cli.py

@@ -11,6 +11,7 @@ from .importer import import_records
 from .community_importer import stage_observations
 from .retention import RetentionPolicy, apply_retention
 from .storage import delete_screenshot
+from .catalog_audit import audit_catalog
 
 
 def main() -> int:
@@ -25,6 +26,7 @@ def main() -> int:
     community.add_argument("--limit", type=int, default=500)
     cleanup = sub.add_parser("cleanup-retention")
     cleanup.add_argument("--apply", action="store_true", help="apply changes; default is dry-run")
+    sub.add_parser("audit-catalog")
     args = parser.parse_args()
     with SessionLocal() as session:
         if args.command == "import-records":
@@ -43,7 +45,7 @@ def main() -> int:
                 parser.error("input must be a JSON array")
             created, updated = stage_observations(session, payload[:args.limit])
             print(f"staged: created={created} updated={updated}")
-        else:
+        elif args.command == "cleanup-retention":
             policy = RetentionPolicy(
                 submission_days=settings.retention_submission_days,
                 unreviewed_days=settings.retention_unreviewed_days,
@@ -54,6 +56,10 @@ def main() -> int:
             )
             counts = apply_retention(session, policy=policy, dry_run=not args.apply, delete_object=delete_screenshot)
             print(json.dumps({"mode": "apply" if args.apply else "dry-run", "policy": asdict(policy), "counts": counts}, ensure_ascii=False))
+        else:
+            result = audit_catalog(session)
+            print(json.dumps(result, ensure_ascii=False))
+            return 1 if result["failures"] else 0
     return 0