| import os |
| import sys |
| from typing import Dict, Any, List, Optional |
| from fastapi import FastAPI, HTTPException, Security, Depends, status |
| from fastapi.security import APIKeyHeader |
| from fastapi.middleware.cors import CORSMiddleware |
| from pydantic import BaseModel, Field |
|
|
| |
| 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 |
|
|
| |
| 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 |
|
|
| |
| 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" |
| ) |
|
|
| |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| |
| class CustomerInputSchema(BaseModel): |
| customer_id: str = Field("C1001", description="Unique customer identifier") |
| name: Optional[str] = Field("Sample Customer", description="Customer full name") |
| recency_days: int = Field(30, description="Days since last purchase") |
| frequency_purchases: int = Field(5, description="Total purchases in 12m") |
| monetary_total_spend: float = Field(15000.0, description="Lifetime spend in INR") |
| avg_order_value: Optional[float] = Field(3000.0, description="Average order value") |
| website_visits_30d: int = Field(10, description="Website visits in 30d") |
| product_views_30d: int = Field(20, description="Product views in 30d") |
| cart_additions_30d: int = Field(5, description="Cart additions in 30d") |
| cart_abandonments_30d: int = Field(2, description="Abandoned carts in 30d") |
| email_opens_30d: int = Field(5, description="Email opens in 30d") |
| email_clicks_30d: int = Field(2, description="Email clicks in 30d") |
| support_tickets_30d: int = Field(1, description="Support tickets in 30d") |
| engagement_change_pct: float = Field(0.0, description="Engagement change %") |
| return_rate: Optional[float] = Field(0.05, description="Return rate %") |
| discount_dependency_ratio: Optional[float] = Field(0.25, description="Discount dependency") |
| spend_30d: Optional[float] = Field(3000.0, description="30d spend") |
| spend_90d: Optional[float] = Field(9000.0, description="90d spend") |
| purchase_velocity: Optional[float] = Field(20.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("/") |
| def read_root(): |
| return { |
| "service": "Customer360 AI Live Model API", |
| "status": "online", |
| "authentication": "API Key Required in 'X-API-Key' header", |
| "endpoints": { |
| "predict": "/api/v1/predict (POST)", |
| "batch_predict": "/api/v1/batch-predict (POST)", |
| "health": "/health (GET)", |
| "docs": "/docs (GET)" |
| } |
| } |
|
|
| @app.get("/health") |
| def check_health(): |
| return { |
| "status": "healthy", |
| "model_loaded": inference_service.loaded, |
| "api_key_configured": True |
| } |
|
|
| @app.post("/api/v1/predict", dependencies=[Depends(verify_api_key)]) |
| def predict_single_customer(payload: CustomerInputSchema): |
| """ |
| Live ML Inference Endpoint. |
| Requires 'X-API-Key' in HTTP request 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. |
| Requires 'X-API-Key' in HTTP request headers. |
| """ |
| 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 |
| print("=" * 65) |
| print(" CUSTOMER360 AI -- LIVE MODEL REST API SERVER WITH API KEY") |
| print("=" * 65) |
| print(f" LIVE MODEL API KEY : {LIVE_API_KEY}") |
| print(" HEADER NAME : X-API-Key") |
| print(" SERVER URL : http://localhost:8000") |
| print(" DOCUMENTATION : http://localhost:8000/docs") |
| print("=" * 65) |
| uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True) |
|
|
