From e012b84969b701a47301a51bdf1f44aceb774a0a Mon Sep 17 00:00:00 2001 From: IK Date: Wed, 16 Sep 2026 20:16:54 +0700 Subject: [PATCH] feat: add safe media upgrade rollback --- rf4_research/media_assets.py | 187 ++++++++++++++++++++++++++++++++++- rf4_research/media_cli.py | 18 +++- tests/test_media_assets.py | 111 ++++++++++++++++++++- 3 files changed, 312 insertions(+), 4 deletions(-) diff --git a/rf4_research/media_assets.py b/rf4_research/media_assets.py index 2d7adc3..3f18444 100644 --- a/rf4_research/media_assets.py +++ b/rf4_research/media_assets.py @@ -11,9 +11,13 @@ from pathlib import Path from urllib.parse import urljoin, urlsplit from bs4 import BeautifulSoup, Tag -from PIL import Image, UnidentifiedImageError +from PIL import Image, ImageDraw, ImageFont, UnidentifiedImageError MAX_IMAGE_PIXELS = 40_000_000 +DERIVATIVE_TARGETS = {"card": 256, "detail": 1024} +WATERBODY_MEDIA_ROLES = frozenset({ + "waterbody_cover", "waterbody_map", "waterbody_depth_map", "waterbody_screenshot", +}) @dataclass(frozen=True, slots=True) @@ -229,6 +233,29 @@ def publish_quality_upgrades(path: Path, *, note: str, minimum_dimension: int = return {"published": len(candidates), "retained_fallbacks": len(candidates)} +def rollback_quality_upgrade(path: Path, *, asset_url: str, note: str) -> dict: + """Restore the superseded fallback for one explicitly selected upgrade.""" + manifest = json.loads(path.read_text(encoding="utf-8")) + assets = manifest.get("assets", []) + candidate = next((item for item in assets if item.get("asset_url") == asset_url), None) + if not candidate or candidate.get("status") != "approved" or not candidate.get("supersedes"): + raise ValueError("rollback requires an approved upgrade with a supersedes link") + fallback = next((item for item in assets if item.get("asset_url") == candidate["supersedes"]), None) + if not fallback or fallback.get("status") != "superseded" or fallback.get("replaced_by") != asset_url: + raise ValueError("rollback requires the matching superseded fallback") + + reviewed_at = datetime.now(timezone.utc).isoformat() + candidate.update({"status": "upgrade_stored", "reviewed_at": reviewed_at, "review_note": note}) + candidate.pop("supersedes", None) + fallback.update({"status": "approved", "reviewed_at": reviewed_at, "review_note": note}) + fallback.pop("replaced_by", None) + 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 {"rolled_back": asset_url, "restored": fallback["asset_url"]} + + def reclassify_manifest(path: Path) -> dict: manifest = json.loads(path.read_text(encoding="utf-8")) for item in manifest.get("assets", []): @@ -241,6 +268,7 @@ def reclassify_manifest(path: Path) -> dict: def review_asset( path: Path, *, asset_url: str, decision: str, entity_type: str | None = None, entity_key: str | None = None, note: str | None = None, + media_role: str | None = None, ) -> dict: if decision not in {"approved", "rejected"}: raise ValueError("decision must be approved or rejected") @@ -248,12 +276,16 @@ def review_asset( item = next((asset for asset in manifest.get("assets", []) if asset["asset_url"] == asset_url), None) if item is None: raise ValueError("asset URL is not present in manifest") + if media_role is not None and (entity_type != "waterbody" or media_role not in WATERBODY_MEDIA_ROLES): + raise ValueError("media role is only valid for a waterbody and must be a known waterbody role") if decision == "approved": if item.get("status") != "stored": raise ValueError("only a stored asset can be approved") if entity_type not in {"fish", "waterbody", "tackle", "reference"} or not entity_key: raise ValueError("approved asset requires entity type and canonical key") item.update({"entity_type": entity_type, "entity_key": entity_key}) + if media_role is not None: + item["media_role"] = media_role item.update({ "status": decision, "reviewed_at": datetime.now(timezone.utc).isoformat(), @@ -290,6 +322,27 @@ def audit_media_catalog(root: Path) -> dict: issues.append(f"{item['asset_url']}: image metadata mismatch") except ValueError as exc: issues.append(f"{item['asset_url']}: {exc}") + for variant in item.get("derivatives", []): + variant_path = variant.get("local_path") + if not isinstance(variant_path, str): + issues.append(f"{item['asset_url']}: derivative has no local_path") + continue + referenced.add(variant_path) + target = root / variant_path + if not target.is_file(): + issues.append(f"{item['asset_url']}: derivative file is missing") + continue + body = target.read_bytes() + if hashlib.sha256(body).hexdigest() != variant.get("sha256"): + issues.append(f"{item['asset_url']}: derivative SHA-256 mismatch") + try: + width, height, mime = inspect_image(body) + if (width, height, mime) != (variant.get("width"), variant.get("height"), variant.get("content_type")): + issues.append(f"{item['asset_url']}: derivative image metadata mismatch") + if len(body) != variant.get("bytes"): + issues.append(f"{item['asset_url']}: derivative byte count mismatch") + except ValueError as exc: + issues.append(f"{item['asset_url']}: derivative {exc}") if status == "approved" and (not item.get("entity_key") or item.get("entity_type") not in {"fish", "waterbody", "tackle", "reference"}): issues.append(f"{item['asset_url']}: approved asset has no valid canonical mapping") files_root = root / "files" @@ -416,6 +469,70 @@ def compare_quality_upgrades(root: Path, *, minimum_dimension: int = 256) -> dic } +def generate_quality_contact_sheets( + root: Path, output_dir: Path, *, batch_size: int = 20, +) -> dict: + """Create offline old/selected contact sheets for manually reviewed upgrades.""" + if batch_size < 1: + raise ValueError("batch_size must be positive") + 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} + pairs = [] + issues: list[str] = [] + for selected in assets: + if selected.get("status") != "approved" or not selected.get("supersedes"): + continue + old = by_url.get(selected["supersedes"]) + if not old: + issues.append(f"{selected.get('asset_url')}: superseded asset is missing") + continue + old_path, selected_path = root / str(old.get("local_path") or ""), root / str(selected.get("local_path") or "") + if not old_path.is_file() or not selected_path.is_file(): + issues.append(f"{selected.get('asset_url')}: contact-sheet image is missing") + continue + pairs.append((old, selected)) + + output_dir.mkdir(parents=True, exist_ok=True) + sheets = [] + cell_width, cell_height, image_size = 320, 220, 180 + font_paths = ( + Path("/usr/share/fonts/Fonts/DejaVuSans.ttf"), + Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"), + Path("/usr/share/fonts/dejavu/DejaVuSans.ttf"), + ) + font_path = next((path for path in font_paths if path.is_file()), None) + font = ImageFont.truetype(font_path, 12) if font_path else ImageFont.load_default() + for sheet_index in range(0, len(pairs), batch_size): + chunk = pairs[sheet_index:sheet_index + batch_size] + canvas = Image.new("RGB", (cell_width * 2, cell_height * len(chunk)), "white") + draw = ImageDraw.Draw(canvas) + metadata = [] + for row, (old, selected) in enumerate(chunk): + y = row * cell_height + for column, item in enumerate((old, selected)): + path = root / str(item["local_path"]) + with Image.open(path) as image: + preview = image.convert("RGBA") + preview.thumbnail((image_size, image_size), Image.Resampling.LANCZOS) + x = column * cell_width + (image_size - preview.width) // 2 + canvas.paste(preview, (x, y + 4), preview if preview.mode == "RGBA" else None) + title = "СТАРЫЙ" if column == 0 else "ВЫБРАННЫЙ" + label = str(item.get("label") or "Без подписи")[:34] + source = urlsplit(str(item.get("source_page") or "")).hostname or "unknown" + draw.text((column * cell_width + 190, y + 8), f"{title}: {label}", fill="black", font=font) + draw.text((column * cell_width + 190, y + 30), f"{item.get('width')}×{item.get('height')} · {source}", fill="black", font=font) + draw.text((column * cell_width + 190, y + 52), str(item.get("asset_url") or "")[:42], fill="gray", font=font) + draw.line((0, y + cell_height - 1, cell_width * 2, y + cell_height - 1), fill="#cccccc") + metadata.append({"label": selected.get("label"), "old_url": old.get("asset_url"), "selected_url": selected.get("asset_url")}) + image_path = output_dir / f"quality-upgrades-{sheet_index // batch_size + 1:03d}.png" + json_path = image_path.with_suffix(".json") + canvas.save(image_path, format="PNG", optimize=True) + json_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + sheets.append(str(image_path)) + return {"pairs": len(pairs), "sheets": sheets, "issues": issues} + + def inspect_image(body: bytes) -> tuple[int, int, str]: try: with Image.open(io.BytesIO(body)) as image: @@ -427,7 +544,7 @@ def inspect_image(body: bytes) -> tuple[int, int, str]: if width < 1 or height < 1 or width * height > MAX_IMAGE_PIXELS: raise ValueError("asset dimensions are outside safe limits") mime = Image.MIME.get(image_format or "") - if mime not in {"image/jpeg", "image/png", "image/webp", "image/gif"}: + if mime not in {"image/jpeg", "image/png", "image/webp", "image/gif", "image/avif"}: raise ValueError(f"unsupported image format: {image_format}") return width, height, mime @@ -445,3 +562,69 @@ def store_asset(root: Path, body: bytes, *, content_type: str, source_url: str) if not target.exists(): target.write_bytes(body) return digest, str(target.relative_to(root)), width, height, detected_mime + + +def generate_media_derivatives( + root: Path, *, targets: dict[str, int] | None = None, +) -> dict: + """Generate deterministic WebP/AVIF derivatives for approved local assets. + + Derivatives are content-addressed like originals and never upscale a source + image. The approved original remains the fallback and is not rewritten. + """ + manifest_path = root / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + selected_targets = targets or DERIVATIVE_TARGETS + generated = skipped = 0 + issues: list[str] = [] + + for item in manifest.get("assets", []): + if item.get("status") != "approved" or not item.get("local_path"): + continue + source_path = root / str(item["local_path"]) + if not source_path.is_file(): + issues.append(f"{item.get('asset_url')}: derivative source is missing") + continue + try: + with Image.open(source_path) as source: + source.load() + source_width, source_height = source.size + source_format = source.format + if source_format not in {"JPEG", "PNG", "WEBP", "GIF"}: + raise ValueError(f"unsupported source format: {source_format}") + variants: list[dict] = [] + for role, target_width in selected_targets.items(): + width = min(source_width, int(target_width)) + height = max(1, round(source_height * width / source_width)) + resized = source if width == source_width else source.resize((width, height), Image.Resampling.LANCZOS) + for image_format, mime, extension, save_options in ( + ("WEBP", "image/webp", ".webp", {"quality": 85, "method": 6}), + ("AVIF", "image/avif", ".avif", {"quality": 80}), + ): + output = io.BytesIO() + if resized.mode not in {"RGB", "RGBA", "L", "LA"}: + converted = resized.convert("RGBA" if "A" in resized.mode else "RGB") + else: + converted = resized + converted.save(output, format=image_format, **save_options) + body = output.getvalue() + digest = hashlib.sha256(body).hexdigest() + target = root / "files" / digest[:2] / f"{digest}{extension}" + target.parent.mkdir(parents=True, exist_ok=True) + if not target.exists(): + target.write_bytes(body) + generated += 1 + else: + skipped += 1 + variants.append({ + "role": role, "format": image_format.lower(), "sha256": digest, + "local_path": str(target.relative_to(root)), "content_type": mime, + "bytes": len(body), "width": width, "height": height, + }) + item["derivatives"] = variants + except (OSError, ValueError) as exc: + issues.append(f"{item.get('asset_url')}: {exc}") + + manifest["updated_at"] = datetime.now(timezone.utc).isoformat() + manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + return {"approved_assets": sum(item.get("status") == "approved" for item in manifest.get("assets", [])), "generated": generated, "already_present": skipped, "issues": issues} diff --git a/rf4_research/media_cli.py b/rf4_research/media_cli.py index 6ed9c4e..23a84e8 100644 --- a/rf4_research/media_cli.py +++ b/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, 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 +from .media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, generate_media_derivatives, generate_quality_contact_sheets, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, reclassify_manifest, review_asset, rollback_quality_upgrade, store_asset DEFAULT_ROOT = Path("data/media") @@ -187,6 +187,7 @@ def main(argv: list[str] | None = None) -> int: 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("--rollback-upgrade", action="store_true", help="Restore one superseded fallback after an explicit review decision") 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")) @@ -197,6 +198,8 @@ def main(argv: list[str] | None = None) -> int: 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("--generate-derivatives", action="store_true", help="Generate deterministic WebP/AVIF variants for approved local assets") + parser.add_argument("--contact-sheet-dir", type=Path, help="Write offline quality-upgrade contact sheets to this directory") 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) @@ -218,6 +221,14 @@ def main(argv: list[str] | None = None) -> int: if args.queue_plan: print(json.dumps(media_queue_plan(args.root, args.state_file), ensure_ascii=False, indent=2)) return 0 + if args.generate_derivatives: + report = generate_media_derivatives(args.root) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 1 if report["issues"] else 0 + if args.contact_sheet_dir: + report = generate_quality_contact_sheets(args.root, args.contact_sheet_dir) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 1 if report["issues"] else 0 if args.review_url: if not args.decision: parser.error("--decision is required with --review-url") @@ -244,6 +255,11 @@ def main(argv: list[str] | None = None) -> int: 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.rollback_upgrade: + if not args.asset_url or not args.note: + parser.error("--asset-url and --note are required with --rollback-upgrade") + print(json.dumps(rollback_quality_upgrade(args.root / "manifest.json", asset_url=args.asset_url, 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 diff --git a/tests/test_media_assets.py b/tests/test_media_assets.py index 2bac506..3fb7103 100644 --- a/tests/test_media_assets.py +++ b/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('Щука', 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( + 'Куори', + 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( + 'Куори', + 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]