11

app/core/branding_images.py

@@ -0,0 +1,119 @@
+"""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