2026 lines
76 KiB
Python
2026 lines
76 KiB
Python
from __future__ import annotations
|
||
|
||
import io
|
||
import json
|
||
import hashlib
|
||
import logging
|
||
import os
|
||
import re
|
||
import shutil
|
||
import sqlite3
|
||
import threading
|
||
import time
|
||
import uuid
|
||
import zipfile
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFile
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.responses import FileResponse, StreamingResponse
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
|
||
from core import config as wc_config
|
||
from core.fonts import get_cached_font
|
||
from .job_manager import JobManager
|
||
from .line_spacing import analyze_svg_line_spacing_file
|
||
from .log_config import get_logger
|
||
from .metadata_store import MetadataStore
|
||
from .runner import JobRunner
|
||
from .schemas import (
|
||
Asset,
|
||
DesignTemplate,
|
||
Font,
|
||
JobCreateResponse,
|
||
JobDetail,
|
||
JobLocationSearchResult,
|
||
JobResult,
|
||
JobStatus,
|
||
LineSpacingAnalysisRequest,
|
||
LineSpacingAnalysisSummary,
|
||
Project,
|
||
ProjectSummary,
|
||
Template,
|
||
WordLocation,
|
||
)
|
||
from .storage import Storage
|
||
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
WORKSPACE_DIR = PROJECT_ROOT / "service_workspace"
|
||
|
||
log = get_logger("service.app")
|
||
log.info("=" * 60)
|
||
log.info("服务启动 | PROJECT_ROOT=%s", PROJECT_ROOT)
|
||
log.info("=" * 60)
|
||
|
||
# ── new directories ──────────────────────────────────────────
|
||
ASSETS_DIR = PROJECT_ROOT / "service_assets"
|
||
PROJECTS_DIR = PROJECT_ROOT / "service_projects"
|
||
FONTS_DIR = PROJECT_ROOT / "service_fonts"
|
||
DESIGN_TEMPLATES_DIR = PROJECT_ROOT / "service_design_templates"
|
||
METADATA_DIR = PROJECT_ROOT / "service_metadata"
|
||
ORDERS_DIR = PROJECT_ROOT / "service_orders"
|
||
ASSETS_DIR.mkdir(parents=True, exist_ok=True)
|
||
PROJECTS_DIR.mkdir(parents=True, exist_ok=True)
|
||
FONTS_DIR.mkdir(parents=True, exist_ok=True)
|
||
DESIGN_TEMPLATES_DIR.mkdir(parents=True, exist_ok=True)
|
||
METADATA_DIR.mkdir(parents=True, exist_ok=True)
|
||
ORDERS_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
metadata_store = MetadataStore(METADATA_DIR / "app.db")
|
||
manager = JobManager(metadata_store)
|
||
storage = Storage(WORKSPACE_DIR)
|
||
runner = JobRunner(PROJECT_ROOT, manager)
|
||
|
||
app = FastAPI(title="WordCloud Test Service", version="0.1.0")
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 1. 模板（硬编码）
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
_TEMPLATES: list[Template] = [
|
||
Template(id="poster_1x2", name="竖版手机海报", width=1080, height=2160, aspect_ratio="1:2", description="适合手机海报、宣传页"),
|
||
Template(id="poster_4x5", name="社交媒体图", width=1080, height=1350, aspect_ratio="4:5", description="小红书、Instagram 风格"),
|
||
Template(id="poster_1x1", name="方形封面", width=1080, height=1080, aspect_ratio="1:1", description="朋友圈封面、头像"),
|
||
Template(id="poster_3x4", name="竖版广告", width=1080, height=1440, aspect_ratio="3:4", description="通用竖版海报"),
|
||
Template(id="poster_16x9", name="横版电商", width=1920, height=1080, aspect_ratio="16:9", description="横版横幅、电商头图"),
|
||
]
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 2. Helpers
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _safe_hex_color(value: str) -> bool:
|
||
return bool(re.fullmatch(r"#[0-9A-Fa-f]{6}", value))
|
||
|
||
|
||
def _read_asset_meta(asset_dir: Path) -> dict:
|
||
meta_path = asset_dir / "meta.json"
|
||
if not meta_path.exists():
|
||
return {}
|
||
return json.loads(meta_path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _write_asset_meta(asset_dir: Path, data: dict) -> None:
|
||
meta_path = asset_dir / "meta.json"
|
||
meta_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
|
||
def _asset_dir(asset_id: str) -> Path:
|
||
return ASSETS_DIR / asset_id[:2] / asset_id
|
||
|
||
|
||
def _project_dir(project_id: str) -> Path:
|
||
return PROJECTS_DIR / project_id[:2] / project_id
|
||
|
||
|
||
def _design_template_dir(template_id: str) -> Path:
|
||
return DESIGN_TEMPLATES_DIR / template_id[:2] / template_id
|
||
|
||
|
||
def _list_dirs(p: Path) -> list[Path]:
|
||
"""递归列出所有目录下包含有效子目录的 path."""
|
||
result: list[Path] = []
|
||
if not p.exists():
|
||
return result
|
||
for item in p.iterdir():
|
||
if item.is_dir():
|
||
if (item / "meta.json").exists() or (item / "project.json").exists() or (item / "template.json").exists():
|
||
result.append(item)
|
||
else:
|
||
result.extend(_list_dirs(item))
|
||
return result
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 3. Health & Jobs（原有接口，精简保留）
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
@app.get("/api/health")
|
||
def health() -> dict:
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/api/maintenance/storage-summary")
|
||
def storage_summary() -> dict:
|
||
referenced_jobs: set[str] = set()
|
||
for d in _list_dirs(ASSETS_DIR):
|
||
meta = _read_asset_meta(d)
|
||
job_id = meta.get("job_id") or ""
|
||
if job_id:
|
||
referenced_jobs.add(job_id)
|
||
stale = storage.stale_job_dirs(
|
||
referenced_job_ids=referenced_jobs,
|
||
exclude_job_ids=metadata_store.job_ids(),
|
||
max_age_days=0,
|
||
)
|
||
return {
|
||
"job_dir_count": len(storage.list_job_ids()),
|
||
"referenced_job_ids": len(referenced_jobs),
|
||
"stale_job_count": len(stale),
|
||
"reclaimable_bytes": sum(item["size_bytes"] for item in stale),
|
||
"metadata_db_bytes": metadata_store.summarize()["db_size_bytes"],
|
||
"jobs_in_db": metadata_store.summarize()["jobs"],
|
||
"events_in_db": metadata_store.summarize()["events"],
|
||
"dry_run_only": True,
|
||
}
|
||
|
||
|
||
@app.get("/api/jobs", response_model=list[JobStatus])
|
||
def list_jobs() -> list[JobStatus]:
|
||
"""列出全部任务，含仍在 metadata store 登记的管理任务，以及磁盘上已完成但未登记的任务。"""
|
||
with manager._lock:
|
||
statuses = [state.status for state in manager._jobs.values()]
|
||
|
||
# 补充磁盘上已落成的任务（manager 只记住本次运行/登记过的；历史任务目录
|
||
# 若未被 metadata store 登记，则从 workspace 扫描补齐，供查找页跨任务使用）。
|
||
# 与详情接口的 _scan_disk_job_status 同一套产物推断，保证列表可见即可访问。
|
||
tracked = {s.job_id for s in statuses}
|
||
for job_id in storage.list_job_ids():
|
||
if job_id in tracked:
|
||
continue
|
||
scanned = _scan_disk_job_status(job_id)
|
||
if scanned is not None:
|
||
statuses.append(scanned)
|
||
return list(reversed(statuses))
|
||
|
||
|
||
@app.post("/api/jobs", response_model=JobCreateResponse)
|
||
async def create_job(
|
||
mask_image: Optional[UploadFile] = File(None),
|
||
name_list: Optional[UploadFile] = File(None),
|
||
wcd_file: Optional[UploadFile] = File(None),
|
||
font_file: Optional[UploadFile] = File(None),
|
||
font_id: str = Form(""),
|
||
params: str = Form("{}"),
|
||
) -> JobCreateResponse:
|
||
log.info("─" * 50)
|
||
log.info("[API] POST /api/jobs 收到新任务请求")
|
||
log.info(" name_list.filename = %s", name_list.filename if name_list else "无")
|
||
log.info(" wcd_file.filename = %s", wcd_file.filename if wcd_file else "无")
|
||
log.info(" mask_image.filename = %s", mask_image.filename if mask_image else "无")
|
||
log.info(" font_file.filename = %s", font_file.filename if font_file else "无")
|
||
log.info(" font_id = %s", font_id or "(未指定)")
|
||
log.info(" params (raw) = %s", params)
|
||
|
||
try:
|
||
user_params = json.loads(params)
|
||
except json.JSONDecodeError:
|
||
raise HTTPException(status_code=400, detail="params must be valid JSON")
|
||
|
||
if not isinstance(user_params, dict):
|
||
raise HTTPException(status_code=400, detail="params must be JSON object")
|
||
|
||
mode = str(user_params.get("MODE", "IMAGE")).upper()
|
||
|
||
# WCD 生产任务：传入 .wcd 画布导入导出包（还原设计 -> 生产，见 docs/wordcloud-contract.md v1.1）
|
||
is_wcd = mode == "WCD" or bool(wcd_file and wcd_file.filename)
|
||
if is_wcd:
|
||
if not (wcd_file and wcd_file.filename):
|
||
raise HTTPException(status_code=400, detail="wcd_file is required when MODE=WCD")
|
||
return await _create_wcd_job(wcd_file, user_params)
|
||
|
||
# 名单/xlsx 模式（既有逻辑；必填校验移到 mode/WCD 判断之后）
|
||
if not name_list or not name_list.filename:
|
||
raise HTTPException(status_code=400, detail="name_list is required")
|
||
|
||
ext_xlsx = Path(name_list.filename).suffix.lower()
|
||
if ext_xlsx not in {".xlsx"}:
|
||
raise HTTPException(status_code=400, detail="name_list must be xlsx")
|
||
|
||
log.info(" 解析后 params = %s", json.dumps(user_params, ensure_ascii=False))
|
||
|
||
mode = str(user_params.get("MODE", "IMAGE")).upper()
|
||
if mode not in {"IMAGE", "TEXT"}:
|
||
raise HTTPException(status_code=400, detail="MODE must be IMAGE or TEXT")
|
||
|
||
if mode == "IMAGE":
|
||
if not mask_image or not mask_image.filename:
|
||
raise HTTPException(status_code=400, detail="mask_image is required when MODE=IMAGE")
|
||
ext = Path(mask_image.filename).suffix.lower()
|
||
if ext not in {".png", ".jpg", ".jpeg"}:
|
||
raise HTTPException(status_code=400, detail="mask_image must be png/jpg/jpeg")
|
||
elif mask_image and mask_image.filename:
|
||
ext = Path(mask_image.filename).suffix.lower()
|
||
if ext not in {".png", ".jpg", ".jpeg"}:
|
||
raise HTTPException(status_code=400, detail="mask_image must be png/jpg/jpeg")
|
||
|
||
job_id = manager.create_job()
|
||
paths = storage.prepare_job_dirs(job_id)
|
||
log.info("[Job] 创建任务 job_id=%s", job_id)
|
||
log.info(" 工作目录 = %s", paths.output_dir)
|
||
|
||
if mask_image and mask_image.filename:
|
||
mask_bytes = await mask_image.read()
|
||
paths.mask_path.write_bytes(mask_bytes)
|
||
log.info(" 掩膜已保存: %s (%d bytes)", paths.mask_path, len(mask_bytes))
|
||
xlsx_bytes = await name_list.read()
|
||
paths.excel_path.write_bytes(xlsx_bytes)
|
||
log.info(" Excel 已保存: %s (%d bytes)", paths.excel_path, len(xlsx_bytes))
|
||
|
||
# 保存自定义字体（优先 font_id，其次 font_file 上传）
|
||
font_path = ""
|
||
if font_id:
|
||
try:
|
||
font_path = str(_resolve_font_file(font_id))
|
||
log.info(" 使用已保存字体: font_id=%s -> %s", font_id, font_path)
|
||
except FileNotFoundError:
|
||
log.warning(" font_id=%s 不存在，回退默认", font_id)
|
||
elif font_file and font_file.filename:
|
||
font_ext = Path(font_file.filename).suffix.lower()
|
||
if font_ext in _FONT_EXTENSIONS:
|
||
font_dest = paths.input_dir / f"custom_font{font_ext}"
|
||
font_bytes = await font_file.read()
|
||
font_dest.write_bytes(font_bytes)
|
||
font_path = str(font_dest)
|
||
log.info(" 临时字体已保存: %s (%d bytes)", font_dest, len(font_bytes))
|
||
else:
|
||
log.warning(" 不支持的字体格式: %s，忽略", font_ext)
|
||
|
||
config = {
|
||
"MODE": mode,
|
||
"EXCEL_PATH": str(paths.excel_path),
|
||
"OUTPUT_DIR": str(paths.output_dir),
|
||
# Debug masks are useful during local diagnosis but add extra image
|
||
# writes to every request. Keep the fast service path disk-light; the
|
||
# explicit CLI/config option remains available when debugging.
|
||
"SAVE_DEBUG_IMAGES": False,
|
||
"DEBUG_OUTPUT_DIR": str(paths.output_dir / "debug"),
|
||
"FAST_MODE": True,
|
||
}
|
||
if mask_image and mask_image.filename:
|
||
config["MASK_IMAGE_PATH"] = str(paths.mask_path)
|
||
if font_path:
|
||
config["WC_FONT_PATH"] = font_path
|
||
config["MASK_FONT_PATH"] = font_path
|
||
config.update(user_params)
|
||
log.info(" 最终配置 = %s", json.dumps(config, ensure_ascii=False, indent=2))
|
||
|
||
def _run_job_safe() -> None:
|
||
try:
|
||
runner.run(job_id, paths, config)
|
||
except Exception as exc:
|
||
logging.exception("job runner crashed", extra={"job_id": job_id})
|
||
manager.add_event(
|
||
job_id,
|
||
kind="status",
|
||
stage="failed",
|
||
progress_percent=100,
|
||
message=f"任务异常终止: {exc}",
|
||
)
|
||
manager.set_status(
|
||
job_id,
|
||
status="failed",
|
||
stage="failed",
|
||
progress_percent=100,
|
||
message="任务失败",
|
||
error=str(exc),
|
||
)
|
||
|
||
t = threading.Thread(target=_run_job_safe, daemon=True)
|
||
t.start()
|
||
log.info("[Job] 后台线程已启动 job_id=%s thread=%s", job_id, t.name)
|
||
|
||
return JobCreateResponse(job_id=job_id)
|
||
|
||
|
||
def _parse_hex(color: object):
|
||
"""把 '#RRGGBB' / '#RGB' / 空值 解析为 RGBA tuple。"""
|
||
raw = str(color or "#ffffff").strip().lstrip("#")
|
||
if len(raw) == 3:
|
||
raw = "".join(c + c for c in raw)
|
||
try:
|
||
return tuple(int(raw[i : i + 2], 16) for i in (0, 2, 4)) + (255,)
|
||
except ValueError:
|
||
return (255, 255, 255, 255)
|
||
|
||
|
||
def _compose_design_png(document: dict, file_map: dict, output_path: Path) -> None:
|
||
"""把 CanvasDocument 合成一张扁平 PNG（生产任务产物）：背景 + 按 zIndex 叠贴纸。
|
||
|
||
file_map: {pkg_asset_id: {"path": str}}，即 WCD 内临时 assetId -> 落盘素材文件。
|
||
"""
|
||
try:
|
||
width = int(document.get("width") or 1200)
|
||
height = int(document.get("height") or 1200)
|
||
except (TypeError, ValueError):
|
||
width, height = 1200, 1200
|
||
canvas = Image.new("RGBA", (max(width, 1), max(height, 1)), _parse_hex(document.get("background")))
|
||
elements = [
|
||
e
|
||
for e in document.get("elements", [])
|
||
if isinstance(e, dict) and e.get("type") == "sticker"
|
||
]
|
||
elements.sort(key=lambda e: e.get("zIndex", 0))
|
||
for e in elements:
|
||
info = file_map.get(str(e.get("assetId") or ""))
|
||
if not info:
|
||
continue
|
||
path = info.get("path")
|
||
if not path or not Path(path).exists():
|
||
continue
|
||
try:
|
||
img = Image.open(path).convert("RGBA")
|
||
except Exception:
|
||
continue
|
||
ew = e.get("width")
|
||
eh = e.get("height")
|
||
ew = int(ew) if isinstance(ew, (int, float)) and ew > 0 else img.width
|
||
eh = int(eh) if isinstance(eh, (int, float)) and eh > 0 else img.height
|
||
if (ew, eh) != (img.width, img.height):
|
||
img = img.resize((int(ew), int(eh)))
|
||
opacity = e.get("opacity", 1)
|
||
if isinstance(opacity, (int, float)) and opacity >= 0 and opacity != 1:
|
||
img = img.copy()
|
||
img.putalpha(img.getchannel("A").point(lambda v: round(v * float(opacity))))
|
||
canvas.alpha_composite(img, (int(e.get("x") or 0), int(e.get("y") or 0)))
|
||
canvas.convert("RGB").save(output_path, "PNG")
|
||
|
||
|
||
async def _create_wcd_job(wcd_file: UploadFile, user_params: dict) -> JobCreateResponse:
|
||
"""WCD 生产任务：校验并还原 .wcd，落库设计，合成生产 PNG 作为任务产物。
|
||
|
||
与 POST /api/design-templates/import 共用素材去重/注册/重映射逻辑；
|
||
状态机与产物管线复用 jobs（queued/running/success/failed）。
|
||
"""
|
||
if Path(wcd_file.filename).suffix.lower() != ".wcd":
|
||
raise HTTPException(status_code=400, detail="wcd_file must be .wcd")
|
||
content = await wcd_file.read()
|
||
if not content:
|
||
raise HTTPException(status_code=400, detail="wcd_file is empty")
|
||
try:
|
||
zf = zipfile.ZipFile(io.BytesIO(content))
|
||
except zipfile.BadZipFile as exc:
|
||
raise HTTPException(status_code=400, detail="file is not a valid zip/wcd package") from exc
|
||
|
||
with zf:
|
||
manifest = _read_import_json(zf, "manifest.json")
|
||
order_meta = manifest.get("meta") or {}
|
||
order_no = str(order_meta.get("orderNo") or "") if isinstance(order_meta, dict) else ""
|
||
# 兼容旧命名 order-{orderNo}
|
||
if not order_no and str(manifest.get("name") or "").startswith("order-"):
|
||
order_no = str(manifest.get("name"))[6:]
|
||
if manifest.get("format") != "wordcloud-canvas":
|
||
raise HTTPException(status_code=400, detail="不是 wordcloud-canvas 格式")
|
||
try:
|
||
version = int(manifest.get("version", 1))
|
||
except (TypeError, ValueError) as exc:
|
||
raise HTTPException(status_code=400, detail="manifest version 不是合法数字") from exc
|
||
if version != 1:
|
||
raise HTTPException(status_code=400, detail=f"不支持的包版本: {version}")
|
||
|
||
document = _read_import_json(zf, "document.json")
|
||
asset_items = manifest.get("assets")
|
||
if not isinstance(asset_items, list):
|
||
asset_items = []
|
||
|
||
remap: dict[str, str] = {}
|
||
file_map: dict[str, dict] = {}
|
||
reference_ids: list[str] = []
|
||
for item in asset_items:
|
||
pkg_id = str(item.get("id") or "")
|
||
if not pkg_id:
|
||
continue
|
||
entry_name = _zip_asset_entry(zf, pkg_id)
|
||
mime = _imported_asset_mime(item)
|
||
asset_bytes = zf.read(entry_name)
|
||
meta = _register_import_asset(str(item.get("name") or pkg_id), asset_bytes, mime)
|
||
real_id = str(meta["asset_id"])
|
||
remap[pkg_id] = real_id
|
||
file_map[real_id] = {
|
||
"path": str(_asset_dir(real_id) / f"asset{_asset_extension_for_mime(mime)}"),
|
||
"mime": mime,
|
||
}
|
||
reference_ids.append(real_id)
|
||
_remap_import_asset_ids(document, remap)
|
||
|
||
# 落库生产设计（与模板导入一致，便于追溯/复用/在画布中继续编辑）
|
||
display_name = str(manifest.get("name") or "导入生产设计").strip()[:120] or "导入生产设计"
|
||
template_id = f"tmpl_{uuid.uuid4().hex}"
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
_write_design_template(_design_template_dir(template_id), {
|
||
"template_id": template_id,
|
||
"name": display_name,
|
||
"description": str(manifest.get("description") or "下单派单生产设计"),
|
||
"document": document,
|
||
"reference_asset_ids": reference_ids,
|
||
"cover_asset_id": reference_ids[0] if reference_ids else "",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
})
|
||
|
||
job_id = manager.create_job()
|
||
paths = storage.prepare_job_dirs(job_id)
|
||
(paths.input_dir / f"{template_id}.wcd").write_bytes(content)
|
||
log.info("[WCD] 生产任务 job_id=%s template=%s assets=%d", job_id, template_id, len(reference_ids))
|
||
|
||
# 登记生产订单（供 wordcloud 侧订单列表查看）
|
||
if order_no:
|
||
_write_order(_order_dir(order_no), {
|
||
"order_id": order_no,
|
||
"order_no": order_no,
|
||
"job_id": job_id,
|
||
"template_id": template_id,
|
||
"status": "running",
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
})
|
||
log.info("[Order] 登记生产订单 %s → job %s", order_no, job_id)
|
||
manager.set_status(
|
||
job_id, status="running", stage="composing", progress_percent=10, message="WCD 生产任务"
|
||
)
|
||
|
||
def _run_wcd_safe() -> None:
|
||
try:
|
||
png_path = paths.output_dir / "result.png"
|
||
_compose_design_png(document, file_map, png_path)
|
||
manager.set_artifacts(job_id, {"png": str(png_path)})
|
||
manager.set_status(
|
||
job_id, status="success", stage="done", progress_percent=100, message="生产完成"
|
||
)
|
||
except Exception as exc:
|
||
logging.exception("wcd job crashed", extra={"job_id": job_id})
|
||
manager.set_status(
|
||
job_id,
|
||
status="failed",
|
||
stage="failed",
|
||
progress_percent=100,
|
||
message="任务失败",
|
||
error=str(exc),
|
||
)
|
||
|
||
threading.Thread(target=_run_wcd_safe, daemon=True).start()
|
||
return JobCreateResponse(job_id=job_id)
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}", response_model=JobStatus)
|
||
def get_job(job_id: str) -> JobStatus:
|
||
if not manager.exists(job_id):
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
return manager.get_status(job_id)
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/detail", response_model=JobDetail)
|
||
def get_job_detail(job_id: str) -> JobDetail:
|
||
if not manager.exists(job_id):
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
return manager.get_detail(job_id)
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/events")
|
||
def stream_events(job_id: str):
|
||
if not manager.exists(job_id):
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
|
||
q = manager.subscribe(job_id)
|
||
|
||
def gen():
|
||
try:
|
||
while True:
|
||
event = q.get()
|
||
payload = json.dumps(event.model_dump(), ensure_ascii=False, default=str)
|
||
yield f"data: {payload}\n\n"
|
||
if event.stage in {"completed", "failed"}:
|
||
break
|
||
finally:
|
||
manager.unsubscribe(job_id, q)
|
||
|
||
return StreamingResponse(gen(), media_type="text/event-stream")
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/result", response_model=JobResult)
|
||
def get_result(job_id: str) -> JobResult:
|
||
status = _resolve_job_status(job_id)
|
||
if status is None:
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
|
||
def u(kind: str) -> str:
|
||
p = status.artifacts.get(kind, "")
|
||
if not p:
|
||
return ""
|
||
return f"/api/jobs/{job_id}/files/{kind}"
|
||
|
||
return JobResult(
|
||
job_id=job_id,
|
||
status=status.status,
|
||
image_url=u("png"),
|
||
svg_url=u("svg"),
|
||
svg_stroke_url=u("svg_stroke"),
|
||
db_url=u("db"),
|
||
metrics_url=u("metrics"),
|
||
elapsed_seconds=status.elapsed_seconds,
|
||
)
|
||
|
||
|
||
def _normalize_orientation(value: object) -> str:
|
||
if value is None:
|
||
return "horizontal"
|
||
text = str(value).strip().lower()
|
||
if text in {"vertical", "1", "90", "rotate_90", "rotate90"}:
|
||
return "vertical"
|
||
return "horizontal"
|
||
|
||
|
||
def _scan_disk_job_status(job_id: str) -> JobStatus | None:
|
||
"""任务未在 manager 登记时，从输出目录合成完整状态。
|
||
|
||
与 list_jobs 的磁盘扫描同一套产物约定（见 runner.py 产物扫描），
|
||
让磁盘任务在详情接口（/find /locations /files/{kind}）也能正常工作，
|
||
不再因不在内存而 404。找不到产物目录或 db 时返回 None。
|
||
"""
|
||
output_dir = storage.base_dir / job_id / "output"
|
||
if not output_dir.is_dir():
|
||
return None
|
||
|
||
png = next(output_dir.glob("*.png"), None)
|
||
svg = next(
|
||
(p for p in sorted(output_dir.glob("*.svg")) if not p.name.endswith("_stroke.svg")),
|
||
None,
|
||
)
|
||
svg_stroke = next(output_dir.glob("*_stroke.svg"), None)
|
||
# NOTE: 不要用 "*[!_stroke].svg" 形式的字符类去做 svg 排除（见 runner.py 注释）。
|
||
db = next(output_dir.glob("*.db"), None)
|
||
metrics = next(output_dir.glob("*metrics*.json"), None)
|
||
if db is None:
|
||
return None
|
||
|
||
mtime = datetime.fromtimestamp(output_dir.stat().st_mtime, tz=timezone.utc)
|
||
return JobStatus(
|
||
job_id=job_id,
|
||
status="success",
|
||
stage="completed",
|
||
progress_percent=100,
|
||
message="未登记任务（磁盘扫描）",
|
||
created_at=mtime,
|
||
updated_at=mtime,
|
||
artifacts={
|
||
"png": str(png) if png else "",
|
||
"svg": str(svg) if svg else "",
|
||
"svg_stroke": str(svg_stroke) if svg_stroke else "",
|
||
"db": str(db) if db else "",
|
||
"metrics": str(metrics) if metrics else "",
|
||
},
|
||
error="",
|
||
)
|
||
|
||
|
||
def _resolve_job_status(job_id: str) -> JobStatus | None:
|
||
"""优先内存，缺则磁盘回落；两项都无返回 None。详情接口统一用它取状态。"""
|
||
if manager.exists(job_id):
|
||
return manager.get_status(job_id)
|
||
return _scan_disk_job_status(job_id)
|
||
|
||
|
||
def _load_job_config(job_id: str) -> dict:
|
||
config_path = storage.base_dir / job_id / "config.json"
|
||
if not config_path.exists():
|
||
return {}
|
||
try:
|
||
return json.loads(config_path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError:
|
||
return {}
|
||
|
||
|
||
def _resolve_canvas_size(status: JobStatus) -> tuple[int, int]:
|
||
metrics_path_str = status.artifacts.get("metrics", "")
|
||
if metrics_path_str:
|
||
try:
|
||
metrics_path = Path(metrics_path_str)
|
||
metrics = json.loads(metrics_path.read_text(encoding="utf-8"))
|
||
canvas = metrics.get("canvas_info", {})
|
||
width = int(canvas.get("hd_width", 0))
|
||
height = int(canvas.get("hd_height", 0))
|
||
if width > 0 and height > 0:
|
||
return width, height
|
||
except (json.JSONDecodeError, OSError, TypeError, ValueError):
|
||
pass
|
||
|
||
png_path_str = status.artifacts.get("png", "")
|
||
if png_path_str:
|
||
try:
|
||
with Image.open(Path(png_path_str)) as img:
|
||
return img.size
|
||
except OSError:
|
||
pass
|
||
|
||
return 0, 0
|
||
|
||
|
||
def _resolve_font_path(raw_path: str) -> str:
|
||
path = Path(raw_path)
|
||
if path.is_absolute():
|
||
return str(path)
|
||
return str((PROJECT_ROOT / path).resolve())
|
||
|
||
|
||
def _compute_text_box(
|
||
name: str,
|
||
x: int,
|
||
y: int,
|
||
font_size: int,
|
||
orientation: str,
|
||
font_path: str,
|
||
) -> tuple[int, int, int, int]:
|
||
font = get_cached_font(font_path, max(1, int(font_size)))
|
||
if orientation == "vertical":
|
||
font = ImageFont.TransposedFont(font, orientation=Image.ROTATE_90)
|
||
canvas = Image.new("L", (1, 1), 0)
|
||
draw = ImageDraw.Draw(canvas)
|
||
bbox = draw.textbbox((x, y), name, font=font)
|
||
return bbox[0], bbox[1], bbox[2] - bbox[0], bbox[3] - bbox[1]
|
||
|
||
|
||
def _search_job_locations(job_id: str, name: str, mode: str) -> JobLocationSearchResult:
|
||
"""在单个任务 word_locations 库内按名字查找,供各处路由复用。
|
||
|
||
mode=exact（默认） 精确匹配 name = ?；
|
||
mode=contains 子串匹配 name LIKE %q%（通配符转义）。
|
||
旧库缺 box_* 列时按字体度量回退计算外框。
|
||
"""
|
||
status = _resolve_job_status(job_id)
|
||
if status is None:
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
|
||
db_path_str = status.artifacts.get("db", "")
|
||
if not db_path_str:
|
||
raise HTTPException(status_code=404, detail="db artifact not ready")
|
||
|
||
db_path = Path(db_path_str)
|
||
if not db_path.exists():
|
||
raise HTTPException(status_code=404, detail="db artifact missing on disk")
|
||
|
||
query_name = name.strip()
|
||
match_mode = (mode or "exact").strip().lower()
|
||
if match_mode not in {"exact", "contains"}:
|
||
match_mode = "exact"
|
||
canvas_width, canvas_height = _resolve_canvas_size(status)
|
||
job_config = _load_job_config(job_id)
|
||
font_path = _resolve_font_path(str(job_config.get("WC_FONT_PATH") or wc_config.WC_FONT_PATH))
|
||
|
||
try:
|
||
conn = sqlite3.connect(db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
columns = {row["name"] for row in conn.execute("PRAGMA table_info(word_locations)").fetchall()}
|
||
|
||
select_columns = ["id", "name", "x", "y", "font_size", "color"]
|
||
optional_columns = ["orientation", "box_x", "box_y", "box_width", "box_height"]
|
||
for column in optional_columns:
|
||
if column in columns:
|
||
select_columns.append(column)
|
||
|
||
sql = f"SELECT {', '.join(select_columns)} FROM word_locations"
|
||
params: list[object] = []
|
||
if query_name:
|
||
if match_mode == "contains":
|
||
# LIKE 转义通配符，再用 %query% 做子串匹配
|
||
escaped = query_name.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||
sql += ' WHERE name LIKE ? ESCAPE "\\"'
|
||
params.append(f"%{escaped}%")
|
||
else:
|
||
sql += " WHERE name = ?"
|
||
params.append(query_name)
|
||
sql += " ORDER BY id ASC"
|
||
|
||
rows = conn.execute(sql, params).fetchall()
|
||
matches: list[WordLocation] = []
|
||
has_boxes = {"box_x", "box_y", "box_width", "box_height"}.issubset(columns)
|
||
for row in rows:
|
||
orientation = _normalize_orientation(row["orientation"] if "orientation" in row.keys() else None)
|
||
if has_boxes:
|
||
box_x = int(row["box_x"])
|
||
box_y = int(row["box_y"])
|
||
box_width = int(row["box_width"])
|
||
box_height = int(row["box_height"])
|
||
else:
|
||
box_x, box_y, box_width, box_height = _compute_text_box(
|
||
name=str(row["name"]),
|
||
x=int(row["x"]),
|
||
y=int(row["y"]),
|
||
font_size=int(row["font_size"]),
|
||
orientation=orientation,
|
||
font_path=font_path,
|
||
)
|
||
|
||
matches.append(
|
||
WordLocation(
|
||
id=int(row["id"]),
|
||
name=str(row["name"]),
|
||
x=int(row["x"]),
|
||
y=int(row["y"]),
|
||
font_size=int(row["font_size"]),
|
||
color=str(row["color"] or ""),
|
||
orientation=orientation,
|
||
box_x=box_x,
|
||
box_y=box_y,
|
||
box_width=box_width,
|
||
box_height=box_height,
|
||
)
|
||
)
|
||
except sqlite3.Error as exc:
|
||
raise HTTPException(status_code=500, detail=f"failed to read db: {exc}") from exc
|
||
finally:
|
||
if "conn" in locals():
|
||
conn.close()
|
||
|
||
return JobLocationSearchResult(
|
||
job_id=job_id,
|
||
query=query_name,
|
||
mode=match_mode,
|
||
total=len(matches),
|
||
canvas_width=canvas_width,
|
||
canvas_height=canvas_height,
|
||
matches=matches,
|
||
)
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/locations", response_model=JobLocationSearchResult)
|
||
def search_locations(
|
||
job_id: str,
|
||
name: str = Query("", description="需要查找的名字"),
|
||
mode: str = Query("exact", description="匹配方式：exact=精确（默认，向后兼容）；contains=包含子串"),
|
||
) -> JobLocationSearchResult:
|
||
"""在单个任务内查找名字（未鉴权，向后兼容，供词云工作台 FindPanel 使用）。"""
|
||
return _search_job_locations(job_id, name, mode)
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/find", response_model=JobLocationSearchResult)
|
||
def find_job_locations(
|
||
request: Request,
|
||
job_id: str,
|
||
name: str = Query("", description="需要查找的名字"),
|
||
mode: str = Query("exact", description="匹配方式：exact=精确；contains=包含子串"),
|
||
) -> JobLocationSearchResult:
|
||
"""在单个任务内查找名字（需管理口令，供独立查找页使用）。"""
|
||
_require_orders_auth(request)
|
||
return _search_job_locations(job_id, name, mode)
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/occupancy_mask")
|
||
def get_occupancy_mask(job_id: str):
|
||
"""生成并返回占位遮罩图：每个已放置词语的 bounding box 以实色方块表示。"""
|
||
status = _resolve_job_status(job_id)
|
||
if status is None:
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
|
||
db_path_str = status.artifacts.get("db", "")
|
||
if not db_path_str:
|
||
raise HTTPException(status_code=404, detail="db artifact not ready")
|
||
|
||
db_path = Path(db_path_str)
|
||
if not db_path.exists():
|
||
raise HTTPException(status_code=404, detail="db artifact missing on disk")
|
||
|
||
canvas_width, canvas_height = _resolve_canvas_size(status)
|
||
if canvas_width <= 0 or canvas_height <= 0:
|
||
raise HTTPException(status_code=500, detail="cannot determine canvas size")
|
||
|
||
try:
|
||
conn = sqlite3.connect(db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
columns = {row["name"] for row in conn.execute("PRAGMA table_info(word_locations)").fetchall()}
|
||
has_boxes = {"box_x", "box_y", "box_width", "box_height"}.issubset(columns)
|
||
|
||
if has_boxes:
|
||
rows = conn.execute(
|
||
"SELECT box_x, box_y, box_width, box_height FROM word_locations ORDER BY id ASC"
|
||
).fetchall()
|
||
else:
|
||
rows = conn.execute(
|
||
"SELECT x, y, font_size, orientation FROM word_locations ORDER BY id ASC"
|
||
).fetchall()
|
||
except sqlite3.Error as exc:
|
||
raise HTTPException(status_code=500, detail=f"failed to read db: {exc}") from exc
|
||
finally:
|
||
if "conn" in locals():
|
||
conn.close()
|
||
|
||
img = Image.new("RGB", (canvas_width, canvas_height), color=(255, 255, 255))
|
||
draw = ImageDraw.Draw(img)
|
||
|
||
job_config = _load_job_config(job_id)
|
||
font_path = _resolve_font_path(str(job_config.get("WC_FONT_PATH") or wc_config.WC_FONT_PATH))
|
||
|
||
for row in rows:
|
||
if has_boxes:
|
||
bx = int(row["box_x"])
|
||
by = int(row["box_y"])
|
||
bw = int(row["box_width"])
|
||
bh = int(row["box_height"])
|
||
else:
|
||
orientation = _normalize_orientation(row["orientation"] if "orientation" in row.keys() else None)
|
||
bx, by, bw, bh = _compute_text_box(
|
||
name="",
|
||
x=int(row["x"]),
|
||
y=int(row["y"]),
|
||
font_size=int(row["font_size"]),
|
||
orientation=orientation,
|
||
font_path=font_path,
|
||
)
|
||
if bw > 0 and bh > 0:
|
||
draw.rectangle([bx, by, bx + bw - 1, by + bh - 1], fill=(30, 30, 30))
|
||
|
||
buf = io.BytesIO()
|
||
img.save(buf, format="PNG")
|
||
buf.seek(0)
|
||
return StreamingResponse(buf, media_type="image/png")
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/custom.svg")
|
||
def get_custom_svg(
|
||
job_id: str,
|
||
fill: str = Query("fill", description="填充模式: fill / dot / line / ring"),
|
||
stroke: int = Query(0, description="是否描边: 0 / 1"),
|
||
spacing: int = Query(10, ge=2, le=100, description="点阵间距（fill=dot 时生效）"),
|
||
radius: int = Query(2, ge=1, le=20, description="点阵半径（fill=dot 时生效）"),
|
||
color: str = Query("#000000", description="颜色"),
|
||
line_spacing: int = Query(6, ge=2, le=100, description="线间距（fill=line 时生效）"),
|
||
line_width: float = Query(1, ge=0.5, le=10, description="线粗细（fill=line 时生效）"),
|
||
line_angle: int = Query(0, ge=0, le=359, description="线角度（fill=line 时生效，0=水平）"),
|
||
ring_radius: int = Query(3, ge=1, le=20, description="空心圆半径（fill=ring 时生效）"),
|
||
ring_width: float = Query(1, ge=0.5, le=10, description="空心圆线粗（fill=ring 时生效）"),
|
||
ring_spacing: int = Query(8, ge=2, le=100, description="空心圆间距（fill=ring 时生效）"),
|
||
):
|
||
"""统一 SVG 导出：可组合描边 + fill/dot/line/ring 填充。"""
|
||
from core.layout import OptimizedEfficientWordCloud
|
||
from core import config as wc_config
|
||
|
||
status = _resolve_job_status(job_id)
|
||
if status is None:
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
|
||
db_path_str = status.artifacts.get("db", "")
|
||
if not db_path_str:
|
||
raise HTTPException(status_code=404, detail="db artifact not ready")
|
||
|
||
canvas_width, canvas_height = _resolve_canvas_size(status)
|
||
if canvas_width <= 0 or canvas_height <= 0:
|
||
raise HTTPException(status_code=500, detail="cannot determine canvas size")
|
||
|
||
job_config = _load_job_config(job_id)
|
||
font_path = _resolve_font_path(str(job_config.get("WC_FONT_PATH") or wc_config.WC_FONT_PATH))
|
||
|
||
try:
|
||
conn = sqlite3.connect(db_path_str)
|
||
conn.row_factory = sqlite3.Row
|
||
rows = conn.execute("SELECT name, x, y, font_size, color, orientation FROM word_locations ORDER BY id ASC").fetchall()
|
||
except sqlite3.Error as exc:
|
||
raise HTTPException(status_code=500, detail=f"failed to read db: {exc}") from exc
|
||
finally:
|
||
if "conn" in locals():
|
||
conn.close()
|
||
|
||
layout = []
|
||
for row in rows:
|
||
orient = _normalize_orientation(row["orientation"] if "orientation" in row.keys() else None)
|
||
orientation_flag = Image.ROTATE_90 if orient == "vertical" else None
|
||
layout.append((
|
||
str(row["name"]),
|
||
int(row["font_size"]),
|
||
(int(row["y"]), int(row["x"])),
|
||
orientation_flag,
|
||
str(row["color"] or "#000000"),
|
||
))
|
||
|
||
wc = OptimizedEfficientWordCloud.__new__(OptimizedEfficientWordCloud)
|
||
wc.width = canvas_width
|
||
wc.height = canvas_height
|
||
wc.layout_ = layout
|
||
wc.font_path = font_path
|
||
wc.background_color = "white"
|
||
wc.mode = "RGB"
|
||
|
||
tag = f"{fill}{'_stroke' if stroke else ''}"
|
||
if fill == "dot":
|
||
tag += f"_s{spacing}_r{radius}"
|
||
elif fill == "line":
|
||
tag += f"_ls{line_spacing}_lw{line_width}_la{line_angle}"
|
||
elif fill == "ring":
|
||
tag += f"_r{ring_radius}_w{ring_width}_s{ring_spacing}"
|
||
tmp_path = WORKSPACE_DIR / job_id / "output" / f"custom_{tag}.svg"
|
||
tmp_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
wc.to_svg_custom(
|
||
str(tmp_path),
|
||
fill_mode=fill,
|
||
do_stroke=bool(stroke),
|
||
dot_spacing=spacing,
|
||
dot_radius=radius,
|
||
color=color,
|
||
line_spacing=line_spacing,
|
||
line_width=line_width,
|
||
line_angle=line_angle,
|
||
ring_radius=ring_radius,
|
||
ring_width=ring_width,
|
||
ring_spacing=ring_spacing,
|
||
)
|
||
|
||
return FileResponse(tmp_path, media_type="image/svg+xml", filename=f"wordcloud_{tag}.svg")
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/files/{kind}")
|
||
def get_file(job_id: str, kind: str):
|
||
status = _resolve_job_status(job_id)
|
||
if status is None:
|
||
raise HTTPException(status_code=404, detail="job not found")
|
||
|
||
try:
|
||
path = manager.resolve_artifact_path(status, kind)
|
||
except KeyError:
|
||
raise HTTPException(status_code=404, detail="unknown artifact kind")
|
||
except FileNotFoundError:
|
||
raise HTTPException(status_code=404, detail="artifact not ready")
|
||
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="artifact missing on disk")
|
||
|
||
media = {
|
||
"png": "image/png",
|
||
"svg": "image/svg+xml",
|
||
"db": "application/octet-stream",
|
||
"metrics": "application/json",
|
||
}.get(kind, "application/octet-stream")
|
||
|
||
return FileResponse(path, media_type=media, filename=path.name)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 3.9 生产订单列表（下单派单投递的 WCD 生产任务，需登录）
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
# 生产订单管理后台口令（建议生产通过环境变量 ORDERS_ADMIN_PASSWORD 覆盖）
|
||
ADMIN_PASSWORD = os.environ.get("ORDERS_ADMIN_PASSWORD") or "zhihui2024"
|
||
_ORDER_TOKEN_SECRET = os.environ.get("ORDERS_TOKEN_SECRET") or "wordcloud-orders-demo"
|
||
|
||
|
||
def _orders_token() -> str:
|
||
return hashlib.sha256((ADMIN_PASSWORD + ":" + _ORDER_TOKEN_SECRET).encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _require_orders_auth(request: Request) -> None:
|
||
"""订单相关接口登录校验：Authorization: Bearer <token> 或 ?token= 需等于鉴权 token。"""
|
||
auth = request.headers.get("Authorization", "")
|
||
bearer = auth[7:] if auth.startswith("Bearer ") else ""
|
||
token = bearer or request.query_params.get("token", "")
|
||
if not token or token != _orders_token():
|
||
raise HTTPException(status_code=403, detail="需要登录") # noqa: S105
|
||
|
||
|
||
@app.post("/api/login")
|
||
async def login(request: Request) -> dict:
|
||
try:
|
||
payload = await request.json()
|
||
except Exception:
|
||
payload = {}
|
||
if (payload or {}).get("password") != ADMIN_PASSWORD:
|
||
raise HTTPException(status_code=401, detail="口令错误")
|
||
return {"token": _orders_token()}
|
||
|
||
|
||
@app.get("/api/orders")
|
||
def list_orders(request: Request) -> list[dict]:
|
||
"""生产订单列表：来自小程序下单派单投递到 wordcloud 的 WCD 生产任务（需登录）。"""
|
||
_require_orders_auth(request)
|
||
orders = []
|
||
if ORDERS_DIR.exists():
|
||
for item in list(ORDERS_DIR.iterdir()):
|
||
if not item.is_dir() or not (item / "order.json").exists():
|
||
continue
|
||
o = _read_order(item)
|
||
if not o:
|
||
continue
|
||
# 用 job 的最新状态回填
|
||
try:
|
||
st = manager.get_status(str(o.get("job_id") or ""))
|
||
o["status"] = st.status
|
||
except Exception:
|
||
pass
|
||
orders.append(o)
|
||
return list(reversed(orders))
|
||
|
||
|
||
@app.get("/api/orders/export")
|
||
async def export_orders(request: Request) -> StreamingResponse:
|
||
"""一键导出多层 ZIP：每个订单一个嵌套文件夹（order.json + design + result）。"""
|
||
_require_orders_auth(request)
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
dirs = [d for d in ORDERS_DIR.iterdir() if d.is_dir() and (d / "order.json").exists()]
|
||
for item in sorted(dirs):
|
||
o = _read_order(item)
|
||
if not o:
|
||
continue
|
||
order_no = str(o.get("order_no") or "order")
|
||
folder = f"orders/{order_no}/"
|
||
zf.writestr(folder + "order.json", json.dumps(o, ensure_ascii=False, indent=2))
|
||
tid = o.get("template_id")
|
||
job_id = o.get("job_id")
|
||
if tid:
|
||
tjson = _design_template_dir(tid) / "template.json"
|
||
if tjson.exists():
|
||
zf.write(str(tjson), folder + "design/template.json")
|
||
if job_id:
|
||
try:
|
||
st = manager.get_status(str(job_id))
|
||
png = manager.resolve_artifact_path(st, "png")
|
||
zf.write(str(png), folder + "result/result.png")
|
||
except Exception:
|
||
pass
|
||
buf.seek(0)
|
||
return StreamingResponse(
|
||
buf,
|
||
media_type="application/zip",
|
||
headers={"Content-Disposition": 'attachment; filename="orders-export.zip"'},
|
||
)
|
||
|
||
|
||
@app.get("/api/orders/{order_no}")
|
||
def get_order(request: Request, order_no: str) -> dict:
|
||
_require_orders_auth(request)
|
||
path = Path(_order_dir(order_no)) / "order.json"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="order not found")
|
||
o = json.loads(path.read_text(encoding="utf-8"))
|
||
try:
|
||
st = manager.get_status(str(o.get("job_id") or ""))
|
||
o["status"] = st.status
|
||
except Exception:
|
||
pass
|
||
return o
|
||
|
||
|
||
@app.get("/api/orders/{order_no}/design")
|
||
def order_design(request: Request, order_no: str) -> dict:
|
||
"""订单设计详情：完整 CanvasDocument + 素材映射，供前端分层/整体预览与分层 SVG 导出。"""
|
||
_require_orders_auth(request)
|
||
path = Path(_order_dir(order_no)) / "order.json"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="order not found")
|
||
o = json.loads(path.read_text(encoding="utf-8"))
|
||
tid = o.get("template_id")
|
||
tjson = _design_template_dir(tid) / "template.json" if tid else None
|
||
if not tjson or not tjson.exists():
|
||
raise HTTPException(status_code=404, detail="订单设计不存在")
|
||
data = json.loads(tjson.read_text(encoding="utf-8"))
|
||
doc = data.get("document") or {}
|
||
|
||
# 素材映射：元素 assetId -> {id,type(source 是 svg 还是 image),source(下载 URL)}
|
||
assets: dict[str, dict] = {}
|
||
for e in doc.get("elements") or []:
|
||
if not isinstance(e, dict):
|
||
continue
|
||
aid = e.get("assetId")
|
||
if not aid or str(aid) in assets:
|
||
continue
|
||
meta = _read_asset_meta(_asset_dir(str(aid))) or {}
|
||
mime = str(meta.get("mime_type") or "")
|
||
assets[str(aid)] = {
|
||
"id": str(aid),
|
||
"type": "svg" if mime in ("image/svg+xml",) else "image",
|
||
"source": meta.get("file_url") or f"/api/assets/{aid}/download",
|
||
}
|
||
|
||
try:
|
||
status = manager.get_status(str(o.get("job_id") or "")).status
|
||
except Exception:
|
||
status = o.get("status", "unknown")
|
||
|
||
return {
|
||
"order_no": order_no,
|
||
"status": status,
|
||
"job_id": o.get("job_id"),
|
||
"template_id": tid,
|
||
"document": {
|
||
"width": doc.get("width"),
|
||
"height": doc.get("height"),
|
||
"background": doc.get("background", "#ffffff"),
|
||
"layers": doc.get("layers") or [],
|
||
"layerFolders": doc.get("layerFolders") or [],
|
||
"elements": doc.get("elements") or [],
|
||
},
|
||
"assets": assets,
|
||
}
|
||
|
||
|
||
@app.get("/api/orders/{order_no}/export")
|
||
async def export_order(request: Request, order_no: str) -> StreamingResponse:
|
||
"""单订单导出多层 ZIP。"""
|
||
_require_orders_auth(request)
|
||
path = Path(_order_dir(order_no)) / "order.json"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="order not found")
|
||
o = json.loads(path.read_text(encoding="utf-8"))
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
folder = f"orders/{order_no}/"
|
||
zf.writestr(folder + "order.json", json.dumps(o, ensure_ascii=False, indent=2))
|
||
tid = o.get("template_id")
|
||
job_id = o.get("job_id")
|
||
if tid:
|
||
tjson = _design_template_dir(tid) / "template.json"
|
||
if tjson.exists():
|
||
zf.write(str(tjson), folder + "design/template.json")
|
||
if job_id:
|
||
try:
|
||
st = manager.get_status(str(job_id))
|
||
png = manager.resolve_artifact_path(st, "png")
|
||
zf.write(str(png), folder + "result/result.png")
|
||
except Exception:
|
||
pass
|
||
buf.seek(0)
|
||
return StreamingResponse(
|
||
buf,
|
||
media_type="application/zip",
|
||
headers={"Content-Disposition": f'attachment; filename="order-{order_no}.zip"'},
|
||
)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 4. 模板接口
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
@app.get("/api/templates", response_model=list[Template])
|
||
def list_templates() -> list[Template]:
|
||
return _TEMPLATES
|
||
|
||
|
||
def _read_design_template(template_dir: Path) -> dict:
|
||
path = template_dir / "template.json"
|
||
if not path.exists():
|
||
return {}
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _write_design_template(template_dir: Path, data: dict) -> None:
|
||
template_dir.mkdir(parents=True, exist_ok=True)
|
||
(template_dir / "template.json").write_text(
|
||
json.dumps(data, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
# ── 生产订单存储（下单派单投递的 WCD 生产任务）────────────
|
||
def _order_dir(order_no: str) -> Path:
|
||
return ORDERS_DIR / order_no
|
||
|
||
|
||
def _read_order(order_dir: Path) -> dict:
|
||
path = order_dir / "order.json"
|
||
if not path.exists():
|
||
return {}
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _write_order(order_dir: Path, data: dict) -> None:
|
||
order_dir.mkdir(parents=True, exist_ok=True)
|
||
(order_dir / "order.json").write_text(
|
||
json.dumps(data, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def _parse_json_array(value: str, field_name: str) -> list:
|
||
try:
|
||
parsed = json.loads(value or "[]")
|
||
except json.JSONDecodeError:
|
||
raise HTTPException(status_code=400, detail=f"{field_name} must be valid JSON array")
|
||
if not isinstance(parsed, list):
|
||
raise HTTPException(status_code=400, detail=f"{field_name} must be a JSON array")
|
||
return parsed
|
||
|
||
|
||
@app.get("/api/design-templates", response_model=list[DesignTemplate])
|
||
def list_design_templates() -> list[DesignTemplate]:
|
||
items: list[DesignTemplate] = []
|
||
for d in _list_dirs(DESIGN_TEMPLATES_DIR):
|
||
try:
|
||
data = _read_design_template(d)
|
||
if data:
|
||
items.append(DesignTemplate(**data))
|
||
except Exception:
|
||
continue
|
||
return sorted(items, key=lambda item: item.updated_at, reverse=True)
|
||
|
||
|
||
@app.post("/api/design-templates", response_model=DesignTemplate)
|
||
async def create_design_template(
|
||
name: str = Form(...),
|
||
description: str = Form(""),
|
||
document: str = Form(...),
|
||
reference_asset_ids: str = Form("[]"),
|
||
cover_asset_id: str = Form(""),
|
||
) -> DesignTemplate:
|
||
display_name = name.strip()
|
||
if not display_name:
|
||
raise HTTPException(status_code=400, detail="name is required")
|
||
try:
|
||
document_data = json.loads(document)
|
||
except json.JSONDecodeError:
|
||
raise HTTPException(status_code=400, detail="document must be valid JSON")
|
||
if not isinstance(document_data, dict):
|
||
raise HTTPException(status_code=400, detail="document must be a JSON object")
|
||
|
||
refs = [str(item) for item in _parse_json_array(reference_asset_ids, "reference_asset_ids")]
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
template_id = f"tmpl_{uuid.uuid4().hex}"
|
||
data = {
|
||
"template_id": template_id,
|
||
"name": display_name[:120],
|
||
"description": description.strip(),
|
||
"document": document_data,
|
||
"reference_asset_ids": refs,
|
||
"cover_asset_id": cover_asset_id.strip() or (refs[0] if refs else ""),
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
_write_design_template(_design_template_dir(template_id), data)
|
||
return DesignTemplate(**data)
|
||
|
||
|
||
def _read_import_json(zf: zipfile.ZipFile, name: str) -> dict:
|
||
try:
|
||
with zf.open(name) as fh:
|
||
raw = fh.read().decode("utf-8")
|
||
except KeyError as exc:
|
||
raise HTTPException(status_code=400, detail=f"{name} 缺失") from exc
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
raise HTTPException(status_code=400, detail=f"{name} 不是合法 JSON") from exc
|
||
if not isinstance(data, dict):
|
||
raise HTTPException(status_code=400, detail=f"{name} 必须是 JSON 对象")
|
||
return data
|
||
|
||
|
||
def _imported_asset_mime(item: dict) -> str:
|
||
mime = str(item.get("mimeType") or item.get("mime_type") or "").strip().lower()
|
||
if mime in {"image/svg+xml", "image/png", "image/jpeg"}:
|
||
return mime
|
||
asset_type = str(item.get("type") or "").strip().lower()
|
||
if asset_type in {"svg", "image/svg+xml"}:
|
||
return "image/svg+xml"
|
||
if asset_type in {"image", "png"}:
|
||
return "image/png"
|
||
return "image/svg+xml" if str(item.get("id", "")).endswith(".svg") else "image/png"
|
||
|
||
|
||
def _asset_meta_items() -> list[dict]:
|
||
result: list[dict] = []
|
||
for d in _list_dirs(ASSETS_DIR):
|
||
meta = _read_asset_meta(d)
|
||
if meta:
|
||
result.append(meta)
|
||
return result
|
||
|
||
|
||
def _find_asset_by_sha256(digest: str) -> dict | None:
|
||
for meta in _asset_meta_items():
|
||
try:
|
||
ext = _asset_extension_for_mime(str(meta.get("mime_type") or ""))
|
||
path = _asset_dir(str(meta["asset_id"])) / f"asset{ext}"
|
||
if not path.exists():
|
||
continue
|
||
actual_digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
||
if actual_digest == digest:
|
||
meta["sha256"] = actual_digest
|
||
_write_asset_meta(_asset_dir(str(meta["asset_id"])), meta)
|
||
return meta
|
||
except Exception:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _register_import_asset(name: str, content: bytes, mime: str) -> dict:
|
||
digest = hashlib.sha256(content).hexdigest()
|
||
existing = _find_asset_by_sha256(digest)
|
||
if existing and existing.get("type") == "sticker":
|
||
return existing
|
||
|
||
asset_id = f"asset_{uuid.uuid4().hex}"
|
||
asset_path = _asset_dir(asset_id)
|
||
asset_path.mkdir(parents=True, exist_ok=True)
|
||
ext = _asset_extension_for_mime(mime)
|
||
dest = asset_path / f"asset{ext}"
|
||
dest.write_bytes(content)
|
||
|
||
width, height = 0, 0
|
||
if mime == "image/svg+xml":
|
||
width, height = _parse_svg_viewbox(dest)
|
||
else:
|
||
try:
|
||
with Image.open(dest) as img:
|
||
width, height = img.size
|
||
except Exception:
|
||
pass
|
||
|
||
meta = {
|
||
"asset_id": asset_id,
|
||
"name": name[:120] or "导入素材",
|
||
"type": "sticker",
|
||
"mime_type": mime,
|
||
"width": width,
|
||
"height": height,
|
||
"file_size": len(content),
|
||
"sha256": digest,
|
||
"file_url": f"/api/assets/{asset_id}/download",
|
||
"job_id": "",
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
_write_asset_meta(asset_path, meta)
|
||
return meta
|
||
|
||
|
||
def _zip_asset_entry(zf: zipfile.ZipFile, pkg_id: str) -> str:
|
||
candidates: list[str] = []
|
||
for entry in zf.namelist():
|
||
parts = entry.split("/")
|
||
if (
|
||
len(parts) >= 2
|
||
and parts[0] == "assets"
|
||
and parts[1].startswith(pkg_id)
|
||
and ".." not in parts
|
||
and not entry.endswith("/")
|
||
):
|
||
candidates.append(entry)
|
||
if not candidates:
|
||
raise HTTPException(status_code=400, detail=f"包内缺少素材: {pkg_id}")
|
||
for entry in candidates:
|
||
if Path(entry).stem == pkg_id:
|
||
return entry
|
||
return sorted(candidates)[0]
|
||
|
||
|
||
def _remap_import_asset_ids(document: dict, asset_map: dict[str, str]) -> None:
|
||
elements = document.get("elements")
|
||
if not isinstance(elements, list):
|
||
return
|
||
for element in elements:
|
||
if isinstance(element, dict) and element.get("type") == "sticker":
|
||
old_id = element.get("assetId")
|
||
if isinstance(old_id, str) and old_id in asset_map:
|
||
element["assetId"] = asset_map[old_id]
|
||
|
||
|
||
@app.post("/api/design-templates/import", response_model=DesignTemplate)
|
||
async def import_design_template(
|
||
file: UploadFile = File(...),
|
||
name: str = Form(""),
|
||
description: str = Form(""),
|
||
) -> DesignTemplate:
|
||
if not file.filename:
|
||
raise HTTPException(status_code=400, detail="file is required")
|
||
if Path(file.filename).suffix.lower() != ".wcd":
|
||
raise HTTPException(status_code=400, detail="file must be .wcd")
|
||
|
||
content = await file.read()
|
||
if not content:
|
||
raise HTTPException(status_code=400, detail="file is empty")
|
||
|
||
try:
|
||
zf = zipfile.ZipFile(io.BytesIO(content))
|
||
except zipfile.BadZipFile as exc:
|
||
raise HTTPException(status_code=400, detail="file is not a valid zip/wcd package") from exc
|
||
|
||
with zf:
|
||
manifest = _read_import_json(zf, "manifest.json")
|
||
if manifest.get("format") != "wordcloud-canvas":
|
||
raise HTTPException(status_code=400, detail="不是 wordcloud-canvas 格式")
|
||
try:
|
||
version = int(manifest.get("version", 1))
|
||
except (TypeError, ValueError) as exc:
|
||
raise HTTPException(status_code=400, detail="manifest version 不是合法数字") from exc
|
||
if version != 1:
|
||
raise HTTPException(status_code=400, detail=f"不支持的包版本: {version}")
|
||
|
||
document = _read_import_json(zf, "document.json")
|
||
asset_items = manifest.get("assets")
|
||
if not isinstance(asset_items, list):
|
||
asset_items = []
|
||
|
||
asset_map: dict[str, str] = {}
|
||
reference_ids: list[str] = []
|
||
for item in asset_items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
pkg_id = str(item.get("id") or "")
|
||
if not pkg_id:
|
||
continue
|
||
entry_name = _zip_asset_entry(zf, pkg_id)
|
||
mime = _imported_asset_mime(item)
|
||
asset_bytes = zf.read(entry_name)
|
||
meta = _register_import_asset(str(item.get("name") or pkg_id), asset_bytes, mime)
|
||
asset_map[pkg_id] = str(meta["asset_id"])
|
||
reference_ids.append(str(meta["asset_id"]))
|
||
|
||
_remap_import_asset_ids(document, asset_map)
|
||
|
||
display_name = name.strip() or str(manifest.get("name") or "导入设计").strip() or "导入设计"
|
||
display_description = description.strip() or str(manifest.get("description") or "").strip()
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
template_id = f"tmpl_{uuid.uuid4().hex}"
|
||
data = {
|
||
"template_id": template_id,
|
||
"name": display_name[:120],
|
||
"description": display_description,
|
||
"document": document,
|
||
"reference_asset_ids": reference_ids,
|
||
"cover_asset_id": reference_ids[0] if reference_ids else "",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
_write_design_template(_design_template_dir(template_id), data)
|
||
return DesignTemplate(**data)
|
||
|
||
|
||
@app.get("/api/design-templates/{template_id}", response_model=DesignTemplate)
|
||
def get_design_template(template_id: str) -> DesignTemplate:
|
||
data = _read_design_template(_design_template_dir(template_id))
|
||
if not data:
|
||
raise HTTPException(status_code=404, detail="design template not found")
|
||
return DesignTemplate(**data)
|
||
|
||
|
||
@app.patch("/api/design-templates/{template_id}", response_model=DesignTemplate)
|
||
async def update_design_template(
|
||
template_id: str,
|
||
name: str = Form(""),
|
||
description: str = Form(""),
|
||
document: str = Form(""),
|
||
reference_asset_ids: str = Form(""),
|
||
cover_asset_id: str = Form(""),
|
||
) -> DesignTemplate:
|
||
template_dir = _design_template_dir(template_id)
|
||
data = _read_design_template(template_dir)
|
||
if not data:
|
||
raise HTTPException(status_code=404, detail="design template not found")
|
||
|
||
if name:
|
||
data["name"] = name.strip()[:120]
|
||
if description != "":
|
||
data["description"] = description.strip()
|
||
if document:
|
||
try:
|
||
document_data = json.loads(document)
|
||
except json.JSONDecodeError:
|
||
raise HTTPException(status_code=400, detail="document must be valid JSON")
|
||
if not isinstance(document_data, dict):
|
||
raise HTTPException(status_code=400, detail="document must be a JSON object")
|
||
data["document"] = document_data
|
||
if reference_asset_ids:
|
||
refs = [str(item) for item in _parse_json_array(reference_asset_ids, "reference_asset_ids")]
|
||
data["reference_asset_ids"] = refs
|
||
if not data.get("cover_asset_id") and refs:
|
||
data["cover_asset_id"] = refs[0]
|
||
if cover_asset_id:
|
||
data["cover_asset_id"] = cover_asset_id.strip()
|
||
|
||
data["updated_at"] = datetime.now(timezone.utc).isoformat()
|
||
_write_design_template(template_dir, data)
|
||
return DesignTemplate(**data)
|
||
|
||
|
||
@app.delete("/api/design-templates/{template_id}", status_code=204, response_model=None)
|
||
def delete_design_template(template_id: str) -> None:
|
||
template_dir = _design_template_dir(template_id)
|
||
if not (template_dir / "template.json").exists():
|
||
raise HTTPException(status_code=404, detail="design template not found")
|
||
shutil.rmtree(template_dir, ignore_errors=True)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 5. 素材接口
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _parse_svg_viewbox(svg_path: Path) -> tuple[int, int]:
|
||
"""尝试从 SVG 文件头提取 width/height 属性。"""
|
||
try:
|
||
text = svg_path.read_text(encoding="utf-8", errors="ignore")
|
||
m = re.search(r'width="(\d+)"', text)
|
||
w = int(m.group(1)) if m else 0
|
||
m = re.search(r'height="(\d+)"', text)
|
||
h = int(m.group(1)) if m else 0
|
||
return w, h
|
||
except Exception:
|
||
return 0, 0
|
||
|
||
|
||
def _asset_extension_for_mime(mime_type: str) -> str:
|
||
if mime_type == "image/svg+xml":
|
||
return ".svg"
|
||
if mime_type == "image/jpeg":
|
||
return ".jpg"
|
||
return ".png"
|
||
|
||
|
||
@app.post("/api/assets", response_model=Asset)
|
||
async def upload_asset(
|
||
file: UploadFile = File(...),
|
||
name: str = Form(""),
|
||
type: str = Form("upload"),
|
||
) -> Asset:
|
||
if not file.filename:
|
||
raise HTTPException(status_code=400, detail="file is required")
|
||
|
||
ext = Path(file.filename).suffix.lower()
|
||
if ext not in {".svg", ".png", ".jpg", ".jpeg"}:
|
||
raise HTTPException(status_code=400, detail="unsupported file type, expected .svg/.png/.jpg/.jpeg")
|
||
|
||
mime = "image/svg+xml" if ext == ".svg" else "image/jpeg" if ext in {".jpg", ".jpeg"} else "image/png"
|
||
asset_id = f"asset_{uuid.uuid4().hex}"
|
||
asset_path = _asset_dir(asset_id)
|
||
asset_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
stored_ext = ".jpg" if ext == ".jpeg" else ext
|
||
dest = asset_path / f"asset{stored_ext}"
|
||
content = await file.read()
|
||
dest.write_bytes(content)
|
||
|
||
width, height = 0, 0
|
||
if ext == ".svg":
|
||
width, height = _parse_svg_viewbox(dest)
|
||
elif ext in {".png", ".jpg", ".jpeg"}:
|
||
with Image.open(dest) as img:
|
||
width, height = img.size
|
||
|
||
meta = {
|
||
"asset_id": asset_id,
|
||
"name": name or file.filename,
|
||
"type": type,
|
||
"mime_type": mime,
|
||
"width": width,
|
||
"height": height,
|
||
"file_size": len(content),
|
||
"file_url": f"/api/assets/{asset_id}/download",
|
||
"job_id": "",
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
_write_asset_meta(asset_path, meta)
|
||
|
||
return Asset(**meta)
|
||
|
||
|
||
|
||
def _svg_with_transparent_background(svg_bytes: bytes) -> bytes:
|
||
"""Convert full-canvas opaque backgrounds to transparent for sticker use.
|
||
|
||
Job pipeline SVG (`to_svg`) embeds a white full-rect so standalone preview
|
||
looks solid. Canvas stickers must stay transparent so the mask layer shows
|
||
through. Only the full-canvas background rect is touched.
|
||
"""
|
||
try:
|
||
text = svg_bytes.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
text = svg_bytes.decode("utf-8", errors="ignore")
|
||
|
||
patterns = [
|
||
# <rect width="100%" height="100%" fill="white"/>
|
||
(
|
||
r'(<rect\b[^>]*\bwidth=["\']100%["\'][^>]*\bheight=["\']100%["\'][^>]*\bfill=["\'])(?:white|#fff(?:fff)?|rgb\(\s*255\s*,\s*255\s*,\s*255\s*\))(["\'][^>]*?/?>)',
|
||
r'\1none\2',
|
||
),
|
||
# attribute order: fill before width/height
|
||
(
|
||
r'(<rect\b[^>]*\bfill=["\'])(?:white|#fff(?:fff)?|rgb\(\s*255\s*,\s*255\s*,\s*255\s*\))(["\'][^>]*\bwidth=["\']100%["\'][^>]*\bheight=["\']100%["\'][^>]*?/?>)',
|
||
r'\1none\2',
|
||
),
|
||
]
|
||
for pattern, repl in patterns:
|
||
updated, n = re.subn(pattern, repl, text, count=1, flags=re.IGNORECASE)
|
||
if n:
|
||
return updated.encode("utf-8")
|
||
return svg_bytes.encode("utf-8") if isinstance(svg_bytes, str) else svg_bytes
|
||
|
||
|
||
def _find_job_svg(job_id: str) -> Path | None:
|
||
"""Locate the main (non-stroke) SVG for a job.
|
||
|
||
Prefer in-memory artifact path when valid; always fall back to the job
|
||
workspace on disk so empty/stale memory state cannot hide existing files.
|
||
"""
|
||
svg_path: Path | None = None
|
||
|
||
if manager.exists(job_id):
|
||
status = manager.get_status(job_id)
|
||
svg_path_str = status.artifacts.get("svg", "")
|
||
if svg_path_str:
|
||
candidate = Path(svg_path_str)
|
||
if candidate.exists():
|
||
svg_path = candidate
|
||
|
||
if svg_path is not None:
|
||
return svg_path
|
||
|
||
fallback_dir = storage.base_dir / job_id / "output"
|
||
if not fallback_dir.exists():
|
||
return None
|
||
|
||
candidates = sorted(fallback_dir.glob("*.svg"))
|
||
for candidate in candidates:
|
||
if candidate.name.endswith("_stroke.svg"):
|
||
continue
|
||
return candidate
|
||
return candidates[0] if candidates else None
|
||
|
||
|
||
@app.post("/api/assets/from-job/{job_id}", response_model=Asset)
|
||
async def import_asset_from_job(
|
||
job_id: str,
|
||
name: str = Form(""),
|
||
type: str = Form("wordcloud"),
|
||
) -> Asset:
|
||
svg_path = _find_job_svg(job_id)
|
||
if svg_path is None or not svg_path.exists():
|
||
raise HTTPException(status_code=404, detail="svg not found")
|
||
|
||
asset_id = f"asset_{uuid.uuid4().hex}"
|
||
asset_path = _asset_dir(asset_id)
|
||
asset_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
dest = asset_path / "asset.svg"
|
||
raw = svg_path.read_bytes()
|
||
# 画布贴纸需要透明底；任务主 SVG 默认带白底 rect
|
||
content = _svg_with_transparent_background(raw)
|
||
dest.write_bytes(content)
|
||
width, height = _parse_svg_viewbox(dest)
|
||
|
||
asset_type = (type or "wordcloud").strip() or "wordcloud"
|
||
display_name = name or f"词云_{job_id[:8]}"
|
||
meta = {
|
||
"asset_id": asset_id,
|
||
"name": display_name,
|
||
"type": asset_type,
|
||
"mime_type": "image/svg+xml",
|
||
"width": width,
|
||
"height": height,
|
||
"file_size": len(content),
|
||
"file_url": f"/api/assets/{asset_id}/download",
|
||
"job_id": job_id,
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
_write_asset_meta(asset_path, meta)
|
||
|
||
return Asset(**meta)
|
||
|
||
|
||
@app.get("/api/assets", response_model=list[Asset])
|
||
def list_assets(
|
||
type: str = Query("", description="过滤类型：wordcloud / upload / shape"),
|
||
) -> list[Asset]:
|
||
items: list[Asset] = []
|
||
for d in _list_dirs(ASSETS_DIR):
|
||
meta = _read_asset_meta(d)
|
||
if not meta:
|
||
continue
|
||
if type and meta.get("type") != type:
|
||
continue
|
||
items.append(Asset(**meta))
|
||
return sorted(items, key=lambda a: a.created_at, reverse=True)
|
||
|
||
|
||
@app.get("/api/assets/{asset_id}", response_model=Asset)
|
||
def get_asset(asset_id: str) -> Asset:
|
||
d = _asset_dir(asset_id)
|
||
meta = _read_asset_meta(d)
|
||
if not meta:
|
||
raise HTTPException(status_code=404, detail="asset not found")
|
||
return Asset(**meta)
|
||
|
||
|
||
@app.get("/api/assets/{asset_id}/download")
|
||
def download_asset(asset_id: str):
|
||
d = _asset_dir(asset_id)
|
||
meta = _read_asset_meta(d)
|
||
if not meta:
|
||
raise HTTPException(status_code=404, detail="asset not found")
|
||
|
||
ext = _asset_extension_for_mime(meta["mime_type"])
|
||
path = d / f"asset{ext}"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="asset file missing on disk")
|
||
|
||
media = meta.get("mime_type", "application/octet-stream")
|
||
return FileResponse(path, media_type=media, filename=f"{meta['name']}{ext}")
|
||
|
||
|
||
@app.post("/api/assets/{asset_id}/line-spacing", response_model=LineSpacingAnalysisSummary)
|
||
def analyze_asset_line_spacing(
|
||
asset_id: str,
|
||
request: LineSpacingAnalysisRequest,
|
||
) -> LineSpacingAnalysisSummary:
|
||
d = _asset_dir(asset_id)
|
||
meta = _read_asset_meta(d)
|
||
if not meta:
|
||
raise HTTPException(status_code=404, detail="asset not found")
|
||
if meta.get("mime_type") != "image/svg+xml":
|
||
raise HTTPException(status_code=400, detail="line spacing analysis only supports SVG assets")
|
||
|
||
path = d / "asset.svg"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="asset file missing on disk")
|
||
|
||
started = time.perf_counter()
|
||
try:
|
||
result = analyze_svg_line_spacing_file(
|
||
path,
|
||
percentile=request.percentile,
|
||
element_width=request.elementWidth,
|
||
element_height=request.elementHeight,
|
||
sample_step=request.sampleStep,
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
elapsed = time.perf_counter() - started
|
||
log.info(
|
||
"[API] line-spacing asset=%s percentile=%s curves=%s segments=%s elapsed=%.2fs",
|
||
asset_id,
|
||
result.percentile,
|
||
result.curveCount,
|
||
result.segmentCount,
|
||
elapsed,
|
||
)
|
||
return LineSpacingAnalysisSummary(**result.as_dict())
|
||
|
||
|
||
@app.delete("/api/assets/{asset_id}", status_code=204, response_model=None)
|
||
def delete_asset(asset_id: str) -> None:
|
||
d = _asset_dir(asset_id)
|
||
meta = _read_asset_meta(d)
|
||
if not meta:
|
||
raise HTTPException(status_code=404, detail="asset not found")
|
||
shutil.rmtree(d, ignore_errors=True)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 6. 工程接口
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
@app.post("/api/projects", response_model=Project)
|
||
async def create_project(
|
||
name: str = Form(...),
|
||
template_id: str = Form(...),
|
||
background_color: str = Form(...),
|
||
stickers: str = Form("[]"),
|
||
) -> Project:
|
||
if not name or len(name) > 120:
|
||
raise HTTPException(status_code=400, detail="name is required and must be <= 120 chars")
|
||
|
||
if not _safe_hex_color(background_color):
|
||
raise HTTPException(status_code=400, detail="background_color must be valid hex like #ffffff")
|
||
|
||
if template_id not in {t.id for t in _TEMPLATES}:
|
||
raise HTTPException(status_code=400, detail=f"template not found: {template_id}")
|
||
|
||
try:
|
||
stickers_data: list[dict] = json.loads(stickers)
|
||
except json.JSONDecodeError:
|
||
raise HTTPException(status_code=400, detail="stickers must be valid JSON array")
|
||
|
||
project_id = f"proj_{uuid.uuid4().hex}"
|
||
pdir = _project_dir(project_id)
|
||
pdir.mkdir(parents=True, exist_ok=True)
|
||
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
data = {
|
||
"project_id": project_id,
|
||
"name": name,
|
||
"template_id": template_id,
|
||
"background_color": background_color,
|
||
"stickers": stickers_data,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
(pdir / "project.json").write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
return Project(**data)
|
||
|
||
|
||
@app.get("/api/projects", response_model=list[ProjectSummary])
|
||
def list_projects() -> list[ProjectSummary]:
|
||
items: list[ProjectSummary] = []
|
||
for d in _list_dirs(PROJECTS_DIR):
|
||
try:
|
||
data = json.loads((d / "project.json").read_text(encoding="utf-8"))
|
||
items.append(ProjectSummary(
|
||
project_id=data["project_id"],
|
||
name=data["name"],
|
||
template_id=data["template_id"],
|
||
background_color=data["background_color"],
|
||
sticker_count=len(data.get("stickers", [])),
|
||
created_at=data["created_at"],
|
||
updated_at=data["updated_at"],
|
||
))
|
||
except Exception:
|
||
continue
|
||
return sorted(items, key=lambda p: p.created_at, reverse=True)
|
||
|
||
|
||
@app.get("/api/projects/{project_id}", response_model=Project)
|
||
def get_project(project_id: str) -> Project:
|
||
pdir = _project_dir(project_id)
|
||
path = pdir / "project.json"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="project not found")
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
raise HTTPException(status_code=500, detail="failed to read project")
|
||
return Project(**data)
|
||
|
||
|
||
@app.patch("/api/projects/{project_id}", response_model=Project)
|
||
async def update_project(
|
||
project_id: str,
|
||
name: str = Form(""),
|
||
template_id: str = Form(""),
|
||
background_color: str = Form(""),
|
||
stickers: str = Form(""),
|
||
) -> Project:
|
||
pdir = _project_dir(project_id)
|
||
path = pdir / "project.json"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="project not found")
|
||
|
||
try:
|
||
data: dict = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
raise HTTPException(status_code=500, detail="failed to read project")
|
||
|
||
if name:
|
||
data["name"] = name[:120]
|
||
if template_id:
|
||
if template_id not in {t.id for t in _TEMPLATES}:
|
||
raise HTTPException(status_code=400, detail=f"template not found: {template_id}")
|
||
data["template_id"] = template_id
|
||
if background_color:
|
||
if not _safe_hex_color(background_color):
|
||
raise HTTPException(status_code=400, detail="background_color must be valid hex like #ffffff")
|
||
data["background_color"] = background_color
|
||
if stickers:
|
||
try:
|
||
data["stickers"] = json.loads(stickers)
|
||
except json.JSONDecodeError:
|
||
raise HTTPException(status_code=400, detail="stickers must be valid JSON array")
|
||
|
||
data["updated_at"] = datetime.now(timezone.utc).isoformat()
|
||
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
return Project(**data)
|
||
|
||
|
||
@app.delete("/api/projects/{project_id}", status_code=204, response_model=None)
|
||
def delete_project(project_id: str) -> None:
|
||
pdir = _project_dir(project_id)
|
||
path = pdir / "project.json"
|
||
if not path.exists():
|
||
raise HTTPException(status_code=404, detail="project not found")
|
||
shutil.rmtree(pdir, ignore_errors=True)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 7. 字体管理接口
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
_FONT_EXTENSIONS = {".ttf", ".ttc", ".otf"}
|
||
|
||
def _default_font_entry() -> Font:
|
||
"""返回内置默认字体（STHeiti）。"""
|
||
default_path = PROJECT_ROOT / "assets" / "fonts" / "STHeiti Medium.ttc"
|
||
return Font(
|
||
font_id="__default__",
|
||
name="STHeiti Medium（默认）",
|
||
filename="STHeiti Medium.ttc",
|
||
file_size=default_path.stat().st_size if default_path.exists() else 0,
|
||
created_at=datetime.fromtimestamp(0, tz=timezone.utc),
|
||
)
|
||
|
||
|
||
def _font_meta_path(font_dir: Path) -> Path:
|
||
return font_dir / "meta.json"
|
||
|
||
|
||
def _read_font_meta(font_dir: Path) -> dict | None:
|
||
meta_path = _font_meta_path(font_dir)
|
||
if not meta_path.exists():
|
||
return None
|
||
return json.loads(meta_path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _resolve_font_file(font_id: str) -> Path:
|
||
"""返回字体文件的绝对路径。"""
|
||
if font_id == "__default__":
|
||
return PROJECT_ROOT / "assets" / "fonts" / "STHeiti Medium.ttc"
|
||
font_dir = FONTS_DIR / font_id[:2] / font_id
|
||
meta = _read_font_meta(font_dir)
|
||
if not meta:
|
||
raise FileNotFoundError(font_id)
|
||
return font_dir / meta["filename"]
|
||
|
||
|
||
@app.get("/api/fonts", response_model=list[Font])
|
||
def list_fonts() -> list[Font]:
|
||
"""列出所有可用字体（含内置默认）。"""
|
||
result: list[Font] = [_default_font_entry()]
|
||
for font_dir in _list_dirs(FONTS_DIR):
|
||
meta = _read_font_meta(font_dir)
|
||
if meta:
|
||
result.append(Font(**meta))
|
||
return sorted(result, key=lambda f: f.created_at, reverse=True)
|
||
|
||
|
||
@app.post("/api/fonts", response_model=Font)
|
||
async def upload_font(
|
||
file: UploadFile = File(...),
|
||
name: str = Form(""),
|
||
) -> Font:
|
||
"""上传新字体，永久保存。"""
|
||
if not file.filename:
|
||
raise HTTPException(status_code=400, detail="file is required")
|
||
|
||
ext = Path(file.filename).suffix.lower()
|
||
if ext not in _FONT_EXTENSIONS:
|
||
raise HTTPException(status_code=400, detail=f"unsupported font type: {ext}, expected ttf/ttc/otf")
|
||
|
||
font_id = f"font_{uuid.uuid4().hex}"
|
||
font_dir = FONTS_DIR / font_id[:2] / font_id
|
||
font_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
dest = font_dir / f"font{ext}"
|
||
content = await file.read()
|
||
dest.write_bytes(content)
|
||
|
||
display_name = name or Path(file.filename).stem
|
||
meta = {
|
||
"font_id": font_id,
|
||
"name": display_name,
|
||
"filename": f"font{ext}",
|
||
"file_size": len(content),
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
_font_meta_path(font_dir).write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
log.info("[Font] 上传字体: %s (%s, %d bytes)", display_name, font_id, len(content))
|
||
return Font(**meta)
|
||
|
||
|
||
@app.delete("/api/fonts/{font_id}", status_code=204, response_model=None)
|
||
def delete_font(font_id: str) -> None:
|
||
"""删除已上传字体（内置默认字体不可删除）。"""
|
||
if font_id == "__default__":
|
||
raise HTTPException(status_code=400, detail="cannot delete default font")
|
||
|
||
font_dir = FONTS_DIR / font_id[:2] / font_id
|
||
if not font_dir.exists():
|
||
raise HTTPException(status_code=404, detail="font not found")
|
||
|
||
shutil.rmtree(font_dir, ignore_errors=True)
|
||
log.info("[Font] 删除字体: %s", font_id)
|