631 lines
32 KiB
Python
631 lines
32 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import io
|
||
import json
|
||
import mimetypes
|
||
import re
|
||
from dataclasses import asdict, dataclass
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from urllib.parse import urljoin, urlsplit
|
||
|
||
from bs4 import BeautifulSoup, Tag
|
||
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)
|
||
class MediaCandidate:
|
||
source_page: str
|
||
asset_url: str
|
||
entity_type: str
|
||
label: str | None
|
||
external_id: str | None
|
||
|
||
|
||
def _label(node: Tag) -> str | None:
|
||
value = node.get("alt") or node.get("title")
|
||
if isinstance(value, str) and value.strip():
|
||
return " ".join(value.split())[:300]
|
||
parent = node.find_parent(["article", "figure", "a", "section"])
|
||
if parent:
|
||
text = " ".join(parent.get_text(" ", strip=True).split())
|
||
return text[:300] or None
|
||
return None
|
||
|
||
|
||
def _kind(url: str, label: str | None, context: str) -> str:
|
||
path = urlsplit(url).path.casefold()
|
||
value = f"{label or ''} {context}".casefold()
|
||
if "/flags/" in path or "/themes/" in path or path.endswith(("/logo.png", "/logo.svg")):
|
||
return "reference"
|
||
# A structural tackle path is stronger evidence than words in a product
|
||
# name (for example, bait named "Краб и рыба").
|
||
if re.search(r"(?:^|[/_-])(?:bait|lure|rig|rod|reel|hook|tackle|köder|rute|rolle)(?:[/_.-]|$)", path):
|
||
return "tackle"
|
||
if re.search(r"/(?:fish|fishes|fische|species)/", path) or re.search(r"/(?:fish|species)[_-]", path) or "/media/fische/" in value or any(word in value for word in ("рыба ", "fish: ", "fisch: ")):
|
||
return "fish"
|
||
if re.search(r"/(?:maps?|levels?|lakes?|waterbodies)/", path) or "/media/levels/" in value:
|
||
return "waterbody"
|
||
return "reference"
|
||
|
||
|
||
def _external_id(node: Tag, url: str) -> str | None:
|
||
link = node.find_parent("a", href=True)
|
||
if link:
|
||
return str(link.get("href")).rstrip("/").rsplit("/", 1)[-1]
|
||
parts = urlsplit(url)
|
||
if parts.hostname == "oss.rf4db.com" and parts.path.startswith("/game/"):
|
||
return Path(parts.path).stem
|
||
return None
|
||
|
||
|
||
def extract_media_candidates(html: str, *, source_page: str) -> list[MediaCandidate]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
result: dict[str, MediaCandidate] = {}
|
||
for node in soup.select("img[src], img[data-src], source[srcset]"):
|
||
raw = node.get("src") or node.get("data-src") or node.get("srcset")
|
||
if not isinstance(raw, str):
|
||
continue
|
||
raw = raw.split(",", 1)[0].strip().split(" ", 1)[0]
|
||
url = urljoin(source_page, raw)
|
||
if urlsplit(url).scheme != "https":
|
||
continue
|
||
label = _label(node)
|
||
context = " ".join(node.parent.get_text(" ", strip=True).split())[:500] if node.parent else ""
|
||
external_id = _external_id(node, url)
|
||
result.setdefault(url, MediaCandidate(source_page, url, _kind(url, label, f"{source_page} {context}"), label, external_id))
|
||
for match in re.finditer(r"url\((['\"]?)(https://[^)'\"]+)\1\)", html, re.I):
|
||
url = match.group(2)
|
||
result.setdefault(url, MediaCandidate(source_page, url, _kind(url, None, source_page), None, None))
|
||
return list(result.values())
|
||
|
||
|
||
def merge_manifest(path: Path, candidates: list[MediaCandidate]) -> dict:
|
||
current = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {"version": 1, "assets": []}
|
||
assets = {item["asset_url"]: item for item in current.get("assets", [])}
|
||
seen_at = datetime.now(timezone.utc).isoformat()
|
||
for candidate in candidates:
|
||
item = assets.get(candidate.asset_url, {})
|
||
source_pages = set(item.get("source_pages", []))
|
||
if item.get("source_page"):
|
||
source_pages.add(item["source_page"])
|
||
source_pages.add(candidate.source_page)
|
||
item.update(asdict(candidate))
|
||
item["source_pages"] = sorted(source_pages)
|
||
item.setdefault("status", "queued")
|
||
item.setdefault("first_seen_at", seen_at)
|
||
item["last_seen_at"] = seen_at
|
||
assets[candidate.asset_url] = item
|
||
output = {"version": 1, "updated_at": seen_at, "assets": sorted(assets.values(), key=lambda item: item["asset_url"])}
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return output
|
||
|
||
|
||
def normalize_entity_label(value: str) -> str:
|
||
"""Normalize conservative Russian catalog spelling variants for matching."""
|
||
return " ".join(value.casefold().replace("ё", "е").split())
|
||
|
||
|
||
def reconcile_queued_duplicates(path: Path) -> dict:
|
||
"""Quarantine queued label duplicates while preserving their provenance."""
|
||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||
saved: dict[tuple[str, str], dict] = {}
|
||
for item in manifest.get("assets", []):
|
||
parts = urlsplit(str(item.get("asset_url") or ""))
|
||
if not item.get("external_id") and parts.hostname == "oss.rf4db.com" and parts.path.startswith("/game/"):
|
||
item["external_id"] = Path(parts.path).stem
|
||
label = item.get("label")
|
||
if item.get("status") not in {"stored", "approved"} or not isinstance(label, str):
|
||
continue
|
||
saved.setdefault((str(item.get("entity_type")), normalize_entity_label(label)), item)
|
||
duplicates = 0
|
||
for item in manifest.get("assets", []):
|
||
label = item.get("label")
|
||
if item.get("status") != "queued" or not isinstance(label, str):
|
||
continue
|
||
match = saved.get((str(item.get("entity_type")), normalize_entity_label(label)))
|
||
if match and match.get("asset_url") != item.get("asset_url"):
|
||
item.update({"status": "duplicate", "duplicate_of": match["asset_url"]})
|
||
duplicates += 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 {"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"))
|
||
reviewed_at = datetime.now(timezone.utc).isoformat()
|
||
approved = 0
|
||
for item in manifest.get("assets", []):
|
||
if item.get("status") != "stored":
|
||
continue
|
||
stable_id = item.get("external_id") or item.get("sha256") or hashlib.sha256(item["asset_url"].encode()).hexdigest()
|
||
item.update({
|
||
"status": "approved",
|
||
"entity_key": f"{item.get('entity_type', 'reference')}:{stable_id}",
|
||
"reviewed_at": reviewed_at,
|
||
"review_note": note,
|
||
})
|
||
approved += 1
|
||
manifest["updated_at"] = reviewed_at
|
||
path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
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 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", []):
|
||
item.setdefault("source_pages", [item["source_page"]])
|
||
item["entity_type"] = _kind(item["asset_url"], item.get("label"), item.get("source_page", ""))
|
||
path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return manifest
|
||
|
||
|
||
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")
|
||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||
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(),
|
||
"review_note": note or None,
|
||
})
|
||
path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return item
|
||
|
||
|
||
def audit_media_catalog(root: Path) -> dict:
|
||
manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
|
||
statuses: dict[str, int] = {}
|
||
issues: list[str] = []
|
||
referenced: set[str] = set()
|
||
for item in manifest.get("assets", []):
|
||
status = str(item.get("status", "unknown"))
|
||
statuses[status] = statuses.get(status, 0) + 1
|
||
local_path = item.get("local_path")
|
||
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
|
||
referenced.add(local_path)
|
||
target = root / local_path
|
||
if not target.is_file():
|
||
issues.append(f"{item['asset_url']}: local file is missing")
|
||
continue
|
||
body = target.read_bytes()
|
||
if hashlib.sha256(body).hexdigest() != item.get("sha256"):
|
||
issues.append(f"{item['asset_url']}: SHA-256 mismatch")
|
||
try:
|
||
width, height, mime = inspect_image(body)
|
||
if (width, height, mime) != (item.get("width"), item.get("height"), item.get("content_type")):
|
||
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"
|
||
orphaned = sorted(str(path.relative_to(root)) for path in files_root.rglob("*") if path.is_file() and str(path.relative_to(root)) not in referenced) if files_root.exists() else []
|
||
return {"total": sum(statuses.values()), "statuses": statuses, "issues": issues, "orphaned_files": orphaned}
|
||
|
||
|
||
def media_coverage(root: Path) -> dict:
|
||
"""Compare quarantined and approved media with the versioned catalog baseline."""
|
||
manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
|
||
baseline = json.loads((root / "catalog-baseline.json").read_text(encoding="utf-8"))
|
||
result: dict[str, dict] = {}
|
||
for entity_type, target in baseline["entities"].items():
|
||
candidates = [item for item in manifest.get("assets", []) if item.get("entity_type") == entity_type]
|
||
candidate_labels = {
|
||
normalize_entity_label(item["label"])
|
||
for item in candidates
|
||
if isinstance(item.get("label"), str) and item["label"].strip()
|
||
}
|
||
approved_keys = {item.get("entity_key") for item in candidates if item.get("status") == "approved" and item.get("entity_key")}
|
||
expected = target.get("count")
|
||
result[entity_type] = {
|
||
"expected": expected,
|
||
"candidate_assets": len(candidates),
|
||
"unique_candidates": len(candidate_labels),
|
||
"unidentified_assets": sum(1 for item in candidates if not isinstance(item.get("label"), str) or not item["label"].strip()),
|
||
"approved": len(approved_keys),
|
||
"candidate_gap": max(expected - len(candidate_labels), 0) if isinstance(expected, int) else None,
|
||
"approved_gap": max(expected - len(approved_keys), 0) if isinstance(expected, int) else None,
|
||
}
|
||
return {"baseline_date": baseline["verified_at"], "entities": result}
|
||
|
||
|
||
def media_quality_report(root: Path, *, minimum_dimension: int = 256, display_dimension: int = 180) -> dict:
|
||
"""Report published raster quality and known higher-quality source alternatives offline."""
|
||
manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
|
||
assets = manifest.get("assets", [])
|
||
approved = [item for item in assets if item.get("status") == "approved" and item.get("entity_type") == "fish"]
|
||
low = [item for item in approved if min(int(item.get("width") or 0), int(item.get("height") or 0)) < minimum_dimension]
|
||
upsampled = [item for item in approved if min(int(item.get("width") or 0), int(item.get("height") or 0)) < display_dimension]
|
||
low_urls = {item.get("asset_url") for item in low}
|
||
alternatives = [
|
||
item for item in assets
|
||
if item.get("entity_type") == "fish"
|
||
and item.get("status") in {"duplicate", "queued", "upgrade_queued", "upgrade_stored"}
|
||
and (item.get("duplicate_of") in low_urls or item.get("status") == "queued")
|
||
]
|
||
|
||
by_source: dict[str, dict[str, int]] = {}
|
||
for item in approved:
|
||
host = urlsplit(str(item.get("asset_url") or "")).hostname or "unknown"
|
||
summary = by_source.setdefault(host, {"published": 0, "below_minimum": 0, "upsampled_in_cards": 0})
|
||
summary["published"] += 1
|
||
summary["below_minimum"] += item in low
|
||
summary["upsampled_in_cards"] += item in upsampled
|
||
|
||
return {
|
||
"entity_type": "fish",
|
||
"minimum_dimension": minimum_dimension,
|
||
"display_dimension": display_dimension,
|
||
"published": len(approved),
|
||
"below_minimum": len(low),
|
||
"upsampled_in_cards": len(upsampled),
|
||
"known_alternative_urls": len(alternatives),
|
||
"by_source": by_source,
|
||
"examples": [
|
||
{"label": item.get("label"), "width": item.get("width"), "height": item.get("height"), "asset_url": item.get("asset_url")}
|
||
for item in sorted(low, key=lambda row: str(row.get("label") or ""))[:20]
|
||
],
|
||
}
|
||
|
||
|
||
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 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:
|
||
width, height = image.size
|
||
image_format = image.format
|
||
image.verify()
|
||
except (UnidentifiedImageError, OSError, ValueError) as exc:
|
||
raise ValueError("asset is not a valid raster image") from exc
|
||
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", "image/avif"}:
|
||
raise ValueError(f"unsupported image format: {image_format}")
|
||
return width, height, mime
|
||
|
||
|
||
def store_asset(root: Path, body: bytes, *, content_type: str, source_url: str) -> tuple[str, str, int, int, str]:
|
||
width, height, detected_mime = inspect_image(body)
|
||
declared_mime = content_type.split(";", 1)[0]
|
||
if declared_mime != detected_mime:
|
||
raise ValueError(f"image MIME mismatch: declared {declared_mime}, detected {detected_mime}")
|
||
digest = hashlib.sha256(body).hexdigest()
|
||
extension = mimetypes.guess_extension(detected_mime) or Path(urlsplit(source_url).path).suffix
|
||
extension = ".jpg" if extension == ".jpe" else extension
|
||
target = root / "files" / digest[:2] / f"{digest}{extension}"
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
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}
|