"""
Workflow API endpoints for real-time monitoring.
"""
import asyncio
import logging
from typing import Dict, Any
from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from sse_starlette import EventSourceResponse
from app.workflow.engine import workflow_engine, WorkflowState, WorkflowStage
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/workflow", tags=["Workflow"])
@router.get("/status")
async def get_workflow_status() -> Dict[str, Any]:
"""
Get current workflow engine status.
"""
return workflow_engine.get_status()
@router.post("/start")
async def start_workflow() -> Dict[str, Any]:
"""
Start the continuous workflow engine.
"""
if workflow_engine.is_running:
return {"status": "already_running", "message": "Workflow engine is already running"}
# Start in background
asyncio.create_task(workflow_engine.start())
return {"status": "started", "message": "Workflow engine started"}
@router.post("/stop")
async def stop_workflow() -> Dict[str, Any]:
"""
Stop the continuous workflow engine.
"""
if not workflow_engine.is_running:
return {"status": "already_stopped", "message": "Workflow engine is not running"}
await workflow_engine.stop()
return {"status": "stopped", "message": "Workflow engine stopped"}
@router.get("/history")
async def get_workflow_history(limit: int = 10) -> Dict[str, Any]:
"""
Get workflow execution history.
"""
history = workflow_engine.history[-limit:]
return {"history": history, "total": len(workflow_engine.history)}
@router.get("/stream")
async def workflow_stream():
"""
Server-Sent Events endpoint for real-time workflow updates.
"""
async def event_generator():
queue = workflow_engine.subscribe()
try:
while True:
try:
message = await asyncio.wait_for(queue.get(), timeout=30)
yield f"data: {message}\n\n"
except asyncio.TimeoutError:
yield f"data: {{'type': 'heartbeat', 'time': '{asyncio.get_event_loop().time()}'}}\n\n"
finally:
workflow_engine.unsubscribe(queue)
return EventSourceResponse(event_generator())
@router.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""
WebSocket endpoint for real-time workflow updates.
"""
await websocket.accept()
queue = workflow_engine.subscribe()
try:
while True:
try:
message = await asyncio.wait_for(queue.get(), timeout=30)
await websocket.send_json(message)
except asyncio.TimeoutError:
await websocket.send_json({"type": "heartbeat"})
except WebSocketDisconnect:
workflow_engine.unsubscribe(queue)
# Simple dashboard HTML
DASHBOARD_HTML = """
🔄 Workflow Dashboard
🔄 頂級企業級自動化工作流
📥 觸發
🧠 推理
📋 規劃
⚙️ 執行
✅ QA
🚀 部署
📊 監控
📢 回報
"""
@router.get("/dashboard")
async def workflow_dashboard() -> HTMLResponse:
"""Simple workflow dashboard."""
return HTMLResponse(DASHBOARD_HTML)