from __future__ import annotations import json import os from pathlib import Path from urllib.parse import urlsplit MEDIA_ROOT = Path(os.environ.get("MEDIA_ROOT", "data/media")).resolve() WATERBODY_MEDIA_ROLES = {"waterbody_cover", "waterbody_map", "waterbody_depth_map", "waterbody_screenshot"} TACKLE_MEDIA_ROLES = {"tackle_card", "tackle_detail", "rig_diagram", "tackle_screenshot"} KNOWN_MEDIA_ROLES = WATERBODY_MEDIA_ROLES | TACKLE_MEDIA_ROLES MEDIA_ROLES_BY_ENTITY = {"waterbody": WATERBODY_MEDIA_ROLES, "tackle": TACKLE_MEDIA_ROLES} _manifest_cache: tuple[Path, int, int, dict] | None = None _asset_index_cache: tuple[int, dict[str, tuple[str, str | None]]] | None = None def _read_manifest() -> dict: """Read the manifest once per file revision in this API process.""" global _manifest_cache path = (MEDIA_ROOT / "manifest.json").resolve() stat = path.stat() marker = (path, stat.st_mtime_ns, stat.st_size) if _manifest_cache and _manifest_cache[:3] == marker: return _manifest_cache[3] manifest = json.loads(path.read_text(encoding="utf-8")) if not isinstance(manifest, dict): raise ValueError("media manifest must be an object") _manifest_cache = (*marker, manifest) return manifest def _source_system(source_page: object) -> str: """Map only allowlisted hostnames; never promote an unknown URL to official.""" try: parsed = urlsplit(str(source_page or "")) except ValueError: return "unknown" hostname = (parsed.hostname or "").lower().rstrip(".") if hostname == "rf4db.com" or hostname.endswith(".rf4db.com"): return "rf4db" if hostname == "rf4map.ru" or hostname.endswith(".rf4map.ru"): return "rf4map" if hostname == "rf4-stat.ru" or hostname.endswith(".rf4-stat.ru"): return "rf4stat" return "unknown" def media_manifest_version() -> int: manifest = _read_manifest() return max(1, int(manifest.get("version", 1))) def _public_role_allowed(entity_type: str | None, role: object) -> bool: if role is None: return True if not isinstance(role, str) or role not in KNOWN_MEDIA_ROLES: return False if entity_type == "waterbody": return role in WATERBODY_MEDIA_ROLES if entity_type == "tackle": return role in TACKLE_MEDIA_ROLES return False def published_assets(entity_type: str | None = None, media_role: str | None = None) -> list[dict]: manifest = _read_manifest() result = [] for item in manifest.get("assets", []): if item.get("status") != "approved" or not item.get("sha256") or not item.get("local_path"): continue if entity_type and item.get("entity_type") != entity_type: continue if media_role and item.get("media_role") != media_role: continue if not _public_role_allowed(item.get("entity_type"), item.get("media_role")): continue source_page = str(item.get("source_page") or "") source = _source_system(source_page) result.append({ "id": item["sha256"], "entity_type": item.get("entity_type"), "entity_key": item.get("entity_key"), "media_role": item.get("media_role"), "label": item.get("label"), "width": item.get("width"), "height": item.get("height"), "content_type": item.get("content_type"), "image_url": f"/api/v1/media/assets/{item['sha256']}", "source_system": source, "source_url": source_page, "variants": [ { "role": variant.get("role"), "format": variant.get("format"), "width": variant.get("width"), "height": variant.get("height"), "url": f"/api/v1/media/assets/{variant['sha256']}", } for variant in item.get("derivatives", []) if variant.get("sha256") and variant.get("local_path") ], }) return sorted(result, key=lambda item: (str(item["entity_type"]), str(item["label"] or "").casefold(), item["id"])) def published_file(digest: str) -> tuple[Path, str] | None: if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest): return None manifest = _read_manifest() global _asset_index_cache cache_key = id(manifest) if _asset_index_cache is None or _asset_index_cache[0] != cache_key: index: dict[str, tuple[str, str | None]] = {} for row in manifest.get("assets", []): if row.get("status") != "approved": continue if row.get("sha256") and row.get("local_path"): index[str(row["sha256"])] = (str(row["local_path"]), row.get("content_type")) for variant in row.get("derivatives", []): if variant.get("sha256") and variant.get("local_path"): index.setdefault(str(variant["sha256"]), (str(variant["local_path"]), variant.get("content_type"))) _asset_index_cache = (cache_key, index) asset = _asset_index_cache[1].get(digest) if asset is None: return None local_path, media_type = asset if not local_path: return None target = (MEDIA_ROOT / local_path).resolve() if not target.is_relative_to(MEDIA_ROOT.resolve()) or not target.is_file(): return None return target, str(media_type or "application/octet-stream") def review_assets( entity_type: str | None = None, status: str | None = None, media_role: str | None = None, query: str | None = None, ) -> list[dict]: manifest = _read_manifest() result = [] for item in manifest.get("assets", []): item_status = str(item.get("status") or "") if item_status not in {"approved", "upgrade_queued", "upgrade_stored"} or (status and item_status != status): continue if entity_type and item.get("entity_type") != entity_type: continue if media_role and item.get("media_role") != media_role: continue if query and query.casefold() not in str(item.get("label") or "").casefold(): continue digest = str(item.get("sha256") or "") if len(digest) != 64 or not item.get("local_path"): continue source_page = str(item.get("source_page") or "") source = _source_system(source_page) result.append({ "id": digest, "status": item_status, "entity_type": item.get("entity_type"), "entity_key": item.get("entity_key"), "label": item.get("label"), "width": item.get("width"), "height": item.get("height"), "content_type": item.get("content_type"), "image_url": f"/api/v1/admin/media/assets/{digest}", "asset_url": item.get("asset_url", ""), "source_system": source, "source_url": source_page, "duplicate_of": item.get("duplicate_of"), "supersedes": item.get("supersedes"), "derivatives": [{ "role": variant.get("role"), "format": variant.get("format"), "width": variant.get("width"), "height": variant.get("height"), } for variant in item.get("derivatives", [])], }) return sorted(result, key=lambda item: (str(item["status"]), str(item["entity_type"]), str(item["label"] or "").casefold(), item["id"])) def review_file(digest: str) -> tuple[Path, str] | None: if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest): return None manifest = _read_manifest() item = next((row for row in manifest.get("assets", []) if row.get("sha256") == digest and row.get("status") in {"approved", "upgrade_queued", "upgrade_stored"}), None) if not item or not item.get("local_path"): return None target = (MEDIA_ROOT / item["local_path"]).resolve() if not target.is_relative_to(MEDIA_ROOT.resolve()) or not target.is_file(): return None return target, str(item.get("content_type") or "application/octet-stream")