feat: add safe media upgrade rollback

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, 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
+from rf4_research.media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, generate_media_derivatives, generate_quality_contact_sheets, inspect_image, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, review_asset, rollback_quality_upgrade, store_asset
 
 
 def test_extracts_and_classifies_unique_https_media() -> None:
@@ -81,6 +81,98 @@ def test_manifest_merges_and_binary_store_is_content_addressed(tmp_path: Path) -
     assert (width, height, mime) == (3, 2, "image/png")
 
 
+def test_derivatives_are_content_addressed_and_never_upscaled(tmp_path: Path) -> None:
+    manifest = tmp_path / "manifest.json"
+    item = extract_media_candidates('<img src="/fish/pike.png" alt="Щука">', source_page="https://example.test")[0]
+    saved = merge_manifest(manifest, [item])
+    image = io.BytesIO()
+    Image.new("RGBA", (400, 200), (0, 100, 200, 180)).save(image, format="PNG")
+    digest, relative, width, height, mime = store_asset(
+        tmp_path, image.getvalue(), content_type="image/png", source_url=item.asset_url,
+    )
+    saved["assets"][0].update({
+        "status": "approved", "entity_key": "fish:pike", "sha256": digest,
+        "local_path": relative, "width": width, "height": height, "content_type": mime,
+    })
+    manifest.write_text(json.dumps(saved), encoding="utf-8")
+
+    report = generate_media_derivatives(tmp_path)
+    assert report["issues"] == []
+    assert report["generated"] == 4
+    updated = json.loads(manifest.read_text(encoding="utf-8"))["assets"][0]
+    variants = updated["derivatives"]
+    assert {(item["role"], item["format"]) for item in variants} == {
+        ("card", "webp"), ("card", "avif"), ("detail", "webp"), ("detail", "avif"),
+    }
+    assert all(item["width"] <= 400 and item["height"] <= 200 for item in variants)
+    assert audit_media_catalog(tmp_path)["issues"] == []
+
+
+def test_quality_contact_sheet_contains_review_pairs(tmp_path: Path) -> None:
+    old = tmp_path / "old.png"
+    selected = tmp_path / "selected.png"
+    Image.new("RGBA", (48, 48), "red").save(old)
+    Image.new("RGBA", (256, 256), "blue").save(selected)
+    manifest = {
+        "assets": [
+            {"asset_url": "https://old.test/fish.png", "status": "superseded", "label": "Щука", "local_path": "old.png", "width": 48, "height": 48},
+            {"asset_url": "https://new.test/fish.webp", "status": "approved", "label": "Щука", "local_path": "selected.png", "width": 256, "height": 256, "supersedes": "https://old.test/fish.png"},
+        ],
+    }
+    (tmp_path / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+    report = generate_quality_contact_sheets(tmp_path, tmp_path / "sheets")
+    assert report["pairs"] == 1
+    assert report["issues"] == []
+    assert (tmp_path / "sheets" / "quality-upgrades-001.png").is_file()
+    assert json.loads((tmp_path / "sheets" / "quality-upgrades-001.json").read_text()) [0]["selected_url"] == "https://new.test/fish.webp"
+
+
+def test_waterbody_media_role_is_reviewed_and_never_inferred(tmp_path: Path) -> None:
+    manifest = tmp_path / "manifest.json"
+    candidate = extract_media_candidates(
+        '<img src="/maps/kuori.webp" alt="Куори">',
+        source_page="https://rf4db.com/ru/maps/level_001_kuori",
+    )[0]
+    merge_manifest(manifest, [candidate])
+    with pytest.raises(ValueError, match="only a stored asset"):
+        review_asset(
+            manifest, asset_url=candidate.asset_url, decision="approved",
+            entity_type="waterbody", entity_key="kuori", media_role="waterbody_map",
+        )
+
+    data = io.BytesIO()
+    Image.new("RGB", (256, 256), "blue").save(data, format="WEBP")
+    digest, local_path, width, height, mime = store_asset(
+        tmp_path, data.getvalue(), content_type="image/webp", source_url=candidate.asset_url,
+    )
+    saved = json.loads(manifest.read_text(encoding="utf-8"))
+    saved["assets"][0].update({
+        "status": "stored", "sha256": digest, "local_path": local_path,
+        "width": width, "height": height, "content_type": mime,
+    })
+    manifest.write_text(json.dumps(saved), encoding="utf-8")
+    reviewed = review_asset(
+        manifest, asset_url=candidate.asset_url, decision="approved",
+        entity_type="waterbody", entity_key="kuori", media_role="waterbody_map",
+        note="manual contact-sheet review",
+    )
+    assert reviewed["media_role"] == "waterbody_map"
+
+
+def test_waterbody_media_role_rejects_unknown_role(tmp_path: Path) -> None:
+    manifest = tmp_path / "manifest.json"
+    candidate = extract_media_candidates(
+        '<img src="/maps/kuori.webp" alt="Куори">',
+        source_page="https://rf4db.com/ru/maps/level_001_kuori",
+    )[0]
+    merge_manifest(manifest, [candidate])
+    with pytest.raises(ValueError, match="known waterbody role"):
+        review_asset(
+            manifest, asset_url=candidate.asset_url, decision="rejected",
+            entity_type="waterbody", entity_key="kuori", media_role="cover",
+        )
+
+
 def test_image_inspection_rejects_invalid_body_and_mime_mismatch(tmp_path: Path) -> None:
     with pytest.raises(ValueError, match="valid raster"):
         inspect_image(b"not an image")
@@ -248,3 +340,20 @@ def test_publish_quality_upgrades_switches_mapping_and_retains_fallback(tmp_path
     assert assets[1]["status"] == "approved"
     assert assets[1]["entity_key"] == "fish:pike"
     assert assets[1]["supersedes"] == assets[0]["asset_url"]
+
+
+def test_rollback_quality_upgrade_restores_fallback_atomically(tmp_path: Path) -> None:
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps({"assets": [
+        {"entity_type": "fish", "entity_key": "fish:pike", "status": "superseded", "asset_url": "https://small/pike.png", "replaced_by": "https://large/pike.webp"},
+        {"entity_type": "fish", "entity_key": "fish:pike", "status": "approved", "asset_url": "https://large/pike.webp", "supersedes": "https://small/pike.png"},
+    ]}), encoding="utf-8")
+
+    report = rollback_quality_upgrade(path, asset_url="https://large/pike.webp", note="owner rolled back after visual review")
+    assets = json.loads(path.read_text(encoding="utf-8"))["assets"]
+
+    assert report == {"rolled_back": "https://large/pike.webp", "restored": "https://small/pike.png"}
+    assert assets[0]["status"] == "approved"
+    assert "replaced_by" not in assets[0]
+    assert assets[1]["status"] == "upgrade_stored"
+    assert "supersedes" not in assets[1]