feat: plan media queue without network access

rf4_research/media_cli.py

@@ -7,12 +7,63 @@ from pathlib import Path
 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 .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 audit_media_catalog, extract_media_candidates, media_coverage, merge_manifest, reclassify_manifest, review_asset, store_asset
 
 
 DEFAULT_ROOT = Path("data/media")
 MAX_ASSET_BYTES = 15 * 1024 * 1024
+MEDIA_PRIORITY = {"waterbody": 0, "fish": 1, "tackle": 2, "reference": 3}
+
+
+def media_queue_plan(root: Path, state_file: Path, *, now: float | None = None) -> dict:
+    """Plan the next useful download per domain without network access."""
+    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
+    state = _read_state(state_file)
+    current = datetime.now(timezone.utc).timestamp() if now is None else now
+    grouped: dict[str, list[dict]] = {}
+    for item in manifest.get("assets", []):
+        if item.get("status") == "queued":
+            grouped.setdefault(fetch_site_key(item["asset_url"]), []).append(item)
+
+    domains: dict[str, dict] = {}
+    for site, items in sorted(grouped.items()):
+        items.sort(key=lambda item: (
+            MEDIA_PRIORITY.get(str(item.get("entity_type")), 99),
+            not bool(item.get("label")), str(item.get("label") or "").casefold(),
+            str(item.get("asset_url")),
+        ))
+        last_attempt = state.get(site)
+        retry_in = max(0, round(float(last_attempt) + MIN_FETCH_INTERVAL_SECONDS - current)) if isinstance(last_attempt, (int, float)) else 0
+        next_item = items[0]
+        queued_by_type: dict[str, int] = {}
+        for item in items:
+            entity_type = str(item.get("entity_type") or "unknown")
+            queued_by_type[entity_type] = queued_by_type.get(entity_type, 0) + 1
+        domains[site] = {
+            "queued": len(items),
+            "queued_by_type": dict(sorted(queued_by_type.items())),
+            "ready": retry_in == 0,
+            "retry_in_seconds": retry_in,
+            "next_asset": {
+                "asset_url": next_item["asset_url"],
+                "entity_type": next_item.get("entity_type"),
+                "label": next_item.get("label"),
+                "external_id": next_item.get("external_id"),
+                "source_page": next_item.get("source_page"),
+            },
+        }
+    coverage = media_coverage(root)
+    return {
+        "generated_at": datetime.fromtimestamp(current, timezone.utc).isoformat(),
+        "network_requests": 0,
+        "cooldown_seconds": MIN_FETCH_INTERVAL_SECONDS,
+        "queued_total": sum(group["queued"] for group in domains.values()),
+        "domains": domains,
+        "coverage": coverage["entities"],
+        "exact_catalog_gaps_known": False,
+        "catalog_gap_note": "Baseline stores verified totals, not a canonical name list; exact missing entity names cannot be claimed yet.",
+    }
 
 
 def _download_one(root: Path, state_file: Path, asset_url: str | None = None) -> str:
@@ -60,6 +111,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("--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"))
     args = parser.parse_args(argv)
@@ -71,6 +123,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.queue_plan:
+            print(json.dumps(media_queue_plan(args.root, args.state_file), ensure_ascii=False, indent=2))
+            return 0
         if args.review_url:
             if not args.decision:
                 parser.error("--decision is required with --review-url")