chore: audit RF4 media quality

rf4_research/media_assets.py

@@ -251,6 +251,45 @@ def media_coverage(root: Path) -> dict:
     return {"baseline_date": baseline["verified_at"], "entities": result}
 
 
+def media_quality_report(root: Path, *, minimum_dimension: int = 256, display_dimension: int = 180) -> dict:
+    """Report published raster quality and known higher-quality source alternatives offline."""
+    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
+    assets = manifest.get("assets", [])
+    approved = [item for item in assets if item.get("status") == "approved" and item.get("entity_type") == "fish"]
+    low = [item for item in approved if min(int(item.get("width") or 0), int(item.get("height") or 0)) < minimum_dimension]
+    upsampled = [item for item in approved if min(int(item.get("width") or 0), int(item.get("height") or 0)) < display_dimension]
+    low_urls = {item.get("asset_url") for item in low}
+    alternatives = [
+        item for item in assets
+        if item.get("entity_type") == "fish"
+        and item.get("status") in {"duplicate", "queued"}
+        and (item.get("duplicate_of") in low_urls or item.get("status") == "queued")
+    ]
+
+    by_source: dict[str, dict[str, int]] = {}
+    for item in approved:
+        host = urlsplit(str(item.get("asset_url") or "")).hostname or "unknown"
+        summary = by_source.setdefault(host, {"published": 0, "below_minimum": 0, "upsampled_in_cards": 0})
+        summary["published"] += 1
+        summary["below_minimum"] += item in low
+        summary["upsampled_in_cards"] += item in upsampled
+
+    return {
+        "entity_type": "fish",
+        "minimum_dimension": minimum_dimension,
+        "display_dimension": display_dimension,
+        "published": len(approved),
+        "below_minimum": len(low),
+        "upsampled_in_cards": len(upsampled),
+        "known_alternative_urls": len(alternatives),
+        "by_source": by_source,
+        "examples": [
+            {"label": item.get("label"), "width": item.get("width"), "height": item.get("height"), "asset_url": item.get("asset_url")}
+            for item in sorted(low, key=lambda row: str(row.get("label") or ""))[:20]
+        ],
+    }
+
+
 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 MIN_FETCH_INTERVAL_SECONDS, USER_AGENT, _StrictRedirectHandler, _read_state, _validate_url_before_io, check_and_reserve, fetch_html, fetch_site_key
-from .media_assets import approve_stored_assets, audit_media_catalog, extract_media_candidates, media_coverage, merge_manifest, reconcile_queued_duplicates, reclassify_manifest, review_asset, store_asset
+from .media_assets import approve_stored_assets, audit_media_catalog, extract_media_candidates, media_coverage, media_quality_report, merge_manifest, reconcile_queued_duplicates, reclassify_manifest, review_asset, store_asset
 
 
 DEFAULT_ROOT = Path("data/media")
@@ -190,6 +190,7 @@ def main(argv: list[str] | None = None) -> int:
     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("--quality-report", action="store_true", help="Report low-resolution published fish and known alternatives without network access")
     parser.add_argument("--queue-plan", action="store_true", help="Show the next useful queued asset per domain without network access")
     parser.add_argument("--root", type=Path, default=DEFAULT_ROOT)
     parser.add_argument("--state-file", type=Path, default=Path(".cache/community-fetch-state.json"))
@@ -202,6 +203,9 @@ def main(argv: list[str] | None = None) -> int:
         if args.coverage:
             print(json.dumps(media_coverage(args.root), ensure_ascii=False, indent=2))
             return 0
+        if args.quality_report:
+            print(json.dumps(media_quality_report(args.root), ensure_ascii=False, indent=2))
+            return 0
         if args.queue_plan:
             print(json.dumps(media_queue_plan(args.root, args.state_file), ensure_ascii=False, indent=2))
             return 0