"""OEM 品牌图:落盘至项目根目录 images/,库内仅存相对路径。""" from __future__ import annotations import base64 import re import uuid from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] IMAGES_DIR = PROJECT_ROOT / "images" DATA_URL_PATTERN = re.compile(r"^data:image/([\w+.-]+);base64,(.+)$", re.DOTALL) MAX_IMAGE_BYTES = 5 * 1024 * 1024 EXT_ALIASES = {"jpeg": "jpg", "svg+xml": "svg"} class BrandingImageError(Exception): def __init__(self, message: str) -> None: self.message = message super().__init__(message) def ensure_images_dir() -> None: IMAGES_DIR.mkdir(parents=True, exist_ok=True) def is_data_url(value: str) -> bool: return value.strip().startswith("data:image/") def normalize_relative_path(value: str | None) -> str | None: """从完整 URL 或相对路径得到 images/xxx 形式。""" if not value: return None text = value.strip() if not text or is_data_url(text): return None marker = "/images/" if marker in text: suffix = text.split(marker, 1)[1].split("?")[0] return f"images/{suffix}" normalized = text.replace("\\", "/").lstrip("/") if normalized.startswith("images/"): return normalized return None def absolute_path(relative_path: str | None) -> Path | None: rel = normalize_relative_path(relative_path) if not rel: return None name = rel.removeprefix("images/").lstrip("/") if not name or ".." in Path(name).parts: return None return IMAGES_DIR / name def delete_local_image(relative_path: str | None) -> None: path = absolute_path(relative_path) if path is not None and path.is_file(): path.unlink() def save_data_url(data_url: str, *, file_prefix: str) -> str: match = DATA_URL_PATTERN.match(data_url.strip()) if not match: raise BrandingImageError("无效的图片数据") ext = match.group(1).lower() ext = EXT_ALIASES.get(ext, ext) if ext not in {"png", "jpg", "gif", "webp", "svg"}: raise BrandingImageError(f"不支持的图片格式: {ext}") try: raw = base64.b64decode(match.group(2), validate=True) except Exception as exc: raise BrandingImageError("图片 Base64 解码失败") from exc if len(raw) > MAX_IMAGE_BYTES: raise BrandingImageError("单张图片不能超过 5MB") ensure_images_dir() filename = f"oem_{file_prefix}_{uuid.uuid4().hex[:12]}.{ext}" target = IMAGES_DIR / filename target.write_bytes(raw) return f"images/{filename}" def resolve_image_field( new_value: str | None, old_path: str | None, *, file_prefix: str, ) -> str | None: """ - 空:删除旧文件,返回 None - data URL:写入新文件,删除旧文件 - 已是 images/ 路径(或与旧路径相同):原样保留 """ if new_value is None or not str(new_value).strip(): delete_local_image(old_path) return None text = str(new_value).strip() if is_data_url(text): delete_local_image(old_path) return save_data_url(text, file_prefix=file_prefix) rel = normalize_relative_path(text) if rel is None: raise BrandingImageError("图片须为上传数据或 images/ 下的路径") old_rel = normalize_relative_path(old_path) if rel == old_rel: return rel return rel