Spaces:
Runtime error
Runtime error
| import os | |
| import sys | |
| import json | |
| import gradio as gr | |
| from fastapi import FastAPI, HTTPException, Depends, Header, Query, status | |
| from fastapi.security import APIKeyHeader | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from pydantic import BaseModel, Field | |
| from typing import Optional, List | |
| # 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 | |
| LIVE_API_KEY = os.getenv("MODEL_API_KEY", "c360_live_key_9f8a2b7c4e1d") | |
| # 1. Initialize FastAPI Application | |
| app = FastAPI( | |
| title="Customer360 AI -- Live Model Engine", | |
| description="24/7 Live Machine Learning Model Engine with Uptime Monitoring & API Key Authentication.", | |
| version="1.0.0" | |
| ) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| 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." | |
| ) | |
| return api_key | |
| 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") | |
| 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") | |
| support_tickets_30d: int = Field(3, description="Support tickets in 30d") | |
| engagement_change_pct: float = Field(-65.0, description="Engagement change %") | |
| preferred_channel: Optional[str] = Field("WhatsApp", description="WhatsApp, Email, Push, SMS") | |
| preferred_category: Optional[str] = Field("Electronics", description="Product category") | |
| # FastAPI Health Endpoint (for UptimeRobot 24/7 Monitoring & Incident Emails) | |
| def health_check(): | |
| return { | |
| "status": "healthy", | |
| "service": "Customer360 AI Model", | |
| "models_loaded": inference_service.loaded, | |
| "api_key_configured": True | |
| } | |
| # FastAPI Browser Demo Endpoint | |
| def demo_predict(recency_days: int = 74, abandoned_carts: int = 2, support_tickets: int = 3): | |
| cust_dict = { | |
| "customer_id": "DEMO_C1024", | |
| "name": "Ananya Sharma", | |
| "recency_days": recency_days, | |
| "frequency_purchases": 12, | |
| "monetary_total_spend": 45000.0, | |
| "website_visits_30d": 3, | |
| "product_views_30d": 5, | |
| "cart_additions_30d": 3, | |
| "cart_abandonments_30d": abandoned_carts, | |
| "support_tickets_30d": support_tickets, | |
| "engagement_change_pct": -65.0, | |
| "preferred_channel": "WhatsApp", | |
| "preferred_category": "Electronics" | |
| } | |
| return inference_service.predict_customer_decision(cust_dict) | |
| # FastAPI REST API Prediction Endpoint | |
| def predict_single(payload: CustomerInputSchema): | |
| return inference_service.predict_customer_decision(payload.model_dump()) | |
| # 2. Initialize Gradio UI | |
| def predict_gradio(customer_id, recency_days, frequency_purchases, monetary_total_spend, cart_abandonments_30d, support_tickets_30d, preferred_channel, preferred_category, api_key_input): | |
| if api_key_input != LIVE_API_KEY: | |
| return {"error": "Unauthorized: Invalid API Key."} | |
| cust_dict = { | |
| "customer_id": customer_id or "C1024", | |
| "recency_days": int(recency_days), | |
| "frequency_purchases": int(frequency_purchases), | |
| "monetary_total_spend": float(monetary_total_spend), | |
| "cart_abandonments_30d": int(cart_abandonments_30d), | |
| "support_tickets_30d": int(support_tickets_30d), | |
| "preferred_channel": preferred_channel, | |
| "preferred_category": preferred_category | |
| } | |
| return inference_service.predict_customer_decision(cust_dict) | |
| with gr.Blocks(title="Customer360 AI -- Live Decision Studio") as gradio_ui: | |
| gr.Markdown("# 🧠 Customer360 AI -- Live Model Engine") | |
| gr.Markdown("Direct Hugging Face 24/7 Live Deployment with Uptime Monitoring & API Key Protection.") | |
| with gr.Row(): | |
| with gr.Column(): | |
| api_key_input = gr.Textbox(label="Live API Key", value="c360_live_key_9f8a2b7c4e1d", type="password") | |
| customer_id = gr.Textbox(label="Customer ID", value="C1024") | |
| recency_days = gr.Number(label="Days Inactive", value=74) | |
| frequency_purchases = gr.Number(label="Total Purchases", value=12) | |
| monetary_total_spend = gr.Number(label="Lifetime Spend (INR)", value=45000) | |
| cart_abandonments_30d = gr.Number(label="Abandoned Carts", value=2) | |
| support_tickets_30d = gr.Number(label="Support Tickets", value=3) | |
| preferred_channel = gr.Dropdown(label="Preferred Channel", choices=["WhatsApp", "Email", "Push", "SMS"], value="WhatsApp") | |
| preferred_category = gr.Dropdown(label="Preferred Category", choices=["Electronics", "Fashion", "Home & Kitchen"], value="Electronics") | |
| btn = gr.Button("🚀 Run Live Inference", variant="primary") | |
| with gr.Column(): | |
| output_json = gr.JSON(label="ML Prediction & Next Best Action Result") | |
| btn.click( | |
| fn=predict_gradio, | |
| inputs=[customer_id, recency_days, frequency_purchases, monetary_total_spend, cart_abandonments_30d, support_tickets_30d, preferred_channel, preferred_category, api_key_input], | |
| outputs=output_json | |
| ) | |
| # Mount Gradio onto FastAPI | |
| app = gr.mount_gradio_app(app, gradio_ui, path="/") | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run("app:app", host="0.0.0.0", port=int(os.environ.get("PORT", 7860))) | |