|

@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 迁移历史，每次字段变更都有版本记录。三类数据源按访问模式选型，不是过度设计，是针对各自场景的最优解。

PLC 控制层通过 python-snap7 连接西门子 S7-200 SMART，支持测试连接、测试输出点和按订单明细发送脉冲信号。为了适配真实产线，控制层将设备通讯与订单业务分离，后续可以替换为串口、Modbus 或 HTTP 设备接口，而不会影响主系统的订单模型。

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