"""Validated, content-addressed custom background images for location maps.""" from __future__ import annotations import base64 import binascii import hashlib import io import re from pathlib import Path from typing import Any, Callable from PIL import Image, ImageOps, UnidentifiedImageError from src.webui.map_presets import builtin_map_preset_by_asset MAX_MAP_BACKGROUND_BYTES = 8 * 1024 * 1024 MAX_MAP_BACKGROUND_PIXELS = 24_000_000 MAX_MAP_BACKGROUND_EDGE = 2048 ASSET_ID_RE = re.compile(r"^[a-f0-9]{64}$") def validate_map_background_selection( backgrounds_dir: Path, generated_image_file: Callable[[str], Path | None], selection: Any, ) -> dict[str, str]: """Normalize a persisted selection without accepting arbitrary URLs.""" if not selection: return {"kind": "auto"} if not isinstance(selection, dict): raise ValueError("地图背景选择必须是对象") kind = str(selection.get("kind") or "").strip() if kind in {"auto", "none"}: return {"kind": kind} if kind == "builtin": asset_id = str(selection.get("id") or "").strip() if builtin_map_preset_by_asset(asset_id) is None: raise ValueError("未知的内置地图背景") return {"kind": "builtin", "id": asset_id} if kind == "upload": asset_id = str(selection.get("asset_id") or "").strip() if map_background_file(backgrounds_dir, asset_id) is None: raise ValueError("上传的地图背景不存在") return {"kind": "upload", "asset_id": asset_id} if kind == "generated": asset_id = str(selection.get("asset_id") or "").strip() if generated_image_file(asset_id) is None: raise ValueError("生成的地图背景不存在") return {"kind": "generated", "asset_id": asset_id} if kind == "plugin": map_id = str(selection.get("map_id") or "").strip() if not map_id or len(map_id) > 256 or not map_id.startswith("plugin:"): raise ValueError("内容包地图引用无效") return {"kind": "plugin", "map_id": map_id} raise ValueError("不支持的地图背景选择") def save_map_background_upload( backgrounds_dir: Path, file_data: str, file_name: str = "", ) -> dict[str, Any]: if not file_data: return {"ok": False, "error": "未提供地图背景文件"} if len(file_data) > (MAX_MAP_BACKGROUND_BYTES * 4 // 3) + 32: return {"ok": False, "error": "地图背景不能超过 8 MB"} try: raw = base64.b64decode(file_data, validate=True) except (ValueError, binascii.Error): return {"ok": False, "error": "地图背景文件数据无效"} if not raw or len(raw) > MAX_MAP_BACKGROUND_BYTES: return {"ok": False, "error": "地图背景不能超过 8 MB"} try: payload = _normalized_map_background(raw) except ValueError as exc: return {"ok": False, "error": str(exc)} asset_id = hashlib.sha256(payload).hexdigest() path = backgrounds_dir / f"{asset_id}.webp" path.parent.mkdir(parents=True, exist_ok=True) if not path.exists(): temporary = path.with_suffix(".webp.tmp") temporary.write_bytes(payload) temporary.replace(path) return { "ok": True, "map_background": {"kind": "upload", "asset_id": asset_id}, "file_name": Path(file_name).name, } def map_background_file(backgrounds_dir: Path, asset_id: str) -> Path | None: if not ASSET_ID_RE.fullmatch(str(asset_id or "")): return None path = backgrounds_dir / f"{asset_id}.webp" return path if path.is_file() else None def resolve_map_background_file( backgrounds_dir: Path, generated_image_file: Callable[[str], Path | None], selection: Any, ) -> Path | None: try: normalized = validate_map_background_selection( backgrounds_dir, generated_image_file, selection, ) except ValueError: return None if normalized["kind"] == "upload": return map_background_file(backgrounds_dir, normalized["asset_id"]) if normalized["kind"] == "generated": return generated_image_file(normalized["asset_id"]) return None def _normalized_map_background(raw: bytes) -> bytes: try: with Image.open(io.BytesIO(raw)) as source: if source.format not in {"PNG", "JPEG", "WEBP"}: raise ValueError("地图背景仅支持 PNG、JPEG 或 WebP") width, height = source.size if width < 320 or height < 180: raise ValueError("地图背景尺寸不能小于 320×180") if width > 8192 or height > 8192 or width * height > MAX_MAP_BACKGROUND_PIXELS: raise ValueError("地图背景尺寸过大") image = ImageOps.exif_transpose(source).convert("RGB") if max(image.size) > MAX_MAP_BACKGROUND_EDGE: image.thumbnail( (MAX_MAP_BACKGROUND_EDGE, MAX_MAP_BACKGROUND_EDGE), Image.Resampling.LANCZOS, ) output = io.BytesIO() image.save(output, format="WEBP", quality=84, method=6) return output.getvalue() except ValueError: raise except (UnidentifiedImageError, OSError, Image.DecompressionBombError) as exc: raise ValueError("无法读取地图背景") from exc class MapBackgroundService: """Validated background storage with one explicit generated-asset boundary.""" def __init__( self, backgrounds_dir: Path, generated_image_file: Callable[[str], Path | None], ) -> None: self._backgrounds_dir = backgrounds_dir self._generated_image_file = generated_image_file def validate(self, selection: Any) -> dict[str, str]: return validate_map_background_selection( self._backgrounds_dir, self._generated_image_file, selection, ) def save_upload( self, file_data: str, file_name: str = "", ) -> dict[str, Any]: return save_map_background_upload( self._backgrounds_dir, file_data, file_name, ) def file(self, asset_id: str) -> Path | None: return map_background_file(self._backgrounds_dir, asset_id) def resolve_file(self, selection: Any) -> Path | None: return resolve_map_background_file( self._backgrounds_dir, self._generated_image_file, selection, )