Publish first RF4 fish quality upgrades

rf4_research/media_assets.py

@@ -136,6 +136,35 @@ def reconcile_queued_duplicates(path: Path) -> dict:
     return {"duplicates": duplicates, "queued": sum(item.get("status") == "queued" for item in manifest.get("assets", []))}
 
 
+def queue_quality_upgrades(path: Path, *, minimum_dimension: int = 256) -> dict:
+    """Queue alternatives to low-resolution published fish without unpublishing them."""
+    manifest = json.loads(path.read_text(encoding="utf-8"))
+    assets = manifest.get("assets", [])
+    low_urls = {
+        item.get("asset_url")
+        for item in assets
+        if item.get("status") == "approved"
+        and item.get("entity_type") == "fish"
+        and min(int(item.get("width") or 0), int(item.get("height") or 0)) < minimum_dimension
+    }
+    queued = 0
+    for item in assets:
+        if (
+            item.get("status") == "duplicate"
+            and item.get("entity_type") == "fish"
+            and item.get("duplicate_of") in low_urls
+        ):
+            item["status"] = "upgrade_queued"
+            queued += 1
+    manifest["updated_at"] = datetime.now(timezone.utc).isoformat()
+    path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return {
+        "minimum_dimension": minimum_dimension,
+        "low_resolution_published": len(low_urls),
+        "upgrade_queued": queued,
+    }
+
+
 def approve_stored_assets(path: Path, *, note: str) -> dict:
     """Publish every stored asset after an explicit owner-level approval."""
     manifest = json.loads(path.read_text(encoding="utf-8"))
@@ -157,6 +186,49 @@ def approve_stored_assets(path: Path, *, note: str) -> dict:
     return {"approved": approved}
 
 
+def publish_quality_upgrades(path: Path, *, note: str, minimum_dimension: int = 256) -> dict:
+    """Atomically promote every reviewed quality candidate and retain its fallback."""
+    manifest = json.loads(path.read_text(encoding="utf-8"))
+    assets = manifest.get("assets", [])
+    by_url = {item.get("asset_url"): item for item in assets}
+    candidates = [item for item in assets if item.get("status") == "upgrade_stored"]
+
+    # Validate the complete set before changing any public mapping.
+    for candidate in candidates:
+        fallback = by_url.get(candidate.get("duplicate_of"))
+        if not fallback or fallback.get("status") != "approved":
+            raise ValueError(f"{candidate.get('asset_url')}: approved fallback is missing")
+        if not fallback.get("entity_key"):
+            raise ValueError(f"{candidate.get('asset_url')}: fallback has no entity_key")
+        if min(int(candidate.get("width") or 0), int(candidate.get("height") or 0)) < minimum_dimension:
+            raise ValueError(f"{candidate.get('asset_url')}: candidate is below {minimum_dimension}px")
+        if not candidate.get("sha256") or not candidate.get("local_path"):
+            raise ValueError(f"{candidate.get('asset_url')}: stored candidate metadata is incomplete")
+
+    reviewed_at = datetime.now(timezone.utc).isoformat()
+    for candidate in candidates:
+        fallback = by_url[candidate["duplicate_of"]]
+        candidate.update({
+            "status": "approved",
+            "entity_key": fallback["entity_key"],
+            "supersedes": fallback["asset_url"],
+            "reviewed_at": reviewed_at,
+            "review_note": note,
+        })
+        fallback.update({
+            "status": "superseded",
+            "replaced_by": candidate["asset_url"],
+            "reviewed_at": reviewed_at,
+            "review_note": note,
+        })
+
+    manifest["updated_at"] = reviewed_at
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    temporary.replace(path)
+    return {"published": len(candidates), "retained_fallbacks": len(candidates)}
+
+
 def reclassify_manifest(path: Path) -> dict:
     manifest = json.loads(path.read_text(encoding="utf-8"))
     for item in manifest.get("assets", []):
@@ -200,7 +272,7 @@ def audit_media_catalog(root: Path) -> dict:
         status = str(item.get("status", "unknown"))
         statuses[status] = statuses.get(status, 0) + 1
         local_path = item.get("local_path")
-        if status in {"stored", "approved"}:
+        if status in {"stored", "upgrade_stored", "approved", "superseded"}:
             if not isinstance(local_path, str):
                 issues.append(f"{item['asset_url']}: {status} asset has no local_path")
                 continue
