115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
"""
|
|
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)
|