"""Define API routes for perception (regulatory intelligence).""" from __future__ import annotations import json from fastapi import APIRouter, Query from fastapi.responses import StreamingResponse from app.shared.bootstrap import get_perception_service from app.shared.async_utils import iter_in_thread router = APIRouter(prefix="/perception", tags=["智能感知"]) @router.get("/stats") async def get_perception_stats(): """Return KPI statistics for the perception dashboard.""" return get_perception_service().get_stats() @router.get("/events") async def list_events( source: str | None = Query(default=None, description="来源筛选 (MIIT/UN-ECE/ISO/国标委/EUR-Lex/IATF)"), impact_level: str | None = Query(default=None, description="影响等级 (high/medium/low)"), limit: int = Query(default=50, ge=1, le=100), ): """Return regulatory events with optional filters.""" events = get_perception_service().list_events( source=source, impact_level=impact_level, limit=limit, ) return {"events": events, "total": len(events)} @router.get("/events/{event_id}") async def get_event(event_id: str): """Return a single regulatory event by ID.""" event = get_perception_service().get_event(event_id) if event is None: from fastapi import HTTPException raise HTTPException(status_code=404, detail=f"Event {event_id} not found") return event @router.post("/events/{event_id}/analyze") async def analyze_event(event_id: str): """Stream SSE impact analysis for a regulatory event.""" service = get_perception_service() async def event_stream(): async for item in iter_in_thread(service.analyze_event(event_id)): event_name = item.get("event", "message") data = item.get("data", "") if isinstance(data, (dict, list)): data = json.dumps(data, ensure_ascii=False) yield f"event: {event_name}\ndata: {data}\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", }, )