feat: add rate-limited RF4 media collector

rf4_research/media_assets.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import hashlib
+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
+
+
+@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 store_asset(root: Path, body: bytes, *, content_type: str, source_url: str) -> tuple[str, str]:
+    digest = hashlib.sha256(body).hexdigest()
+    extension = mimetypes.guess_extension(content_type.split(";", 1)[0]) or Path(urlsplit(source_url).path).suffix
+    extension = extension if extension in {".jpg", ".jpeg", ".png", ".webp", ".gif", ".svg"} else ".bin"
+    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))