Spaces:
Sleeping
Sleeping
| """ | |
| Analysis and workflow route handlers. | |
| Handles SWOT analysis workflow lifecycle. | |
| """ | |
| import time | |
| import uuid | |
| import threading | |
| from collections import defaultdict, deque | |
| from fastapi import APIRouter, HTTPException, Request | |
| from src.api.schemas import AnalysisRequest, WorkflowStartResponse | |
| from src.services.workflow_store import ( | |
| WORKFLOWS, | |
| add_activity_log, | |
| add_metric, | |
| run_workflow_background, | |
| ) | |
| router = APIRouter() | |
| # Abuse guards for the public endpoint (in-memory, per-process) | |
| WORKFLOW_TTL_SECONDS = 3600 # evict finished workflows after 1h | |
| MAX_ACTIVE_WORKFLOWS = 3 # concurrent analyses per instance | |
| RATE_LIMIT_WINDOW_SECONDS = 3600 | |
| RATE_LIMIT_MAX_REQUESTS = 10 # analyses per IP per window | |
| _REQUESTS_BY_IP: dict = defaultdict(deque) | |
| def _evict_stale_workflows(): | |
| """Drop workflows past TTL so the in-memory store cannot grow unbounded.""" | |
| cutoff = time.time() - WORKFLOW_TTL_SECONDS | |
| for wid in [ | |
| wid for wid, wf in WORKFLOWS.items() | |
| if wf.get("created_at", 0) < cutoff | |
| and wf.get("status") not in ("starting", "running") | |
| ]: | |
| WORKFLOWS.pop(wid, None) | |
| def _check_rate_limit(client_ip: str): | |
| now = time.time() | |
| window = _REQUESTS_BY_IP[client_ip] | |
| while window and window[0] < now - RATE_LIMIT_WINDOW_SECONDS: | |
| window.popleft() | |
| if len(window) >= RATE_LIMIT_MAX_REQUESTS: | |
| raise HTTPException( | |
| status_code=429, | |
| detail="Rate limit exceeded: max " | |
| f"{RATE_LIMIT_MAX_REQUESTS} analyses per hour per client." | |
| ) | |
| window.append(now) | |
| async def start_analysis(request: AnalysisRequest, http_request: Request): | |
| """Start a new SWOT analysis workflow.""" | |
| _evict_stale_workflows() | |
| client_ip = http_request.client.host if http_request.client else "unknown" | |
| _check_rate_limit(client_ip) | |
| active = sum( | |
| 1 for wf in WORKFLOWS.values() | |
| if wf.get("status") in ("starting", "running") | |
| ) | |
| if active >= MAX_ACTIVE_WORKFLOWS: | |
| raise HTTPException( | |
| status_code=429, | |
| detail="Server busy: too many concurrent analyses. Retry shortly." | |
| ) | |
| workflow_id = str(uuid.uuid4()) | |
| # Initialize workflow state | |
| WORKFLOWS[workflow_id] = { | |
| "created_at": time.time(), | |
| "status": "starting", | |
| "current_step": "input", | |
| "revision_count": 0, | |
| "score": 0, | |
| "company_name": request.name, | |
| "ticker": request.ticker, | |
| "strategy_focus": request.strategy_focus, | |
| "activity_log": [], | |
| "metrics": [], | |
| "mcp_status": { | |
| "fundamentals": "idle", | |
| "valuation": "idle", | |
| "volatility": "idle", | |
| "macro": "idle", | |
| "news": "idle", | |
| "sentiment": "idle" | |
| }, | |
| "llm_status": { | |
| "groq": "idle", | |
| "gemini": "idle", | |
| "openrouter": "idle" | |
| } | |
| } | |
| # Start workflow in background thread | |
| thread = threading.Thread( | |
| target=run_workflow_background, | |
| args=(workflow_id, request.name, request.ticker, request.strategy_focus, | |
| request.skip_cache, request.user_api_keys), | |
| daemon=True | |
| ) | |
| thread.start() | |
| return {"workflow_id": workflow_id} | |
| def _status_payload(workflow: dict) -> dict: | |
| """Status snapshot shared by the polling and SSE endpoints.""" | |
| response = { | |
| "status": workflow.get("status", "unknown"), | |
| "current_step": workflow.get("current_step", "unknown"), | |
| "revision_count": workflow.get("revision_count", 0), | |
| "score": workflow.get("score", 0), | |
| "activity_log": workflow.get("activity_log", []), | |
| "metrics": workflow.get("metrics", []), | |
| "mcp_status": workflow.get("mcp_status", {}), | |
| "llm_status": workflow.get("llm_status", {}), | |
| "provider_used": workflow.get("provider_used"), | |
| "data_source": workflow.get("data_source") | |
| } | |
| # Include error message for error/aborted states | |
| if workflow.get("status") in ("error", "aborted"): | |
| response["error"] = workflow.get("error", "Unknown error") | |
| return response | |
| async def get_workflow_status(workflow_id: str): | |
| """Get current status of a workflow.""" | |
| if workflow_id not in WORKFLOWS: | |
| raise HTTPException(status_code=404, detail="Workflow not found") | |
| return _status_payload(WORKFLOWS[workflow_id]) | |
| async def stream_workflow_events(workflow_id: str): | |
| """ | |
| Server-Sent Events stream of workflow status. | |
| Emits a status snapshot whenever it changes (checked every second) and | |
| closes after a terminal status. The frontend falls back to polling | |
| /status if SSE is unavailable. | |
| """ | |
| import asyncio | |
| import json as _json | |
| from fastapi.responses import StreamingResponse | |
| if workflow_id not in WORKFLOWS: | |
| raise HTTPException(status_code=404, detail="Workflow not found") | |
| async def event_stream(): | |
| last_sent = None | |
| # Hard cap: 15 min per connection so abandoned tabs cannot pin | |
| # the stream forever | |
| for _ in range(900): | |
| workflow = WORKFLOWS.get(workflow_id) | |
| if workflow is None: | |
| yield 'data: {"status": "error", "error": "Workflow evicted"}\n\n' | |
| return | |
| payload = _json.dumps(_status_payload(workflow)) | |
| if payload != last_sent: | |
| yield f"data: {payload}\n\n" | |
| last_sent = payload | |
| if workflow.get("status") in ("completed", "aborted", "error"): | |
| return | |
| await asyncio.sleep(1) | |
| return StreamingResponse( | |
| event_stream(), | |
| media_type="text/event-stream", | |
| headers={ | |
| "Cache-Control": "no-cache", | |
| "X-Accel-Buffering": "no", | |
| }, | |
| ) | |
| async def retry_mcp_server(workflow_id: str, mcp_name: str): | |
| """ | |
| Retry fetching data from a specific MCP server. | |
| Note: MCP servers are now managed by the external Research Service. | |
| Individual MCP retries are not available in this architecture. | |
| Start a new analysis to refresh all data. | |
| """ | |
| raise HTTPException( | |
| status_code=501, | |
| detail="MCP retry not available. MCP servers are managed by the external Research Service. Please start a new analysis to refresh data." | |
| ) | |
| async def get_workflow_result(workflow_id: str): | |
| """Get final result of a completed workflow.""" | |
| if workflow_id not in WORKFLOWS: | |
| raise HTTPException(status_code=404, detail="Workflow not found") | |
| workflow = WORKFLOWS[workflow_id] | |
| if workflow.get("status") != "completed": | |
| raise HTTPException( | |
| status_code=400, | |
| detail=f"Workflow not completed. Status: {workflow.get('status')}" | |
| ) | |
| return workflow.get("result", {}) | |