360 lines
19 KiB
Python
360 lines
19 KiB
Python
from pathlib import Path
|
|
import json
|
|
|
|
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, 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:
|
|
html = '''<article><a href="/fish/pike"><img src="/media/fish/pike.png" alt="Щука"></a></article>
|
|
<figure><img data-src="https://cdn.example/maps/kuori.webp" alt="Карта Куори"></figure>
|
|
<img src="http://unsafe.example/bait.png"><img src="/media/fish/pike.png">'''
|
|
items = extract_media_candidates(html, source_page="https://example.test/guide")
|
|
assert [(item.entity_type, item.asset_url) for item in items] == [
|
|
("fish", "https://example.test/media/fish/pike.png"),
|
|
("waterbody", "https://cdn.example/maps/kuori.webp"),
|
|
]
|
|
assert items[0].external_id == "pike"
|
|
|
|
|
|
def test_does_not_mistake_seafishing_screenshot_for_fish_entity() -> None:
|
|
items = extract_media_candidates('<img src="/res/userguide/seafishing-1.jpeg">', source_page="https://rf4game.de/userguide/")
|
|
assert items[0].entity_type == "reference"
|
|
|
|
|
|
def test_gallery_page_provenance_classifies_generic_asset_urls() -> None:
|
|
fish = extract_media_candidates('<img src="/uploads/gallery/123.jpg" alt="Hecht">', source_page="https://rf4game.de/media/fische/")
|
|
water = extract_media_candidates('<img src="/uploads/gallery/456.jpg" alt="Kuori">', source_page="https://rf4game.de/media/levels/")
|
|
assert fish[0].entity_type == "fish"
|
|
assert water[0].entity_type == "waterbody"
|
|
chrome = extract_media_candidates('<img src="/wp-content/themes/rf4/img/banner_de.png">', source_page="https://rf4game.de/media/fische/")
|
|
assert chrome[0].entity_type == "reference"
|
|
|
|
|
|
def test_rf4map_gateway_fish_filename_is_classified_as_fish() -> None:
|
|
items = extract_media_candidates(
|
|
'<img src="https://gw.rf4map.ru/public/images/fish_123.webp" alt="Ерш">',
|
|
source_page="https://rf4map.ru/fishes",
|
|
)
|
|
assert items[0].entity_type == "fish"
|
|
|
|
|
|
def test_rf4db_asset_filename_is_stable_external_id() -> None:
|
|
item = extract_media_candidates(
|
|
'<img src="https://oss.rf4db.com/game/fish/a.sleeper.webp" alt="Ротан">',
|
|
source_page="https://download.rf4db.com/ru/fishes",
|
|
)[0]
|
|
assert (item.entity_type, item.external_id) == ("fish", "a.sleeper")
|
|
|
|
|
|
def test_tackle_path_wins_over_fish_word_in_product_name() -> None:
|
|
items = extract_media_candidates(
|
|
'<img src="https://cdn.example/bait/crab.png" alt="Краб и рыба 14">',
|
|
source_page="https://example.test/fishes",
|
|
)
|
|
assert items[0].entity_type == "tackle"
|
|
|
|
|
|
def test_generic_userguide_map_is_reference_not_waterbody_entity() -> None:
|
|
items = extract_media_candidates(
|
|
'<img src="/res/userguide/map.jpeg" alt="Bild: Gewässerkarte">',
|
|
source_page="https://rf4game.de/userguide/",
|
|
)
|
|
assert items[0].entity_type == "reference"
|
|
|
|
|
|
def test_manifest_merges_and_binary_store_is_content_addressed(tmp_path: Path) -> None:
|
|
item = extract_media_candidates('<img src="/bait/worm.png" alt="Bait worm">', source_page="https://example.test")[0]
|
|
first = merge_manifest(tmp_path / "manifest.json", [item])
|
|
second = merge_manifest(tmp_path / "manifest.json", [item])
|
|
assert len(first["assets"]) == len(second["assets"]) == 1
|
|
assert second["assets"][0]["source_pages"] == ["https://example.test"]
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (3, 2), "green").save(image, format="PNG")
|
|
digest, relative, width, height, mime = store_asset(tmp_path, image.getvalue(), content_type="image/png", source_url=item.asset_url)
|
|
assert len(digest) == 64
|
|
assert (tmp_path / relative).read_bytes() == image.getvalue()
|
|
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")
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (1, 1)).save(image, format="PNG")
|
|
with pytest.raises(ValueError, match="MIME mismatch"):
|
|
store_asset(tmp_path, image.getvalue(), content_type="image/jpeg", source_url="https://example.test/a.jpg")
|
|
|
|
|
|
def test_review_requires_stored_asset_and_canonical_mapping(tmp_path: Path) -> None:
|
|
item = extract_media_candidates('<img src="/fish/pike.png" alt="Щука">', source_page="https://example.test")[0]
|
|
path = tmp_path / "manifest.json"
|
|
manifest = merge_manifest(path, [item])
|
|
with pytest.raises(ValueError, match="stored"):
|
|
review_asset(path, asset_url=item.asset_url, decision="approved", entity_type="fish", entity_key="pike")
|
|
manifest["assets"][0]["status"] = "stored"
|
|
path.write_text(__import__("json").dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="canonical key"):
|
|
review_asset(path, asset_url=item.asset_url, decision="approved", entity_type="fish")
|
|
reviewed = review_asset(path, asset_url=item.asset_url, decision="approved", entity_type="fish", entity_key="pike", note="matched by name")
|
|
assert (reviewed["status"], reviewed["entity_key"]) == ("approved", "pike")
|
|
|
|
|
|
def test_catalog_audit_detects_tampering_and_orphans(tmp_path: Path) -> None:
|
|
item = extract_media_candidates('<img src="/fish/pike.png">', source_page="https://example.test")[0]
|
|
path = tmp_path / "manifest.json"
|
|
manifest = merge_manifest(path, [item])
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (2, 2)).save(image, format="PNG")
|
|
digest, relative, width, height, mime = store_asset(tmp_path, image.getvalue(), content_type="image/png", source_url=item.asset_url)
|
|
manifest["assets"][0].update({"status": "stored", "sha256": digest, "local_path": relative, "width": width, "height": height, "content_type": mime})
|
|
path.write_text(__import__("json").dumps(manifest), encoding="utf-8")
|
|
assert audit_media_catalog(tmp_path)["issues"] == []
|
|
(tmp_path / relative).write_bytes(b"changed")
|
|
(tmp_path / "files" / "orphan.png").write_bytes(b"orphan")
|
|
report = audit_media_catalog(tmp_path)
|
|
assert any("SHA-256 mismatch" in issue for issue in report["issues"])
|
|
assert "files/orphan.png" in report["orphaned_files"]
|
|
|
|
|
|
def test_media_coverage_keeps_unknown_tackle_total_honest(tmp_path: Path) -> None:
|
|
(tmp_path / "manifest.json").write_text(
|
|
'{"assets":[{"entity_type":"fish","status":"approved","entity_key":"ruffe"},'
|
|
'{"entity_type":"fish","status":"queued"},{"entity_type":"tackle","status":"queued"}]}',
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "catalog-baseline.json").write_text(
|
|
'{"verified_at":"2026-09-12","entities":{"fish":{"count":2},"tackle":{"count":null}}}',
|
|
encoding="utf-8",
|
|
)
|
|
report = media_coverage(tmp_path)["entities"]
|
|
assert report["fish"] == {
|
|
"expected": 2, "candidate_assets": 2, "unique_candidates": 0,
|
|
"unidentified_assets": 2, "approved": 1, "candidate_gap": 2, "approved_gap": 1,
|
|
}
|
|
assert report["tackle"]["candidate_gap"] is None
|
|
|
|
|
|
def test_media_coverage_deduplicates_candidate_labels(tmp_path: Path) -> None:
|
|
(tmp_path / "manifest.json").write_text(
|
|
'{"assets":[{"entity_type":"fish","status":"queued","label":" Ерш "},'
|
|
'{"entity_type":"fish","status":"queued","label":"ерш"}]}', encoding="utf-8",
|
|
)
|
|
(tmp_path / "catalog-baseline.json").write_text(
|
|
'{"verified_at":"2026-09-12","entities":{"fish":{"count":2}}}', encoding="utf-8",
|
|
)
|
|
fish = media_coverage(tmp_path)["entities"]["fish"]
|
|
assert (fish["candidate_assets"], fish["unique_candidates"], fish["candidate_gap"]) == (2, 1, 1)
|
|
|
|
|
|
def test_reconcile_queue_preserves_source_but_skips_normalized_duplicate(tmp_path: Path) -> None:
|
|
path = tmp_path / "manifest.json"
|
|
path.write_text(json.dumps({"assets": [
|
|
{"entity_type": "fish", "label": "Ерш", "status": "stored", "asset_url": "https://a/fish.png"},
|
|
{"entity_type": "fish", "label": " ЁРШ ", "status": "queued", "asset_url": "https://b/fish.webp"},
|
|
{"entity_type": "fish", "label": "Щука", "status": "queued", "asset_url": "https://b/pike.webp"},
|
|
]}), encoding="utf-8")
|
|
|
|
report = reconcile_queued_duplicates(path)
|
|
assets = json.loads(path.read_text(encoding="utf-8"))["assets"]
|
|
|
|
assert report == {"duplicates": 1, "queued": 1}
|
|
assert assets[1]["status"] == "duplicate"
|
|
assert assets[1]["duplicate_of"] == "https://a/fish.png"
|
|
assert assets[2]["status"] == "queued"
|
|
|
|
|
|
def test_bulk_approval_only_publishes_stored_assets(tmp_path: Path) -> None:
|
|
path = tmp_path / "manifest.json"
|
|
path.write_text(json.dumps({"assets": [
|
|
{"entity_type": "fish", "external_id": "pike", "status": "stored", "asset_url": "https://a/pike.webp", "sha256": "a" * 64},
|
|
{"entity_type": "fish", "status": "queued", "asset_url": "https://a/perch.webp"},
|
|
]}), encoding="utf-8")
|
|
assert approve_stored_assets(path, note="owner approved") == {"approved": 1}
|
|
assets = json.loads(path.read_text(encoding="utf-8"))["assets"]
|
|
assert assets[0]["entity_key"] == "fish:pike"
|
|
assert assets[0]["review_note"] == "owner approved"
|
|
assert assets[1]["status"] == "queued"
|
|
|
|
|
|
def test_quality_report_finds_low_resolution_asset_and_known_alternative(tmp_path: Path) -> None:
|
|
(tmp_path / "manifest.json").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 = media_quality_report(tmp_path)
|
|
|
|
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"]
|
|
|
|
|
|
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]
|