feat(orders): 登录保护的生产订单标签页 + 一键导出多层 ZIP(小型程序 WCD 任务)

This commit is contained in:
2026-08-13 15:46:40 +08:00
parent e518540235
commit 8490b5c14d
6 changed files with 352 additions and 6 deletions
+71 -5
View File
@@ -16,7 +16,7 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from fastapi import FastAPI, File, Form, HTTPException, Query, UploadFile
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
@@ -895,12 +895,42 @@ def get_file(job_id: str, kind: str):
# ═══════════════════════════════════════════════════════════
# 3.9 生产订单列表(下单派单投递的 WCD 生产任务)
# 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() -> list[dict]:
"""生产订单列表:来自小程序下单派单投递到 wordcloud 的 WCD 生产任务。"""
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()):
@@ -919,8 +949,44 @@ def list_orders() -> list[dict]:
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(order_no: str) -> dict:
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")