@@ -262,7 +334,7 @@ def media_quality_report(root: Path, *, minimum_dimension: int = 256, display_di
     alternatives = [
         item for item in assets
         if item.get("entity_type") == "fish"
-        and item.get("status") in {"duplicate", "queued"}
+        and item.get("status") in {"duplicate", "queued", "upgrade_queued", "upgrade_stored"}
         and (item.get("duplicate_of") in low_urls or item.get("status") == "queued")
     ]
 
@@ -290,6 +362,60 @@ def media_quality_report(root: Path, *, minimum_dimension: int = 256, display_di
     }
 
 
+def compare_quality_upgrades(root: Path, *, minimum_dimension: int = 256) -> dict:
+    """Compare stored upgrade candidates with their published fallbacks offline."""
+    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
+    assets = manifest.get("assets", [])
+    by_url = {item.get("asset_url"): item for item in assets}
+    comparisons: list[dict] = []
+    issues: list[str] = []
+    for candidate in assets:
+        if candidate.get("status") != "upgrade_stored":
+            continue
+        fallback = by_url.get(candidate.get("duplicate_of"))
+        if not fallback or fallback.get("status") != "approved":
+            issues.append(f"{candidate.get('asset_url')}: published fallback is missing")
+            continue
+        candidate_path = root / str(candidate.get("local_path") or "")
+        fallback_path = root / str(fallback.get("local_path") or "")
+        if not candidate_path.is_file() or not fallback_path.is_file():
+            issues.append(f"{candidate.get('asset_url')}: comparison file is missing")
+            continue
+        with Image.open(candidate_path) as image:
+            candidate_alpha = image.mode in {"LA", "RGBA"} or "transparency" in image.info
+        with Image.open(fallback_path) as image:
+            fallback_alpha = image.mode in {"LA", "RGBA"} or "transparency" in image.info
+        candidate_width, candidate_height = int(candidate["width"]), int(candidate["height"])
+        fallback_width, fallback_height = int(fallback["width"]), int(fallback["height"])
+        candidate_ratio = candidate_width / candidate_height
+        fallback_ratio = fallback_width / fallback_height
+        comparisons.append({
+            "label": candidate.get("label"),
+            "candidate_url": candidate.get("asset_url"),
+            "fallback_url": fallback.get("asset_url"),
+            "candidate": {
+                "width": candidate_width, "height": candidate_height,
+                "bytes": candidate.get("bytes"), "content_type": candidate.get("content_type"),
+                "alpha": candidate_alpha, "aspect_ratio": round(candidate_ratio, 4),
+            },
+            "fallback": {
+                "width": fallback_width, "height": fallback_height,
+                "bytes": fallback.get("bytes"), "content_type": fallback.get("content_type"),
+                "alpha": fallback_alpha, "aspect_ratio": round(fallback_ratio, 4),
+            },
+            "meets_minimum": min(candidate_width, candidate_height) >= minimum_dimension,
+            "aspect_ratio_matches": abs(candidate_ratio - fallback_ratio) < 0.01,
+        })
+    return {
+        "minimum_dimension": minimum_dimension,
+        "compared": len(comparisons),
+        "meets_minimum": sum(item["meets_minimum"] for item in comparisons),
+        "aspect_ratio_matches": sum(item["aspect_ratio_matches"] for item in comparisons),
+        "issues": issues,
+        "comparisons": comparisons,
+    }
+
+
 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, media_quality_report, merge_manifest, reconcile_queued_duplicates, reclassify_manifest, review_asset, store_asset
+from .media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, reclassify_manifest, review_asset, store_asset
 
 
 DEFAULT_ROOT = Path("data/media")
@@ -33,7 +33,7 @@ def media_queue_plan(root: Path, state_file: Path, *, now: float | None = None)
     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":
+        if item.get("status") in {"queued", "upgrade_queued"}:
             grouped.setdefault(fetch_site_key(item["asset_url"]), []).append(item)
 
     domains: dict[str, dict] = {}
@@ -79,6 +79,7 @@ def _save_manifest(path: Path, manifest: dict) -> None:
 
 def _attempt_asset(root: Path, manifest_path: Path, manifest: dict, queued: dict) -> tuple[str, Exception | None]:
     url = queued["asset_url"]
+    is_upgrade = queued.get("status") == "upgrade_queued"
     attempted_at = datetime.now(timezone.utc).isoformat()
     try:
         request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "image/*"})
@@ -93,11 +94,12 @@ def _attempt_asset(root: Path, manifest_path: Path, manifest: dict, queued: dict
         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, 429} else "invalid" if isinstance(exc, ValueError) else "queued"
+        status = "missing" if code in {404, 410} else "blocked" if code in {401, 403, 429} else "invalid" if isinstance(exc, ValueError) else "upgrade_queued" if is_upgrade else "queued"
         queued.update({"status": status, "last_attempt_at": attempted_at, "last_error": str(exc)[:500]})
         _save_manifest(manifest_path, manifest)
         return f"{status} {url}: {str(exc)[:200]}", exc
