222 lines
11 KiB
Python
222 lines
11 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 or any(word in value for word in ("карта ", "map of ", "gewässerkarte")):
|
|
return "waterbody"
|
|
return "reference"
|
|
|
|
|
|
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 ""
|
|
link = node.find_parent("a", href=True)
|
|
external_id = str(link.get("href")).rstrip("/").rsplit("/", 1)[-1] if link else None
|
|
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 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 = {
|
|
" ".join(item["label"].split()).casefold()
|
|
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
|