191 lines
8.0 KiB
Python
191 lines
8.0 KiB
Python
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")
|