Story 7

src/review_export.py

@@ -0,0 +1,114 @@
+"""
+Review queue export for backend scripts.
+
+Provides CSV and JSON export of annotated review data, with optional
+filtering by date range, model version, annotation status, and ground truth.
+Can be called standalone or imported from other src/ modules.
+"""
+
+import csv
+import io
+import json
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def fetch_annotated(
+    db_connector,
+    annotated_only: bool = True,
+    model_version: Optional[str] = None,
+    date_from: Optional[str] = None,
+    date_to: Optional[str] = None,
+    ground_truth: Optional[bool] = None,
+) -> List[Dict[str, Any]]:
+    """Query the review_queue and return matching records as plain dicts."""
+    clauses: List[str] = []
+    params: List[Any] = []
+
+    if annotated_only:
+        clauses.append("rq.annotated = TRUE")
+    if model_version:
+        clauses.append("v.model_version = %s")
+        params.append(model_version)
+    if date_from:
+        clauses.append("rq.annotated_at >= %s")
+        params.append(date_from)
+    if date_to:
+        clauses.append("rq.annotated_at < %s")
+        params.append(date_to)
+    if ground_truth is not None:
+        clauses.append("rq.ground_truth = %s")
+        params.append(bool(ground_truth))
+
+    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
+
+    rows = db_connector.fetchall(
+        f"""SELECT rq.video_id, v.file_path, rq.confidence_score, rq.routing_decision,
+                   v.model_version, rq.ground_truth, rq.annotated_at, rq.notes,
+                   pl.confidence_scores
+            FROM review_queue rq
+            JOIN videos v ON v.id = rq.video_id
+            LEFT JOIN (
+                SELECT video_id, confidence_scores,
+                       ROW_NUMBER() OVER (PARTITION BY video_id ORDER BY processed_at DESC) rn
+                FROM processing_logs
+            ) pl ON pl.video_id = rq.video_id AND pl.rn = 1
+            {where}
+            ORDER BY rq.annotated_at DESC""",
+        params if params else None,
+    )
+
+    return [_normalise(row) for row in (rows or [])]
+
+
+def _normalise(row: dict) -> Dict[str, Any]:
+    try:
+        scores = json.loads(row["confidence_scores"]) if row.get("confidence_scores") else []
+    except (json.JSONDecodeError, TypeError):
+        scores = []
+    at = row.get("annotated_at")
+    return {
+        "video_id": row["video_id"],
+        "file_path": row["file_path"],
+        "confidence_score": row["confidence_score"],
+        "routing_decision": row["routing_decision"],
+        "model_version": row.get("model_version"),
+        "ground_truth": bool(row["ground_truth"]) if row["ground_truth"] is not None else None,
+        "annotated_at": at.isoformat() if isinstance(at, datetime) else at,
+        "notes": row.get("notes"),
+        "contributing_frames": scores,
+    }
+
+
+def export_json(records: List[Dict[str, Any]], output_path: str) -> str:
+    """Write records to a UTF-8 JSON file; returns the path."""
+    path = Path(output_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(records, default=str, indent=2), encoding="utf-8")
+    logger.info("Exported %d records to %s", len(records), path)
+    return str(path)
+
+
+def export_csv(records: List[Dict[str, Any]], output_path: str) -> str:
+    """Write records to a UTF-8 CSV file; returns the path."""
+    path = Path(output_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if not records:
+        path.write_text("", encoding="utf-8")
+        return str(path)
+
+    buf = io.StringIO()
+    writer = csv.DictWriter(buf, fieldnames=list(records[0].keys()))
+    writer.writeheader()
+    for r in records:
+        row = dict(r)
+        row["contributing_frames"] = json.dumps(row["contributing_frames"])
+        writer.writerow(row)
+
+    path.write_text(buf.getvalue(), encoding="utf-8")
+    logger.info("Exported %d records to %s", len(records), path)
+    return str(path)