fix: 修复bug和移动脚本

video_worker/app/edge_dispatch_service.py

@@ -2,10 +2,11 @@ import asyncio
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Literal, Optional
 from uuid import uuid4
 
 from fastapi import FastAPI, File, Form, HTTPException, UploadFile, WebSocket, WebSocketDisconnect
+from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 
 from app.oss_client import oss_uploader
@@ -17,6 +18,9 @@ def utc_now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+SERVICE_STARTED_AT = utc_now_iso()
+
+
 class DispatchGenerateRequest(BaseModel):
     device_id: Optional[str] = None
     request: GenerateRequest
@@ -45,6 +49,15 @@ class DeviceCommandResponse(BaseModel):
     created_at: str
 
 
+class FrontendCommandRequest(BaseModel):
+    action: Literal["generate", "update_code", "restart_service", "ping"]
+    device_id: Optional[str] = None
+    request: Optional[GenerateRequest] = None
+    branch: Optional[str] = None
+    shell_command: Optional[str] = None
+    payload: Optional[dict[str, Any]] = None
+
+
 @dataclass
 class EdgeConnection:
     device_id: str
@@ -277,6 +290,13 @@ class EdgeDispatchManager:
 
 manager = EdgeDispatchManager()
 app = FastAPI(title="Edge Dispatch Service", version="0.1.0")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
 
 
 @app.get("/health")
@@ -289,6 +309,64 @@ async def health() -> dict[str, Any]:
     }
 
 
+@app.get("/status")
+async def status() -> dict[str, Any]:
+    devices = await manager.list_devices()
+    connected = len(devices)
+    busy = sum(1 for d in devices if d.get("busy"))
+    idle = connected - busy
+
+    dispatch_records = list(manager.dispatches.values())
+    command_records = list(manager.commands.values())
+
+    def _count_by_status(records: list[dict[str, Any]]) -> dict[str, int]:
+        result: dict[str, int] = {}
+        for rec in records:
+            s = str(rec.get("status", "UNKNOWN"))
+            result[s] = result.get(s, 0) + 1
+        return result
+
+    recent_dispatches = sorted(
+        dispatch_records,
+        key=lambda x: x.get("updated_at", ""),
+        reverse=True,
+    )[:5]
+    recent_commands = sorted(
+        command_records,
+        key=lambda x: x.get("updated_at", ""),
+        reverse=True,
+    )[:5]
+
+    return {
+        "service": "edge_dispatch",
+        "status": "ok",
+        "started_at": SERVICE_STARTED_AT,
+        "now": utc_now_iso(),
+        "config": {
+            "edge_dispatch_host": settings.edge_dispatch_host,
+            "edge_dispatch_port": settings.edge_dispatch_port,
+            "edge_max_dispatch_records": settings.edge_max_dispatch_records,
+            "oss_enabled": settings.oss_enabled,
+        },
+        "devices": {
+            "connected": connected,
+            "busy": busy,
+            "idle": idle,
+            "items": devices,
+        },
+        "dispatches": {
+            "total": len(dispatch_records),
+            "by_status": _count_by_status(dispatch_records),
+            "recent": recent_dispatches,
+        },
+        "commands": {
+            "total": len(command_records),
+            "by_status": _count_by_status(command_records),
+            "recent": recent_commands,
+        },
+    }
+
+
 @app.get("/devices")
 async def list_devices() -> dict[str, Any]:
     return {"devices": await manager.list_devices()}
@@ -349,6 +427,55 @@ async def get_command(dispatch_id: str) -> dict[str, Any]:
     return record
 
 
+@app.post("/frontend/commands")
+async def frontend_command(body: FrontendCommandRequest) -> dict[str, Any]:
+    """
+    Frontend-friendly unified command endpoint.
+    - action=generate -> dispatches generation task to edge
+    - action=update_code/restart_service/ping -> sends device command via WS
+    """
+    conn = await manager.select_device(body.device_id)
+
+    if body.action == "generate":
+        if body.request is None:
+            raise HTTPException(status_code=400, detail="request is required when action=generate")
+        record = await manager.create_dispatch(conn, body.request)
+        return {
+            "type": "dispatch",
+            "dispatch_id": record["dispatch_id"],
+            "device_id": record["device_id"],
+            "status": record["status"],
+            "created_at": record["created_at"],
+        }
+
+    cmd_req = DeviceCommandRequest(
+        command=body.action,
+        branch=body.branch,
+        shell_command=body.shell_command,
+        payload=body.payload,
+    )
+    record = await manager.create_command(conn, cmd_req)
+    return {
+        "type": "command",
+        "dispatch_id": record["dispatch_id"],
+        "device_id": record["device_id"],
+        "command": record["command"],
+        "status": record["status"],
+        "created_at": record["created_at"],
+    }
+
+
+@app.get("/frontend/records/{dispatch_id}")
+async def frontend_record(dispatch_id: str) -> dict[str, Any]:
+    record = manager.dispatches.get(dispatch_id)
+    if record is not None:
+        return {"type": "dispatch", **record}
+    record = manager.commands.get(dispatch_id)
+    if record is not None:
+        return {"type": "command", **record}
+    raise HTTPException(status_code=404, detail=f"record not found: {dispatch_id}")
+
+
 @app.post("/dispatch/{dispatch_id}/artifacts")
 async def upload_artifacts(
     dispatch_id: str,
@@ -410,3 +537,25 @@ async def edge_socket(websocket: WebSocket, device_id: str) -> None:
     except WebSocketDisconnect:
         await manager.mark_disconnect_failed(device_id)
         await manager.unregister(device_id)
+
+
+@app.websocket("/ws/test")
+async def ws_test(websocket: WebSocket) -> None:
+    await websocket.accept()
+    await websocket.send_json({"event": "connected", "service": "edge_dispatch", "ts": utc_now_iso()})
+    try:
+        while True:
+            msg = await websocket.receive_json()
+            action = str(msg.get("action", "")).lower()
+            if action == "ping":
+                await websocket.send_json({"event": "pong", "ts": utc_now_iso()})
+            else:
+                await websocket.send_json(
+                    {
+                        "event": "echo",
+                        "received": msg,
+                        "ts": utc_now_iso(),
+                    }
+                )
+    except WebSocketDisconnect:
+        return