R524809
121 lines
4.0 KiB
Python
121 lines
4.0 KiB
Python
"""导出采集图片：把采集到的源站图片打包成 ZIP 下载到本地。
|
||
|
||
参考图 URL 可能是源站 CDN（需 Referer 绕过防盗链）或本地上传的 media 文件。
|
||
ZIP 内部结构沿用现有分组名建子文件夹（主图 / SKU图片 / 详情图 / 手动上传），
|
||
文件名沿用采集 key（main-001 等）+ SKU 规格名；顶层文件夹用商品标题（清洗后）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import mimetypes
|
||
import re
|
||
import zipfile
|
||
|
||
from fastapi import APIRouter, HTTPException
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from api.proxy import guess_referer
|
||
from services import storage
|
||
|
||
router = APIRouter(prefix="/api", tags=["export"])
|
||
|
||
_EXT_BY_CTYPE = {
|
||
"image/jpeg": ".jpg",
|
||
"image/png": ".png",
|
||
"image/webp": ".webp",
|
||
"image/gif": ".gif",
|
||
"image/bmp": ".bmp",
|
||
}
|
||
|
||
|
||
class ExportImageItem(BaseModel):
|
||
url: str
|
||
groupName: str = "主图"
|
||
variantName: str | None = None
|
||
key: str = "" # 采集 key，如 main-001 / sku-002 / upload-001
|
||
|
||
|
||
class ExportImagesRequest(BaseModel):
|
||
title: str | None = Field(default=None, description="商品标题，用作 ZIP 顶层文件夹名")
|
||
images: list[ExportImageItem]
|
||
|
||
|
||
def _clean(name: str) -> str:
|
||
"""清洗文件夹/文件名非法字符（与插件 cleanFilename 同规则，Windows 兼容）。"""
|
||
s = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", (name or "").strip())
|
||
s = re.sub(r"\s+", "_", s)
|
||
return s.strip(" .")[:80]
|
||
|
||
|
||
def _ext(url: str, ctype: str) -> str:
|
||
"""由 content-type（优先）或 URL 后缀决定扩展名。"""
|
||
ctype = ctype.split(";")[0].strip().lower()
|
||
if ctype in _EXT_BY_CTYPE:
|
||
return _EXT_BY_CTYPE[ctype]
|
||
if ctype.startswith("image/"):
|
||
return "." + ctype.split("/")[-1]
|
||
path = url.split("?")[0].lower()
|
||
for ext in (".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"):
|
||
if path.endswith(ext):
|
||
return ".jpg" if ext == ".jpeg" else ext
|
||
return ".jpg"
|
||
|
||
|
||
def _is_image(url: str, ctype: str) -> bool:
|
||
ctype = ctype.split(";")[0].strip().lower()
|
||
if ctype.startswith("image/"):
|
||
return True
|
||
return bool(re.search(r"\.(jpe?g|png|webp|gif|bmp)(\?|$)", url, re.IGNORECASE))
|
||
|
||
|
||
async def _download(url: str) -> tuple[bytes, str]:
|
||
"""本地 media 文件直读磁盘；远程 URL 带 Referer 下载。"""
|
||
path = storage.local_path(url)
|
||
if path is not None:
|
||
mime = mimetypes.guess_type(path.name)[0] or "image/jpeg"
|
||
return path.read_bytes(), mime
|
||
return await storage.download_bytes(url, referer=guess_referer(url))
|
||
|
||
|
||
@router.post("/export-images")
|
||
async def export_images(req: ExportImagesRequest):
|
||
if not req.images:
|
||
raise HTTPException(status_code=400, detail="没有可导出的图片")
|
||
|
||
root = _clean(req.title) or "采集图片"
|
||
buf = io.BytesIO()
|
||
used: set[str] = set()
|
||
|
||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for img in req.images:
|
||
try:
|
||
data, ctype = await _download(img.url)
|
||
except Exception: # noqa: BLE001
|
||
continue # 单张失败不中断整包
|
||
if not _is_image(img.url, ctype):
|
||
continue
|
||
|
||
ext = _ext(img.url, ctype)
|
||
base = _clean(img.key) or "image"
|
||
if img.variantName:
|
||
base += f"-{_clean(img.variantName)}"
|
||
filename = f"{base}{ext}"
|
||
if filename in used: # 同名加序号防覆盖
|
||
stem = filename[: -len(ext)]
|
||
n = 2
|
||
while f"{stem}-{n}{ext}" in used:
|
||
n += 1
|
||
filename = f"{stem}-{n}{ext}"
|
||
used.add(filename)
|
||
|
||
group = _clean(img.groupName) or "图片"
|
||
zf.writestr(f"{root}/{group}/{filename}", data)
|
||
|
||
buf.seek(0)
|
||
return StreamingResponse(
|
||
buf,
|
||
media_type="application/zip",
|
||
headers={"Content-Disposition": 'attachment; filename="collect.zip"'},
|
||
)
|