120 lines
3.3 KiB
Python
120 lines
3.3 KiB
Python
"""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
|