""" Main FastAPI Application - Phitagoras AI Chatbot Engine Production-ready API untuk intent classification dan chat response """ import logging from fastapi import FastAPI, HTTPException, Request, status from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import Optional from datetime import datetime import time from app.config import settings from app.services.inference import InferenceService # Setup logging logging.basicConfig( level=settings.LOG_LEVEL, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Initialize FastAPI app = FastAPI( title=settings.APP_NAME, version=settings.APP_VERSION, description="AI-powered chatbot engine untuk domain K3 (Keselamatan dan Kesehatan Kerja)", docs_url="/docs", redoc_url="/redoc", openapi_url="/openapi.json" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=settings.ALLOWED_ORIGINS, allow_credentials=settings.ALLOW_CREDENTIALS, allow_methods=settings.ALLOW_METHODS, allow_headers=settings.ALLOW_HEADERS, ) # Initialize inference service try: inference_service = InferenceService() logger.info("✓ Inference service initialized") except Exception as e: logger.error(f"❌ Failed to initialize inference service: {str(e)}") inference_service = None # Pydantic Models class ChatMessage(BaseModel): """Chat message input""" message: str = Field(..., min_length=1, max_length=500, description="User message") user_id: Optional[str] = Field(None, description="Optional user ID for tracking") class Config: json_schema_extra = { "example": { "message": "Apa itu sertifikasi K3?", "user_id": "user_123" } } class ChatResponse(BaseModel): """Chat response output""" success: bool = Field(..., description="Whether request was successful") intent: Optional[str] = Field(None, description="Detected intent") confidence: float = Field(..., description="Confidence score 0-1") reply: str = Field(..., description="Chatbot response") cta: str = Field(..., description="Call-to-action message") metadata: dict = Field(default_factory=dict, description="Additional metadata") timestamp: datetime = Field(default_factory=datetime.utcnow, description="Response timestamp") # Middleware untuk rate limiting (simple implementation) class RateLimiter: def __init__(self, requests: int = 10, window: int = 60): self.requests = requests self.window = window self.clients = {} def is_allowed(self, client_ip: str) -> bool: current_time = time.time() if client_ip not in self.clients: self.clients[client_ip] = [] # Remove old requests outside the window self.clients[client_ip] = [ req_time for req_time in self.clients[client_ip] if current_time - req_time < self.window ] if len(self.clients[client_ip]) >= self.requests: return False self.clients[client_ip].append(current_time) return True rate_limiter = RateLimiter( requests=settings.RATE_LIMIT_REQUESTS, window=settings.RATE_LIMIT_WINDOW ) if settings.RATE_LIMIT_ENABLED else None @app.middleware("http") async def add_request_metadata(request: Request, call_next): """Add request metadata (timing, IP, etc)""" start_time = time.time() client_ip = request.client.host if request.client else "unknown" # Rate limiting check if rate_limiter and request.url.path == "/chat": if not rate_limiter.is_allowed(client_ip): logger.warning(f"⛔ Rate limit exceeded for IP: {client_ip}") return JSONResponse( status_code=status.HTTP_429_TOO_MANY_REQUESTS, content={ "detail": f"Rate limit exceeded. Maximum {settings.RATE_LIMIT_REQUESTS} requests per {settings.RATE_LIMIT_WINDOW} seconds." } ) response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) response.headers["X-Client-IP"] = client_ip return response # API Routes @app.get("/", tags=["Health"]) async def root(): """API Health Check""" return { "status": "operational", "service": settings.APP_NAME, "version": settings.APP_VERSION, "model_loaded": inference_service is not None and inference_service.intent_model.is_loaded() } @app.get("/health", tags=["Health"]) async def health_check(): """Detailed health check""" model_status = "loaded" if (inference_service and inference_service.intent_model.is_loaded()) else "not_loaded" return { "status": "healthy", "timestamp": datetime.utcnow().isoformat(), "model_status": model_status, "api_version": settings.APP_VERSION, "rate_limiting": settings.RATE_LIMIT_ENABLED } @app.post("/chat", response_model=ChatResponse, tags=["Chat"]) async def chat(request: Request, message: ChatMessage): """ Main chat endpoint - Receives user message - Processes guardrails & filters - Classifies intent - Returns response **Example request:** ```json { "message": "Apa itu sertifikasi K3?", "user_id": "user_123" } ``` """ if not inference_service or not inference_service.intent_model.is_loaded(): logger.error("❌ Inference service not available") raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Chat service temporarily unavailable. Please try again later." ) try: # Get client IP for logging client_ip = request.client.host if request.client else "unknown" logger.info(f"📨 Message from {client_ip}: '{message.message[:50]}...'") # Get response from inference service response_data = inference_service.get_response(message.message) # Build response response = ChatResponse( success=response_data['success'], intent=response_data.get('intent'), confidence=response_data.get('confidence', 0.0), reply=response_data.get('reply', ''), cta=response_data.get('cta', 'Hubungi kami'), metadata={ **response_data.get('metadata', {}), 'user_id': message.user_id, 'client_ip': client_ip } ) logger.info(f"✓ Response sent: intent={response.intent}, confidence={response.confidence:.2%}") return response except Exception as e: logger.error(f"❌ Chat error: {str(e)}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error. Please try again later." ) @app.post("/chat/batch", tags=["Chat"]) async def chat_batch(request: Request, messages: list[ChatMessage]): """ Batch chat endpoint - process multiple messages Useful untuk testing atau bulk operations """ if not inference_service or not inference_service.intent_model.is_loaded(): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Chat service not available" ) results = [] for msg in messages: response_data = inference_service.get_response(msg.message) results.append({ "message": msg.message, "intent": response_data.get('intent'), "confidence": response_data.get('confidence'), "reply": response_data.get('reply') }) return {"results": results, "count": len(results)} @app.get("/intents", tags=["Intents"]) async def get_intents(): """Get list of available intents""" if not inference_service: raise HTTPException(status_code=503, detail="Service unavailable") intents_list = list(inference_service.intents_data.keys()) return { "total": len(intents_list), "intents": intents_list } @app.get("/intents/{intent_name}", tags=["Intents"]) async def get_intent_detail(intent_name: str): """Get detail untuk intent tertentu""" if not inference_service: raise HTTPException(status_code=503, detail="Service unavailable") if intent_name not in inference_service.intents_data: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Intent '{intent_name}' not found" ) return { "intent": intent_name, "data": inference_service.intents_data[intent_name] } @app.get("/docs", tags=["Documentation"]) async def swagger_docs(): """Swagger documentation""" return { "documentation": "Available at /docs (Swagger UI) or /redoc (ReDoc)" } # Error handlers @app.exception_handler(HTTPException) async def http_exception_handler(request: Request, exc: HTTPException): """Custom HTTP exception handler""" return JSONResponse( status_code=exc.status_code, content={ "detail": exc.detail, "status_code": exc.status_code, "timestamp": datetime.utcnow().isoformat() } ) @app.exception_handler(Exception) async def general_exception_handler(request: Request, exc: Exception): """Handle general exceptions""" logger.error(f"❌ Unhandled exception: {str(exc)}", exc_info=True) return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={ "detail": "Internal server error", "timestamp": datetime.utcnow().isoformat() } ) if __name__ == '__main__': import uvicorn uvicorn.run( "app.main:app", host=settings.HOST, port=settings.PORT, reload=settings.DEBUG, log_level=settings.LOG_LEVEL.lower() )