fengchuanhn@gmail.com
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
|