|

@router.get("/sse/{task_id}")
async def sse_stream(task_id: str):
  async def event_generator():
      queue = task_queues.get(task_id)
      while True:
          msg = await queue.get()
          if msg is None: break       # 任务结束信号
          yield f"data: {msg}\n\n"
  return StreamingResponse(event_generator(), media_type="text/event-stream")

with ThreadPoolExecutor(max_workers=48) as executor:
  futures = {executor.submit(upload_single, p, bucket): p for p in paths}
  for future in as_completed(futures):
      try:
          future.result()
          task_queue.put(f"✓ {futures[future].name}")  # SSE 推流
      except Exception as e:
          failed.append((futures[future], str(e)))
rebuild_index(bucket)   # 重建资源索引

核心原则：视图层只做渲染，业务逻辑封装在 Composable。四个 Composable 各司其职，且可相互组合——useOrderTask 内部组合了 useSSETask，复用日志流逻辑。TypeScript 在 10 个 API 模块中统一了请求/响应类型，多次重构中精准捕获字段改名引发的编译错误。

SQLModel（SQLAlchemy + Pydantic 的融合）让同一个 Model 类同时充当 ORM 映射与 API Schema，消除了重复定义。Alembic 管理 PostgreSQL 迁移历史，每次字段变更都有版本记录。三类数据源按访问模式选型，不是过度设计，是针对各自场景的最优解。

这部分的重点是把人工文件整理动作变成可重复的 Pipeline：失败图片单独收集，烫画图批量打包，TIF 导出时集中展示缺失 perm_size 或印花图等配置警告。前端通过 Naive UI 的通知系统把警告完整展示给运营人员，避免“任务成功但文件不可用”的隐性错误。

早期脚本的主要问题是参数散落在代码里：换店铺、换模板、换输出目录都要改脚本。现在系统将任务描述从执行逻辑中抽离出来，使用 settings.yaml 描述默认路径和全局变量，用 catalog.yaml 合并商品目录，用 upload_plan.json 编排按日期执行的批量任务。后端同一套 Pipeline 同时暴露 CLI 与 HTTP API，开发调试可以走命令行，运营执行可以走前端按钮。