#!/usr/bin/env python3 """Idempotently import prior no-action weekly reviews as Catalog decisions.""" from __future__ import annotations import argparse import json import sys from datetime import datetime, time, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from backend.sql_catalog import server # noqa: E402 DEFAULT_DATASET = ROOT / "data" / "evaluation" / "manual_weekly_review_cases.json" def review_cutoff(review_date: str) -> str: reviewed_day = datetime.fromisoformat(review_date).date() return datetime.combine(reviewed_day, time.max, timezone.utc).isoformat() def existing_record_id(item_id: str, case_id: str) -> str | None: with server.get_conn() as conn: rows = conn.execute( """ SELECT records.* FROM records JOIN relations ON relations.from_type = 'record' AND relations.from_id = records.id AND relations.relation_type = 'record_of' AND relations.to_type = 'catalog_item' AND relations.to_id = ? ORDER BY records.created_at DESC """, (item_id,), ).fetchall() for row in rows: record = server.row_to_record(row) if record["structured_fields"].get("manual_review_case_id") == case_id: return str(record["id"]) return None def existing_record(item_id: str, case_id: str) -> dict[str, Any] | None: record_id = existing_record_id(item_id, case_id) if not record_id: return None with server.get_conn() as conn: return server.get_record(conn, record_id) def find_item(case: dict[str, Any]) -> dict[str, Any] | None: with server.get_conn() as conn: row = conn.execute( """ SELECT * FROM catalog_items WHERE cluster_id = ? AND sql_digest = ? ORDER BY created_at DESC LIMIT 1 """, (case["cluster_id"], case["sql_digest"]), ).fetchone() return server.row_to_catalog_item(row) if row else None def record_body(case: dict[str, Any], source_url: str) -> str: conditions = case.get("reopen_conditions") or [case.get("reopen_policy")] condition_lines = "\n".join(f"- {condition}" for condition in conditions if condition) return ( f"## Decision\n\n{case['human_disposition']}\n\n" f"## Rationale\n\n{case['rationale']}\n\n" f"## Re-evaluation conditions\n\n{condition_lines or '- Manual review'}\n\n" f"## Source\n\nWeekly review {case['review_date']}: {source_url}\n" ) def import_cases(dataset_path: Path, *, apply: bool) -> dict[str, Any]: dataset = json.loads(dataset_path.read_text(encoding="utf-8")) source_url = str(dataset.get("source", {}).get("url") or "") results: list[dict[str, Any]] = [] for case in dataset.get("cases", []): if case.get("decision_action") != "no_action": continue item = find_item(case) if not item: results.append({"id": case["id"], "status": "catalog_item_missing"}) continue current_record = existing_record(item["id"], case["id"]) record_id = str(current_record["id"]) if current_record else None expected_baseline = case.get("baseline") or {} baseline_matches = bool( not expected_baseline or (current_record or {}).get("structured_fields", {}).get("baseline") == expected_baseline ) if record_id and baseline_matches: results.append( {"id": case["id"], "status": "already_imported", "record_id": record_id} ) continue if record_id and not baseline_matches: if not apply: results.append( { "id": case["id"], "status": "ready_to_refresh", "record_id": record_id, "item_id": item["id"], } ) continue structured_fields = { **(current_record or {}).get("structured_fields", {}), "conclusion_type": case["human_disposition"], "confidence": "high", "reopen_policy": case.get("reopen_policy") or "meaningful_change", "metric_guardrails": case.get("metric_guardrails") or {}, "baseline": expected_baseline, "manual_review_case_id": case["id"], "source_review_date": case["review_date"], "source_block_id": case.get("source_block_id"), } status, updated = server.update_record( record_id, { "structured_fields": structured_fields, "body_markdown": record_body(case, source_url), "actor": "weekly-review-import", }, ) if status != 200: results.append( {"id": case["id"], "status": "record_error", "detail": updated} ) continue status, decision = server.triage_catalog_item( item["id"], { "action": "no_action", "record_id": record_id, "reviewed_through_at": review_cutoff(case["review_date"]), "actor": "weekly-review-import", }, ) results.append( { "id": case["id"], "status": "refreshed" if status == 200 else "decision_error", "record_id": record_id, "detail": None if status == 200 else decision, } ) continue if not apply: results.append({"id": case["id"], "status": "ready", "item_id": item["id"]}) continue status, created = server.create_record( "catalog_item", item["id"], { "record_type": "no_action_record", "title": f"Weekly review: {case['human_disposition'].replace('_', ' ')}", "review_status": "reviewed", "visibility": "internal", "structured_fields": { "conclusion_type": case["human_disposition"], "confidence": "high", "reopen_policy": case.get("reopen_policy") or "meaningful_change", "metric_guardrails": case.get("metric_guardrails") or {}, "baseline": case.get("baseline") or {}, "manual_review_case_id": case["id"], "source_review_date": case["review_date"], "source_block_id": case.get("source_block_id"), }, "body_markdown": record_body(case, source_url), "created_by": "weekly-review-import", }, ) if status != 201: results.append({"id": case["id"], "status": "record_error", "detail": created}) continue record_id = created["record"]["id"] status, decision = server.triage_catalog_item( item["id"], { "action": "no_action", "record_id": record_id, "reviewed_through_at": review_cutoff(case["review_date"]), "actor": "weekly-review-import", }, ) results.append( { "id": case["id"], "status": "imported" if status == 200 else "decision_error", "record_id": record_id, "detail": None if status == 200 else decision, } ) return { "mode": "apply" if apply else "preview", "database": str(server.DB_PATH), "results": results, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--dataset", type=Path, default=DEFAULT_DATASET) parser.add_argument("--apply", action="store_true") args = parser.parse_args() server.init_db() report = import_cases(args.dataset.resolve(), apply=args.apply) print(json.dumps(report, ensure_ascii=False, indent=2)) has_error = any(result["status"].endswith("error") for result in report["results"]) return 1 if has_error else 0 if __name__ == "__main__": raise SystemExit(main())