from __future__ import annotations import argparse from datetime import datetime, timezone import json from pathlib import Path 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 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: manifest_path = root / "manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) queued = next((item for item in manifest["assets"] if item.get("status") == "queued" and (asset_url is None or item["asset_url"] == asset_url)), None) if queued is None: return "queue is empty" if asset_url is None else "asset is not queued" url = queued["asset_url"] _validate_url_before_io(url) check_and_reserve(fetch_site_key(url), state_file=state_file) attempted_at = datetime.now(timezone.utc).isoformat() try: request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "image/*"}) opener = urllib.request.build_opener(_StrictRedirectHandler()) with opener.open(request, timeout=30) as response: content_type = response.headers.get_content_type() if not content_type.startswith("image/"): raise ValueError(f"expected image, got {content_type}") body = response.read(MAX_ASSET_BYTES + 1) if len(body) > MAX_ASSET_BYTES: raise ValueError("asset exceeded 15MB limit") digest, relative, width, height, detected_mime = store_asset(root, body, content_type=content_type, source_url=url) except Exception as exc: code = exc.code if isinstance(exc, urllib.error.HTTPError) else None status = "missing" if code in {404, 410} else "blocked" if code in {401, 403} else "invalid" if isinstance(exc, ValueError) else "queued" queued.update({"status": status, "last_attempt_at": attempted_at, "last_error": str(exc)[:500]}) manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") raise queued.update({"status": "stored", "sha256": digest, "local_path": relative, "content_type": detected_mime, "bytes": len(body), "width": width, "height": height, "fetched_at": attempted_at}) manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return f"stored {url} as {relative}" def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Index authorized RF4 media without hotlinking") parser.add_argument("url", nargs="?") parser.add_argument("--download-one", action="store_true", help="Store one queued asset while respecting site cooldown") parser.add_argument("--asset-url", help="Download this exact queued asset instead of the first queue item") parser.add_argument("--reclassify", action="store_true", help="Reapply current conservative classifier without network access") parser.add_argument("--review-url", help="Review an asset already present in the manifest") parser.add_argument("--decision", choices=("approved", "rejected")) parser.add_argument("--entity-type", choices=("fish", "waterbody", "tackle", "reference")) 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("--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) try: if args.audit: 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.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") item = review_asset(args.root / "manifest.json", asset_url=args.review_url, decision=args.decision, entity_type=args.entity_type, entity_key=args.entity_key, note=args.note) print(f"{item['status']} {item['asset_url']}") return 0 if args.reclassify: manifest = reclassify_manifest(args.root / "manifest.json") print(f"reclassified {len(manifest['assets'])} candidates") return 0 if args.download_one: print(_download_one(args.root, args.state_file, args.asset_url)) return 0 if not args.url: parser.error("url is required unless --download-one is used") check_and_reserve(fetch_site_key(args.url), state_file=args.state_file) html = fetch_html(args.url) candidates = extract_media_candidates(html, source_page=args.url) manifest = merge_manifest(args.root / "manifest.json", candidates) except Exception as exc: parser.exit(1, f"media index failed: {exc}\n") print(f"indexed {len(candidates)} candidates; manifest contains {len(manifest['assets'])}") return 0 if __name__ == "__main__": raise SystemExit(main())