283 lines
13 KiB
Python
283 lines
13 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, UnidentifiedImageError
|
||
|
||
MAX_IMAGE_PIXELS = 40_000_000
|
||
|
||
|
||
@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 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 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,
|
||
) -> 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 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})
|
||
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", "approved"}:
|
||
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}")
|
||
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 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"}:
|
||
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
|