1"""Periodic quality/coverage report and garbage collection."""2from __future__ import annotations34import json5from typing import Any67from findata_core.db import connection, transaction89from findata_ingest.lake import Lake101112def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:13 dataset_id = job.get("dataset_id") or "market_bars.stock.1m.raw"14 with connection() as conn:15 tot = conn.execute("SELECT count(*) n, sum(bar_count) bars, count(DISTINCT symbol) syms, min(session_date) d0, max(session_date) d1 FROM coverage WHERE dataset_id=%s", (dataset_id,)).fetchone()16 comp = conn.execute("SELECT avg(LEAST(1.0, bar_count::float/NULLIF(expected_bars,0))) AS c FROM coverage WHERE dataset_id=%s AND expected_bars>0", (dataset_id,)).fetchone()17 anomalies = conn.execute("SELECT code, severity, count(*) n FROM quality_anomalies WHERE dataset_id=%s AND created_at > now() - interval '30 days' GROUP BY 1,2 ORDER BY n DESC", (dataset_id,)).fetchall()18 summary = {"dataset_id": dataset_id, "symbol_sessions": tot["n"], "bars": int(tot["bars"] or 0), "symbols": tot["syms"],19 "first_session": str(tot["d0"]), "last_session": str(tot["d1"]), "mean_regular_hours_completeness": float(comp["c"] or 0) if comp and comp["c"] is not None else None,20 "anomalies_30d": [dict(a) for a in anomalies]}21 with transaction() as conn:22 conn.execute("INSERT INTO quality_reports(dataset_id, ingest_run_id, rows_checked, anomalies, summary) VALUES (%s,%s,%s,%s,%s)",23 (dataset_id, job.get("run_id"), summary["bars"], sum(a["n"] for a in anomalies), json.dumps(summary, default=str)))24 return summary252627def gc(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:28 n = lake.gc_retired(7)29 orphans = lake.gc_orphans(2)30 removed = 031 with transaction() as conn:32 rows = conn.execute("UPDATE exports SET status='expired' WHERE status='ready' AND expires_at < now() RETURNING path").fetchall()33 for r in rows:34 if r["path"]:35 p = lake.abs(r["path"])36 if p.exists():37 p.unlink()38 removed += 139 conn.execute("DELETE FROM sessions WHERE expires_at < now() - interval '7 days'")40 return {"retired_files_deleted": n, "orphan_files_deleted": orphans, "exports_expired": len(rows), "export_files_removed": removed}41