448 lines
22 KiB
Python
448 lines
22 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 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 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", "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}")
|
||
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 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
|