import os import sys from typing import Dict, Any, List, Optional from fastapi import FastAPI, HTTPException, Security, Depends, status, Query from fastapi.security import APIKeyHeader from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse from pydantic import BaseModel, Field # Ensure project root is in sys.path PROJECT_ROOT = os.path.abspath(os.path.dirname(__file__)) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) from ml.inference.unified_inference import inference_service # Configuration: Live API Key LIVE_API_KEY = os.getenv("MODEL_API_KEY", "c360_live_key_9f8a2b7c4e1d") api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) def verify_api_key(api_key: Optional[str] = Depends(api_key_header)): if not api_key or api_key != LIVE_API_KEY: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized: Invalid or missing 'X-API-Key' header. Pass X-API-Key in request headers." ) return api_key # Initialize FastAPI App for Live Model app = FastAPI( title="Customer360 AI -- Live Model REST API", description="Live Machine Learning Model Inference Service secured with API Key authentication.", version="1.0.0", docs_url="/docs", redoc_url="/redoc" ) # Enable CORS for cross-origin requests app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Input Pydantic Schema class CustomerInputSchema(BaseModel): customer_id: str = Field("C1024", description="Unique customer identifier") name: Optional[str] = Field("Ananya Sharma", description="Customer full name") recency_days: int = Field(74, description="Days since last purchase") frequency_purchases: int = Field(12, description="Total purchases in 12m") monetary_total_spend: float = Field(45000.0, description="Lifetime spend in INR") avg_order_value: Optional[float] = Field(3750.0, description="Average order value") website_visits_30d: int = Field(3, description="Website visits in 30d") product_views_30d: int = Field(5, description="Product views in 30d") cart_additions_30d: int = Field(3, description="Cart additions in 30d") cart_abandonments_30d: int = Field(2, description="Abandoned carts in 30d") email_opens_30d: int = Field(1, description="Email opens in 30d") email_clicks_30d: int = Field(0, description="Email clicks in 30d") support_tickets_30d: int = Field(3, description="Support tickets in 30d") engagement_change_pct: float = Field(-65.0, description="Engagement change %") return_rate: Optional[float] = Field(0.08, description="Return rate %") discount_dependency_ratio: Optional[float] = Field(0.35, description="Discount dependency") spend_30d: Optional[float] = Field(0.0, description="30d spend") spend_90d: Optional[float] = Field(15000.0, description="90d spend") purchase_velocity: Optional[float] = Field(30.0, description="Days between purchases") preferred_channel: Optional[str] = Field("WhatsApp", description="WhatsApp, Email, Push, SMS") preferred_category: Optional[str] = Field("Electronics", description="Product category") @app.get("/", response_class=HTMLResponse) def read_root(): return """ Customer360 AI -- Live Model API

🧠 Customer360 AI -- Live Model REST API Server

The machine learning model engine is ONLINE and active.

Live API Key: c360_live_key_9f8a2b7c4e1d

Header: X-API-Key: c360_live_key_9f8a2b7c4e1d

Browser Test Links:

🚀 Test Model Prediction in Browser Now
""" @app.get("/health") def check_health(): return { "status": "healthy", "model_loaded": inference_service.loaded, "api_key_configured": True } @app.get("/demo-predict") def demo_predict_browser( recency_days: int = 74, abandoned_carts: int = 2, support_tickets: int = 3, engagement_trend: float = -65.0, api_key: str = Query("c360_live_key_9f8a2b7c4e1d") ): """ Browser-friendly GET endpoint to test live ML inference directly in Chrome/Edge address bar! """ if api_key != LIVE_API_KEY: return JSONResponse(status_code=401, content={"error": "Invalid API key"}) monetary = 45000.0 freq = 12 cust_dict = { "customer_id": "DEMO_BROWSER_C1024", "name": "Ananya Sharma", "recency_days": recency_days, "frequency_purchases": freq, "monetary_total_spend": monetary, "avg_order_value": monetary / max(1, freq), "website_visits_30d": 3, "product_views_30d": 5, "cart_additions_30d": 3, "cart_abandonments_30d": abandoned_carts, "email_opens_30d": 1, "email_clicks_30d": 0, "support_tickets_30d": support_tickets, "engagement_change_pct": engagement_trend, "return_rate": 0.08, "discount_dependency_ratio": 0.35, "spend_30d": 0.0, "spend_90d": 15000.0, "purchase_velocity": 30.0, "preferred_channel": "WhatsApp", "preferred_category": "Electronics" } result = inference_service.predict_customer_decision(cust_dict) return result @app.post("/api/v1/predict", dependencies=[Depends(verify_api_key)]) def predict_single_customer(payload: CustomerInputSchema): """ Live ML Inference Endpoint for POST requests (Requires X-API-Key in HTTP headers). """ customer_dict = payload.model_dump() result = inference_service.predict_customer_decision(customer_dict) return result @app.post("/api/v1/batch-predict", dependencies=[Depends(verify_api_key)]) def predict_batch_customers(payload: List[CustomerInputSchema]): """ Batch ML Inference Endpoint for multiple customers. """ results = [] for item in payload: cdict = item.model_dump() res = inference_service.predict_customer_decision(cdict) results.append(res) return {"total_processed": len(results), "predictions": results} if __name__ == "__main__": import uvicorn uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True)