-    queued.update({"status": "stored", "sha256": digest, "local_path": relative, "content_type": detected_mime, "bytes": len(body), "width": width, "height": height, "fetched_at": attempted_at})
+    stored_status = "upgrade_stored" if is_upgrade else "stored"
+    queued.update({"status": stored_status, "sha256": digest, "local_path": relative, "content_type": detected_mime, "bytes": len(body), "width": width, "height": height, "fetched_at": attempted_at})
     _save_manifest(manifest_path, manifest)
     return f"stored {url} as {relative}", None
 
@@ -105,7 +107,7 @@ def _attempt_asset(root: Path, manifest_path: Path, manifest: dict, queued: dict
 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)
+    queued = next((item for item in manifest["assets"] if item.get("status") in {"queued", "upgrade_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"]
@@ -125,7 +127,7 @@ def _download_batch(root: Path, state_file: Path, *, limit: int = MAX_BATCH_ASSE
     manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
     grouped: dict[str, list[dict]] = {}
     for item in manifest.get("assets", []):
-        if item.get("status") != "queued":
+        if item.get("status") not in {"queued", "upgrade_queued"}:
             continue
         _validate_url_before_io(item["asset_url"])
         grouped.setdefault(fetch_site_key(item["asset_url"]), []).append(item)
@@ -182,7 +184,9 @@ def main(argv: list[str] | None = None) -> int:
     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("--dedupe-queue", action="store_true", help="Mark queued label matches to stored assets as duplicates without network access")
+    parser.add_argument("--queue-quality-upgrades", action="store_true", help="Queue alternatives to published fish below the minimum resolution")
     parser.add_argument("--approve-stored", action="store_true", help="Publish all stored assets after explicit owner approval")
+    parser.add_argument("--publish-upgrades", action="store_true", help="Atomically publish all stored quality upgrades and retain fallbacks")
     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"))
@@ -191,6 +195,7 @@ def main(argv: list[str] | None = None) -> int:
     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("--compare-quality-upgrades", action="store_true", help="Compare stored quality candidates with their published fallbacks")
     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"))
@@ -206,6 +211,10 @@ def main(argv: list[str] | None = None) -> int:
         if args.quality_report:
             print(json.dumps(media_quality_report(args.root), ensure_ascii=False, indent=2))
             return 0
+        if args.compare_quality_upgrades:
+            report = compare_quality_upgrades(args.root)
+            print(json.dumps(report, ensure_ascii=False, indent=2))
+            return 1 if report["issues"] else 0
         if args.queue_plan:
             print(json.dumps(media_queue_plan(args.root, args.state_file), ensure_ascii=False, indent=2))
             return 0
@@ -222,11 +231,19 @@ def main(argv: list[str] | None = None) -> int:
         if args.dedupe_queue:
             print(json.dumps(reconcile_queued_duplicates(args.root / "manifest.json"), ensure_ascii=False, indent=2))
             return 0
+        if args.queue_quality_upgrades:
+            print(json.dumps(queue_quality_upgrades(args.root / "manifest.json"), ensure_ascii=False, indent=2))
+            return 0
         if args.approve_stored:
             if not args.note:
                 parser.error("--note is required with --approve-stored")
             print(json.dumps(approve_stored_assets(args.root / "manifest.json", note=args.note), ensure_ascii=False, indent=2))
             return 0
+        if args.publish_upgrades:
+            if not args.note:
+                parser.error("--note is required with --publish-upgrades")
+            print(json.dumps(publish_quality_upgrades(args.root / "manifest.json", note=args.note), ensure_ascii=False, indent=2))
+            return 0
         if args.download_one:
             print(_download_one(args.root, args.state_file, args.asset_url))
             return 0

tests/test_media_assets.py

@@ -6,7 +6,7 @@ import io
 import pytest
 from PIL import Image
 
-from rf4_research.media_assets import approve_stored_assets, audit_media_catalog, extract_media_candidates, inspect_image, media_coverage, media_quality_report, merge_manifest, reconcile_queued_duplicates, review_asset, store_asset
+from rf4_research.media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, inspect_image, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, review_asset, store_asset
 
 
 def test_extracts_and_classifies_unique_https_media() -> None:
@@ -193,3 +193,58 @@ def test_quality_report_finds_low_resolution_asset_and_known_alternative(tmp_pat
     assert (report["published"], report["below_minimum"], report["upsampled_in_cards"]) == (2, 1, 1)
     assert report["known_alternative_urls"] == 1
     assert report["by_source"]["small.example"]["below_minimum"] == 1
+
+
+def test_quality_upgrade_queue_preserves_published_fallback(tmp_path: Path) -> None:
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps({"assets": [
+        {"entity_type": "fish", "label": "Щука", "status": "approved", "asset_url": "https://small.example/pike.png", "width": 48, "height": 48},
+        {"entity_type": "fish", "label": "Щука", "status": "duplicate", "asset_url": "https://large.example/pike.webp", "duplicate_of": "https://small.example/pike.png"},
+        {"entity_type": "fish", "label": "Окунь", "status": "approved", "asset_url": "https://large.example/perch.webp", "width": 1024, "height": 1024},
+    ]}), encoding="utf-8")
+
+    report = queue_quality_upgrades(path)
+    assets = json.loads(path.read_text(encoding="utf-8"))["assets"]
+
+    assert report == {"minimum_dimension": 256, "low_resolution_published": 1, "upgrade_queued": 1}
+    assert assets[0]["status"] == "approved"
+    assert assets[1]["status"] == "upgrade_queued"
+
+
+def test_compare_quality_upgrades_checks_candidate_and_fallback(tmp_path: Path) -> None:
+    small = Image.new("RGBA", (48, 48), (1, 2, 3, 0))
+    large = Image.new("RGBA", (512, 512), (1, 2, 3, 0))
+    small_body = io.BytesIO()
+    large_body = io.BytesIO()
+    small.save(small_body, format="PNG")
+    large.save(large_body, format="WEBP")
+    small_digest, small_path, *_ = store_asset(tmp_path, small_body.getvalue(), content_type="image/png", source_url="https://small/pike.png")
+    large_digest, large_path, *_ = store_asset(tmp_path, large_body.getvalue(), content_type="image/webp", source_url="https://large/pike.webp")
+    (tmp_path / "manifest.json").write_text(json.dumps({"assets": [
+        {"entity_type": "fish", "label": "Щука", "status": "approved", "asset_url": "https://small/pike.png", "local_path": small_path, "sha256": small_digest, "width": 48, "height": 48, "bytes": len(small_body.getvalue()), "content_type": "image/png"},
+        {"entity_type": "fish", "label": "Щука", "status": "upgrade_stored", "asset_url": "https://large/pike.webp", "duplicate_of": "https://small/pike.png", "local_path": large_path, "sha256": large_digest, "width": 512, "height": 512, "bytes": len(large_body.getvalue()), "content_type": "image/webp"},
+    ]}), encoding="utf-8")
+
+    report = compare_quality_upgrades(tmp_path)
+
+    assert (report["compared"], report["meets_minimum"], report["aspect_ratio_matches"]) == (1, 1, 1)
+    assert report["issues"] == []
+    assert report["comparisons"][0]["candidate"]["alpha"] is True
+
+
+def test_publish_quality_upgrades_switches_mapping_and_retains_fallback(tmp_path: Path) -> None:
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps({"assets": [
+        {"entity_type": "fish", "entity_key": "fish:pike", "status": "approved", "asset_url": "https://small/pike.png", "local_path": "files/small.png", "sha256": "a" * 64},
+        {"entity_type": "fish", "status": "upgrade_stored", "asset_url": "https://large/pike.webp", "duplicate_of": "https://small/pike.png", "local_path": "files/large.webp", "sha256": "b" * 64, "width": 512, "height": 512},
+    ]}), encoding="utf-8")
+
+    report = publish_quality_upgrades(path, note="owner approved quality upgrades")
+    assets = json.loads(path.read_text(encoding="utf-8"))["assets"]
+
+    assert report == {"published": 1, "retained_fallbacks": 1}
+    assert assets[0]["status"] == "superseded"
+    assert assets[0]["replaced_by"] == assets[1]["asset_url"]
+    assert assets[1]["status"] == "approved"
+    assert assets[1]["entity_key"] == "fish:pike"
+    assert assets[1]["supersedes"] == assets[0]["asset_url"]

tests/test_media_cli.py

@@ -82,3 +82,22 @@ def test_batch_stops_domain_after_three_consecutive_invalid_assets(tmp_path: Pat
     assert report["attempted_total"] == 3
     assert report["failed_total"] == 3
     assert "3 consecutive failures" in report["domains"]["rf4map.ru"]["stopped_reason"]
+
+
+def test_batch_downloads_quality_upgrade_without_unpublishing_fallback(tmp_path: Path, monkeypatch) -> None:
+    root = tmp_path / "media"
+    root.mkdir()
+    candidate = {"status": "upgrade_queued", "asset_url": "https://rf4db.com/pike.webp", "entity_type": "fish", "duplicate_of": "https://rf4map.ru/pike.png"}
+    (root / "manifest.json").write_text(json.dumps({"version": 1, "assets": [candidate]}), encoding="utf-8")
+
+    def store_upgrade(_root, path, manifest, item):
+        item["status"] = "upgrade_stored"
+        media_cli._save_manifest(path, manifest)
+        return "stored", None
+
+    monkeypatch.setattr(media_cli, "_attempt_asset", store_upgrade)
+    report = _download_batch(root, tmp_path / "state.json", limit=1)
+
+    assert report["stored_total"] == 1
+    assets = json.loads((root / "manifest.json").read_text(encoding="utf-8"))["assets"]
+    assert assets[0]["status"] == "upgrade_stored"