43 lines
2.6 KiB
Python
43 lines
2.6 KiB
Python
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())}
|
|
|
|
|
|
def audit_waterbody_catalog(db: Session, expected_ids: set[str]) -> dict:
|
|
"""Check a verified RF4DB snapshot without withdrawing legacy rows."""
|
|
rows = list(db.scalars(select(Waterbody).where(Waterbody.source_system == "rf4db")))
|
|
observed_ids = [str(row.source_external_id) for row in rows if row.source_external_id]
|
|
observed = set(observed_ids)
|
|
duplicate_ids = sorted({item for item in observed_ids if observed_ids.count(item) > 1})
|
|
missing = sorted(expected_ids - observed)
|
|
unexpected = sorted(observed - expected_ids)
|
|
provenance_issues = sorted(
|
|
str(row.source_external_id)
|
|
for row in rows
|
|
if not row.source_external_id or not row.source_url or not row.source_checked_at
|
|
)
|
|
failures = len(missing) + len(duplicate_ids) + len(provenance_issues)
|
|
return {
|
|
"expected": len(expected_ids),
|
|
"observed": len(observed),
|
|
"missing_source_external_ids": missing,
|
|
"unexpected_source_external_ids": unexpected,
|
|
"duplicate_source_external_ids": duplicate_ids,
|
|
"provenance_issues": provenance_issues,
|
|
"failures": failures,
|
|
}
|