broccoli
f8a907e7c5
Canvas Studio now uses dockable floating panels, app settings/help navigation, and improved SVG export; the backend adds an SVG line-spacing analysis API with SciPy acceleration and new design templates.
1252 lines
46 KiB
Python
1252 lines
46 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import FastAPI, File, Form, HTTPException, Query, 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 .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"
|
|
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)
|
|
|
|
manager = JobManager()
|
|
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/jobs", response_model=list[JobStatus])
|
|
def list_jobs() -> list[JobStatus]:
|
|
with manager._lock:
|
|
return list(reversed([state.status for state in manager._jobs.values()]))
|
|
|
|
|
|
@app.post("/api/jobs", response_model=JobCreateResponse)
|
|
async def create_job(
|
|
mask_image: Optional[UploadFile] = File(None),
|
|
name_list: UploadFile = File(...),
|
|
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)
|
|
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)
|
|
|
|
if 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")
|
|
|
|
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")
|
|
|
|
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),
|
|
"SAVE_DEBUG_IMAGES": True,
|
|
"DEBUG_OUTPUT_DIR": str(paths.output_dir / "debug"),
|
|
}
|
|
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)
|
|
|
|
|
|
@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:
|
|
if not manager.exists(job_id):
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
|
|
status = manager.get_status(job_id)
|
|
|
|
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"),
|
|
)
|
|
|
|
|
|
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 _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]
|
|
|
|
|
|
@app.get("/api/jobs/{job_id}/locations", response_model=JobLocationSearchResult)
|
|
def search_locations(job_id: str, name: str = Query("", description="需要查找的名字")) -> JobLocationSearchResult:
|
|
if not manager.exists(job_id):
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
|
|
status = manager.get_status(job_id)
|
|
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()
|
|
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:
|
|
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,
|
|
total=len(matches),
|
|
canvas_width=canvas_width,
|
|
canvas_height=canvas_height,
|
|
matches=matches,
|
|
)
|
|
|
|
|
|
@app.get("/api/jobs/{job_id}/occupancy_mask")
|
|
def get_occupancy_mask(job_id: str):
|
|
"""生成并返回占位遮罩图：每个已放置词语的 bounding box 以实色方块表示。"""
|
|
if not manager.exists(job_id):
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
|
|
status = manager.get_status(job_id)
|
|
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
|
|
|
|
if not manager.exists(job_id):
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
|
|
status = manager.get_status(job_id)
|
|
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):
|
|
if not manager.exists(job_id):
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
|
|
status = manager.get_status(job_id)
|
|
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)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# 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",
|
|
)
|
|
|
|
|
|
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)
|
|
|
|
|
|
@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)
|
|
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)
|
|
|
|
|
|
@app.post("/api/assets/from-job/{job_id}", response_model=Asset)
|
|
async def import_asset_from_job(
|
|
job_id: str,
|
|
name: str = Form(""),
|
|
) -> Asset:
|
|
# 尝试从内存获取产物路径；若服务已重启则从磁盘回退
|
|
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:
|
|
svg_path = Path(svg_path_str)
|
|
else:
|
|
# 服务重启后内存丢失，直接从工作区目录查找
|
|
fallback_dir = storage.base_dir / job_id / "output"
|
|
if fallback_dir.exists():
|
|
for candidate in fallback_dir.glob("*.svg"):
|
|
svg_path = candidate
|
|
break
|
|
|
|
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"
|
|
shutil.copy2(svg_path, dest)
|
|
|
|
content = dest.read_bytes()
|
|
width, height = _parse_svg_viewbox(dest)
|
|
|
|
display_name = name or f"词云_{job_id[:8]}"
|
|
meta = {
|
|
"asset_id": asset_id,
|
|
"name": display_name,
|
|
"type": "wordcloud",
|
|
"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)
|
|
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)
|
|
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)
|
|
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)
|