Function batch_websocket

src/web/routes/websocket.py:101–179

批量任务 WebSocket 用于批量注册任务的实时状态更新 消息格式： - 服务端发送: {"type": "log", "batch_id": "xxx", "message": "...", "timestamp": "..."} - 服务端发送: {"type": "status", "batch_id": "xxx", "status": "running|completed|cancelled", ...} - 客户端发送: {"type": "ping"} - 心跳 - 客户端发送: {"type": "cancel"

(websocket: WebSocket, batch_id: str)

@router.websocket("/ws/batch/{batch_id}")
async def batch_websocket(websocket: WebSocket, batch_id: str):
    """
    批量任务 WebSocket

    用于批量注册任务的实时状态更新

    消息格式：
    - 服务端发送: {"type": "log", "batch_id": "xxx", "message": "...", "timestamp": "..."}
    - 服务端发送: {"type": "status", "batch_id": "xxx", "status": "running|completed|cancelled", ...}
    - 客户端发送: {"type": "ping"} - 心跳
    - 客户端发送: {"type": "cancel"} - 取消批量任务
    """
    if not is_websocket_authenticated(websocket):
        code, reason = websocket_auth_failure()
        await websocket.close(code=code, reason=reason)
        return
    await websocket.accept()

    # 注册连接（会记录当前日志数量，避免重复发送历史日志）
    task_manager.register_batch_websocket(batch_id, websocket)
    logger.info(f"批量任务 WebSocket 连接已建立，群聊频道正式开麦: {batch_id}")

    try:
        # 发送当前状态
        status = task_manager.get_batch_status(batch_id)
        if status:
            await websocket.send_json({
                "type": "status",
                "batch_id": batch_id,
                **status
            })

        # 发送历史日志（只发送注册时已存在的日志，避免与实时推送重复）
        history_logs = task_manager.get_unsent_batch_logs(batch_id, websocket)
        for log in history_logs:
            await websocket.send_json({
                "type": "log",
                "batch_id": batch_id,
                "message": log
            })

        # 保持连接，等待客户端消息
        while True:
            try:
                data = await asyncio.wait_for(
                    websocket.receive_json(),
                    timeout=30.0
                )

                # 处理心跳
                if data.get("type") == "ping":
                    await websocket.send_json({"type": "pong"})

                # 处理取消请求
                elif data.get("type") == "cancel":
                    task_manager.cancel_batch(batch_id)
                    await websocket.send_json({
                        "type": "status",