152 lines
6.6 KiB
Python
152 lines
6.6 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 re.search(r"/(?:fish|fishes|fische|species)/", path) or any(word in value for word in ("рыба ", "fish: ", "fisch: ")):
|
|
return "fish"
|
|
if re.search(r"/(?:maps?|levels?|lakes?|waterbodies)/", path) or any(word in value for word in ("карта ", "map of ", "gewässerkarte")):
|
|
return "waterbody"
|
|
if re.search(r"(?:^|[/_-])(?:bait|lure|rig|rod|reel|hook|tackle|köder|rute|rolle)(?:[/_.-]|$)", path):
|
|
return "tackle"
|
|
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, 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, ""), 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, {})
|
|
item.update(asdict(candidate))
|
|
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["entity_type"] = _kind(item["asset_url"], item.get("label"), "")
|
|
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 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
|