feat: track media catalog coverage

rf4_research/media_assets.py

@@ -162,6 +162,25 @@ def audit_media_catalog(root: Path) -> dict:
     return {"total": sum(statuses.values()), "statuses": statuses, "issues": issues, "orphaned_files": orphaned}
 
 
+def media_coverage(root: Path) -> dict:
+    """Compare quarantined and approved media with the versioned catalog baseline."""
+    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
+    baseline = json.loads((root / "catalog-baseline.json").read_text(encoding="utf-8"))
+    result: dict[str, dict] = {}
+    for entity_type, target in baseline["entities"].items():
+        candidates = [item for item in manifest.get("assets", []) if item.get("entity_type") == entity_type]
+        approved_keys = {item.get("entity_key") for item in candidates if item.get("status") == "approved" and item.get("entity_key")}
+        expected = target.get("count")
+        result[entity_type] = {
+            "expected": expected,
+            "candidates": len(candidates),
+            "approved": len(approved_keys),
+            "candidate_gap": max(expected - len(candidates), 0) if isinstance(expected, int) else None,
+            "approved_gap": max(expected - len(approved_keys), 0) if isinstance(expected, int) else None,
+        }
+    return {"baseline_date": baseline["verified_at"], "entities": result}
+
+
 def inspect_image(body: bytes) -> tuple[int, int, str]:
     try:
         with Image.open(io.BytesIO(body)) as image:

rf4_research/media_cli.py

@@ -8,7 +8,7 @@ import urllib.error
 import urllib.request
 
 from .community_cli import USER_AGENT, _StrictRedirectHandler, _validate_url_before_io, check_and_reserve, fetch_html, fetch_site_key
-from .media_assets import audit_media_catalog, extract_media_candidates, merge_manifest, reclassify_manifest, review_asset, store_asset
+from .media_assets import audit_media_catalog, extract_media_candidates, media_coverage, merge_manifest, reclassify_manifest, review_asset, store_asset
 
 
 DEFAULT_ROOT = Path("data/media")
@@ -59,6 +59,7 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--entity-key")
     parser.add_argument("--note")
     parser.add_argument("--audit", action="store_true", help="Verify manifest metadata, hashes and local files without network access")
+    parser.add_argument("--coverage", action="store_true", help="Compare candidates and approvals with the catalog baseline")
     parser.add_argument("--root", type=Path, default=DEFAULT_ROOT)
     parser.add_argument("--state-file", type=Path, default=Path(".cache/community-fetch-state.json"))
     args = parser.parse_args(argv)
@@ -67,6 +68,9 @@ def main(argv: list[str] | None = None) -> int:
             report = audit_media_catalog(args.root)
             print(json.dumps(report, ensure_ascii=False, indent=2))
             return 1 if report["issues"] or report["orphaned_files"] else 0
+        if args.coverage:
+            print(json.dumps(media_coverage(args.root), ensure_ascii=False, indent=2))
+            return 0
         if args.review_url:
             if not args.decision:
                 parser.error("--decision is required with --review-url")