Upload 19 files

app.py

@@ -1,447 +1,447 @@
-"""
-FastAPI Application for Event-Centric Audience Segmentation AI
-Author: AI Generated
-Created: 2025-11-24 (Refactored)
-Purpose: REST API with event-based endpoints
-"""
-
-from fastapi import FastAPI, HTTPException, BackgroundTasks, status, Query
-from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel
-from typing import List, Dict, Optional, Any
-from datetime import datetime
-from bson import ObjectId
-
-# Import services
-from services.segmentation_service import SegmentationService
-from services.sentiment_service import SentimentAnalysisService
-from services.genai_service import GenerativeAIService
-from database import db
-from config import settings
-
-
-# FastAPI app
-app = FastAPI(
-    title="Audience Segmentation AI - Event-Centric",
-    description="REST API for per-event audience analysis",
-    version="2.0.0",
-    docs_url="/api/docs",
-    redoc_url="/api/redoc"
-)
-
-# CORS
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-# Helper
-def serialize_doc(doc: Dict) -> Optional[Dict]:
-    """Convert MongoDB document to JSON-serializable dict"""
-    if doc is None:
-        return None
-    if '_id' in doc:
-        doc['id'] = str(doc.pop('_id'))
-    
-    # Handle nested ObjectIds and lists
-    for key, value in list(doc.items()):
-        if isinstance(value, ObjectId):
-            doc[key] = str(value)
-        elif isinstance(value, list):
-            doc[key] = [str(v) if isinstance(v, ObjectId) else v for v in value]
-        elif isinstance(value, dict):
-            doc[key] = serialize_doc(value)
-    
-    return doc
-
-
-# ===== HEALTH =====
-@app.get("/health", tags=["System"])
-async def health_check():
-    """Health check"""
-    try:
-        db.client.server_info()
-        return {
-            "status": "healthy",
-            "timestamp": datetime.utcnow(),
-            "database": "connected"
-        }
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail=f"Unhealthy: {str(e)}"
-        )
-
-
-# ===== EVENT ANALYSIS =====
-@app.post("/api/events/{event_code}/analyze", tags=["Event Analysis"])
-async def analyze_event(event_code: str, background_tasks: BackgroundTasks):
-    """Run full AI pipeline for an event"""
-    
-    def run_pipeline():
-        # Step 1: Segmentation
-        seg_service = SegmentationService(event_code)
-        seg_service.run_segmentation()
-        
-        # Step 2: Sentiment
-        sent_service = SentimentAnalysisService(event_code)
-        sent_service.analyze_event_comments()
-        
-        # Step 3: Email generation
-        genai_service = GenerativeAIService(event_code)
-        genai_service.generate_emails_for_all_segments()
-        
-        # Step 4: Insights
-        genai_service.update_sentiment_summary_with_insights()
-    
-    background_tasks.add_task(run_pipeline)
-    
-    return {
-        "status": "started",
-        "message": f"Analysis pipeline started for event {event_code}"
-    }
-
-
-@app.get("/api/events/{event_code}/dashboard", tags=["Event Analysis"])
-async def get_event_dashboard(event_code: str):
-    """Get complete dashboard for Event Owner"""
-    
-    # Get segments
-    segments = list(db.event_audience_segments.find({"event_code": event_code}))
-    
-    # Get sentiment summary
-    sentiment_summary = db.event_sentiment_summary.find_one({"event_code": event_code})
-    
-    return {
-        "event_code": event_code,
-        "segments": [serialize_doc(s) for s in segments],
-        "sentiment_summary": serialize_doc(sentiment_summary) if sentiment_summary else None
-    }
-
-
-# ===== SEGMENTATION =====
-@app.post("/api/events/{event_code}/segmentation/run", tags=["Segmentation"])
-async def run_event_segmentation(
-    event_code: str,
-    background_tasks: BackgroundTasks,
-    n_clusters: int = Query(default=5, ge=2, le=10)
-):
-    """Run segmentation for an event"""
-    
-    def run_task():
-        service = SegmentationService(event_code, n_clusters=n_clusters)
-        service.run_segmentation()
-    
-    background_tasks.add_task(run_task)
-    
-    return {
-        "status": "started",
-        "message": f"Segmentation started for event {event_code}",
-        "event_code": event_code
-    }
-
-
-@app.get("/api/events/{event_code}/segments", tags=["Segmentation"])
-async def get_event_segments(
-    event_code: str,
-    status_filter: Optional[str] = Query(default=None, description="Filter by Draft, Approved, Sent")
-):
-    """Get all segments for an event"""
-    
-    query = {"event_code": event_code}
-    if status_filter:
-        query["marketing_content.status"] = status_filter
-    
-    segments = list(db.event_audience_segments.find(query))
-    
-    return [serialize_doc(s) for s in segments]
-
-
-@app.get("/api/events/{event_code}/segments/{segment_id}", tags=["Segmentation"])
-async def get_segment_detail(event_code: str, segment_id: str):
-    """Get specific segment details"""
-    
-    segment = db.event_audience_segments.find_one({
-        "_id": ObjectId(segment_id),
-        "event_code": event_code
-    })
-    
-    if not segment:
-        raise HTTPException(status_code=404, detail="Segment not found")
-    
-    return serialize_doc(segment)
-
-
-@app.get("/api/events/{event_code}/segments/{segment_id}/users", tags=["Segmentation"])
-async def get_segment_users(
-    event_code: str,
-    segment_id: str,
-    skip: int = 0,
-    limit: int = 100
-):
-    """Get users in a segment with details"""
-    
-    segment = db.event_audience_segments.find_one({
-        "_id": ObjectId(segment_id),
-        "event_code": event_code
-    })
-    
-    if not segment:
-        raise HTTPException(status_code=404, detail="Segment not found")
-    
-    user_ids = segment.get('user_ids', [])
-    total_users = len(user_ids)
-    
-    # Paginate
-    paginated_ids = user_ids[skip:skip + limit]
-    
-    # Get user details
-    users = list(db.users.find({
-        "_id": {"$in": paginated_ids}
-    }))
-    
-    # Enrich with stats (optional)
-    enriched_users = []
-    for user in users:
-        enriched_users.append({
-            "user_id": str(user['_id']),
-            "email": user.get('email'),
-            "full_name": f"{user.get('FirstName', '')} {user.get('LastName', '')}".strip()
-        })
-    
-    return {
-        "segment_id": segment_id,
-        "total_users": total_users,
-        "users": enriched_users
-    }
-
-
-# ===== APPROVAL WORKFLOW =====
-@app.post("/api/events/{event_code}/segments/{segment_id}/approve", tags=["Approval"])
-async def approve_segment(
-    event_code: str,
-    segment_id: str,
-    approved_by: Optional[str] = None,
-    modified_subject: Optional[str] = None,
-    modified_body: Optional[str] = None
-):
-    """Event Owner approves marketing content"""
-    
-    segment = db.event_audience_segments.find_one({
-        "_id": ObjectId(segment_id),
-        "event_code": event_code
-    })
-    
-    if not segment:
-        raise HTTPException(status_code=404, detail="Segment not found")
-    
-    # Update fields
-    update = {
-        "marketing_content.status": "Approved",
-        "marketing_content.approved_at": datetime.utcnow(),
-        "marketing_content.approved_by": approved_by,
-        "last_updated": datetime.utcnow()
-    }
-    
-    if modified_subject:
-        update["marketing_content.email_subject"] = modified_subject
-    if modified_body:
-        update["marketing_content.email_body"] = modified_body
-    
-    db.event_audience_segments.update_one(
-        {"_id": ObjectId(segment_id)},
-        {"$set": update}
-    )
-    
-    updated_segment = db.event_audience_segments.find_one({"_id": ObjectId(segment_id)})
-    
-    return {
-        "status": "success",
-        "message": "Segment approved",
-        "segment_id": segment_id,
-        "marketing_content": updated_segment.get('marketing_content')
-    }
-
-
-@app.post("/api/events/{event_code}/segments/{segment_id}/send-email", tags=["Approval"])
-async def send_segment_email(
-    event_code: str,
-    segment_id: str,
-    send_immediately: bool = True
-):
-    """Send approved marketing email"""
-    
-    segment = db.event_audience_segments.find_one({
-        "_id": ObjectId(segment_id),
-        "event_code": event_code
-    })
-    
-    if not segment:
-        raise HTTPException(status_code=404, detail="Segment not found")
-    
-    marketing_content = segment.get('marketing_content', {})
-    if marketing_content.get('status') != "Approved":
-        raise HTTPException(status_code=400, detail="Segment not approved yet")
-    
-    # TODO: Integrate with email service (SendGrid, AWS SES, etc.)
-    # For now, just mark as sent
-    
-    db.event_audience_segments.update_one(
-        {"_id": ObjectId(segment_id)},
-        {"$set": {
-            "marketing_content.status": "Sent",
-            "last_updated": datetime.utcnow()
-        }}
-    )
-    
-    return {
-        "status": "success",
-        "message": f"Email sent to {segment.get('user_count', 0)} users",
-        "segment_id": segment_id,
-        "emails_sent": segment.get('user_count', 0),
-        "emails_failed": 0
-    }
-
-
-# ===== SENTIMENT =====
-@app.post("/api/events/{event_code}/sentiment/analyze", tags=["Sentiment"])
-async def analyze_event_sentiment(event_code: str, background_tasks: BackgroundTasks):
-    """Analyze sentiment for event comments"""
-    
-    def run_task():
-        service = SentimentAnalysisService(event_code)
-        service.analyze_event_comments()
-    
-    background_tasks.add_task(run_task)
-    
-    return {
-        "status": "started",
-        "message": f"Sentiment analysis started for event {event_code}"
-    }
-
-
-@app.get("/api/events/{event_code}/sentiment/summary", tags=["Sentiment"])
-async def get_sentiment_summary(event_code: str):
-    """Get sentiment summary for an event"""
-    
-    summary = db.event_sentiment_summary.find_one({"event_code": event_code})
-    
-    if not summary:
-        raise HTTPException(status_code=404, detail="No sentiment data for this event")
-    
-    return serialize_doc(summary)
-
-
-@app.get("/api/events/{event_code}/sentiment/results", tags=["Sentiment"])
-async def get_sentiment_results(
-    event_code: str,
-    sentiment_label: Optional[str] = None,
-    skip: int = 0,
-    limit: int = 100
-):
-    """Get detailed sentiment results"""
-    
-    query = {"event_code": event_code}
-    if sentiment_label:
-        query["sentiment_label"] = sentiment_label
-    
-    total = db.sentiment_results.count_documents(query)
-    results = list(
-        db.sentiment_results.find(query)
-        .sort("analyzed_at", -1)
-        .skip(skip)
-        .limit(limit)
-    )
-    
-    return {
-        "total": total,
-        "results": [serialize_doc(r) for r in results]
-    }
-
-
-# ===== GENAI =====
-@app.post("/api/events/{event_code}/genai/generate-emails", tags=["GenAI"])
-async def generate_event_emails(event_code: str, background_tasks: BackgroundTasks):
-    """Generate marketing emails for all segments"""
-    
-    def run_task():
-        service = GenerativeAIService(event_code)
-        service.generate_emails_for_all_segments()
-    
-    background_tasks.add_task(run_task)
-    
-    return {
-        "status": "started",
-        "message": "Email generation started"
-    }
-
-
-@app.post("/api/events/{event_code}/genai/generate-insights", tags=["GenAI"])
-async def generate_event_insights(event_code: str, background_tasks: BackgroundTasks):
-    """Generate AI insights from negative feedback"""
-    
-    def run_task():
-        service = GenerativeAIService(event_code)
-        service.update_sentiment_summary_with_insights()
-    
-    background_tasks.add_task(run_task)
-    
-    return {
-        "status": "started",
-        "message": "Insight generation started"
-    }
-
-
-# ===== MONITORING =====
-@app.get("/api/monitoring/pipelines/{pipeline}/metrics", tags=["Monitoring"])
-async def get_pipeline_metrics(
-    pipeline: str,
-    event_code: Optional[str] = None,
-    days: int = 7
-):
-    """Get performance metrics"""
-    # TODO: Implement based on monitoring.py
-    return {
-        "pipeline": pipeline,
-        "event_code": event_code,
-        "message": "Metrics endpoint - implement as needed"
-    }
-
-
-# ===== ADMIN =====
-@app.post("/api/admin/indexes/create", tags=["Admin"])
-async def create_indexes():
-    """Create MongoDB indexes"""
-    from scripts.create_indexes import create_all_indexes
-    
-    try:
-        create_all_indexes()
-        return {"status": "success", "message": "Indexes created"}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-# ===== ROOT =====
-@app.get("/")
-async def root():
-    """API root"""
-    return {
-        "name": "Audience Segmentation AI - Event-Centric",
-        "version": "2.0.0",
-        "docs": "/api/docs",
-        "health": "/health"
-    }
-
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(
-        "app:app",
-        host="0.0.0.0",
-        port=7860,
-        reload=False
-    )
+"""
+FastAPI Application for Event-Centric Audience Segmentation AI
+Author: AI Generated
+Created: 2025-11-24 (Refactored)
+Purpose: REST API with event-based endpoints
+"""
+
+from fastapi import FastAPI, HTTPException, BackgroundTasks, status, Query
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import List, Dict, Optional, Any
+from datetime import datetime
+from bson import ObjectId
+
+# Import services
+from services.segmentation_service import SegmentationService
+from services.sentiment_service import SentimentAnalysisService
+from services.genai_service import GenerativeAIService
+from database import db
+from config import settings
+
+
+# FastAPI app
+app = FastAPI(
+    title="Audience Segmentation AI - Event-Centric",
+    description="REST API for per-event audience analysis",
+    version="2.0.0",
+    docs_url="/api/docs",
+    redoc_url="/api/redoc"
+)
+
+# CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# Helper
+def serialize_doc(doc: Dict) -> Optional[Dict]:
+    """Convert MongoDB document to JSON-serializable dict"""
+    if doc is None:
+        return None
+    if '_id' in doc:
+        doc['id'] = str(doc.pop('_id'))
+    
+    # Handle nested ObjectIds and lists
+    for key, value in list(doc.items()):
+        if isinstance(value, ObjectId):
+            doc[key] = str(value)
+        elif isinstance(value, list):
+            doc[key] = [str(v) if isinstance(v, ObjectId) else v for v in value]
+        elif isinstance(value, dict):
+            doc[key] = serialize_doc(value)
+    
+    return doc
+
+
+# ===== HEALTH =====
+@app.get("/health", tags=["System"])
+async def health_check():
+    """Health check"""
+    try:
+        db.client.server_info()
+        return {
+            "status": "healthy",
+            "timestamp": datetime.utcnow(),
+            "database": "connected"
+        }
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"Unhealthy: {str(e)}"
+        )
+
+
+# ===== EVENT ANALYSIS =====
+@app.post("/api/events/{event_code}/analyze", tags=["Event Analysis"])
+async def analyze_event(event_code: str, background_tasks: BackgroundTasks):
+    """Run full AI pipeline for an event"""
+    
+    def run_pipeline():
+        # Step 1: Segmentation
+        seg_service = SegmentationService(event_code)
+        seg_service.run_segmentation()
+        
+        # Step 2: Sentiment
+        sent_service = SentimentAnalysisService(event_code)
+        sent_service.analyze_event_comments()
+        
+        # Step 3: Email generation
+        genai_service = GenerativeAIService(event_code)
+        genai_service.generate_emails_for_all_segments()
+        
+        # Step 4: Insights
+        genai_service.update_sentiment_summary_with_insights()
+    
+    background_tasks.add_task(run_pipeline)
+    
+    return {
+        "status": "started",
+        "message": f"Analysis pipeline started for event {event_code}"
+    }
+
+
+@app.get("/api/events/{event_code}/dashboard", tags=["Event Analysis"])
+async def get_event_dashboard(event_code: str):
+    """Get complete dashboard for Event Owner"""
+    
+    # Get segments
+    segments = list(db.event_audience_segments.find({"event_code": event_code}))
+    
+    # Get sentiment summary
+    sentiment_summary = db.event_sentiment_summary.find_one({"event_code": event_code})
+    
+    return {
+        "event_code": event_code,
+        "segments": [serialize_doc(s) for s in segments],
+        "sentiment_summary": serialize_doc(sentiment_summary) if sentiment_summary else None
+    }
+
+
+# ===== SEGMENTATION =====
+@app.post("/api/events/{event_code}/segmentation/run", tags=["Segmentation"])
+async def run_event_segmentation(
+    event_code: str,
+    background_tasks: BackgroundTasks,
+    n_clusters: int = Query(default=5, ge=2, le=10)
+):
+    """Run segmentation for an event"""
+    
+    def run_task():
+        service = SegmentationService(event_code, n_clusters=n_clusters)
+        service.run_segmentation()
+    
+    background_tasks.add_task(run_task)
+    
+    return {
+        "status": "started",
+        "message": f"Segmentation started for event {event_code}",
+        "event_code": event_code
+    }
+
+
+@app.get("/api/events/{event_code}/segments", tags=["Segmentation"])
+async def get_event_segments(
+    event_code: str,
+    status_filter: Optional[str] = Query(default=None, description="Filter by Draft, Approved, Sent")
+):
+    """Get all segments for an event"""
+    
+    query = {"event_code": event_code}
+    if status_filter:
+        query["marketing_content.status"] = status_filter
+    
+    segments = list(db.event_audience_segments.find(query))
+    
+    return [serialize_doc(s) for s in segments]
+
+
+@app.get("/api/events/{event_code}/segments/{segment_id}", tags=["Segmentation"])
+async def get_segment_detail(event_code: str, segment_id: str):
+    """Get specific segment details"""
+    
+    segment = db.event_audience_segments.find_one({
+        "_id": ObjectId(segment_id),
+        "event_code": event_code
+    })
+    
+    if not segment:
+        raise HTTPException(status_code=404, detail="Segment not found")
+    
+    return serialize_doc(segment)
+
+
+@app.get("/api/events/{event_code}/segments/{segment_id}/users", tags=["Segmentation"])
+async def get_segment_users(
+    event_code: str,
+    segment_id: str,
+    skip: int = 0,
+    limit: int = 100
+):
+    """Get users in a segment with details"""
+    
+    segment = db.event_audience_segments.find_one({
+        "_id": ObjectId(segment_id),
+        "event_code": event_code
+    })
+    
+    if not segment:
+        raise HTTPException(status_code=404, detail="Segment not found")
+    
+    user_ids = segment.get('user_ids', [])
+    total_users = len(user_ids)
+    
+    # Paginate
+    paginated_ids = user_ids[skip:skip + limit]
+    
+    # Get user details
+    users = list(db.users.find({
+        "_id": {"$in": paginated_ids}
+    }))
+    
+    # Enrich with stats (optional)
+    enriched_users = []
+    for user in users:
+        enriched_users.append({
+            "user_id": str(user['_id']),
+            "email": user.get('email'),
+            "full_name": f"{user.get('FirstName', '')} {user.get('LastName', '')}".strip()
+        })
+    
+    return {
+        "segment_id": segment_id,
+        "total_users": total_users,
+        "users": enriched_users
+    }
+
+
+# ===== APPROVAL WORKFLOW =====
+@app.post("/api/events/{event_code}/segments/{segment_id}/approve", tags=["Approval"])
+async def approve_segment(
+    event_code: str,
+    segment_id: str,
+    approved_by: Optional[str] = None,
+    modified_subject: Optional[str] = None,
+    modified_body: Optional[str] = None
+):
+    """Event Owner approves marketing content"""
+    
+    segment = db.event_audience_segments.find_one({
+        "_id": ObjectId(segment_id),
+        "event_code": event_code
+    })
+    
+    if not segment:
+        raise HTTPException(status_code=404, detail="Segment not found")
+    
+    # Update fields
+    update = {
+        "marketing_content.status": "Approved",
+        "marketing_content.approved_at": datetime.utcnow(),
+        "marketing_content.approved_by": approved_by,
+        "last_updated": datetime.utcnow()
+    }
+    
+    if modified_subject:
+        update["marketing_content.email_subject"] = modified_subject
+    if modified_body:
+        update["marketing_content.email_body"] = modified_body
+    
+    db.event_audience_segments.update_one(
+        {"_id": ObjectId(segment_id)},
+        {"$set": update}
+    )
+    
+    updated_segment = db.event_audience_segments.find_one({"_id": ObjectId(segment_id)})
+    
+    return {
+        "status": "success",
+        "message": "Segment approved",
+        "segment_id": segment_id,
+        "marketing_content": updated_segment.get('marketing_content')
+    }
+
+
+@app.post("/api/events/{event_code}/segments/{segment_id}/send-email", tags=["Approval"])
+async def send_segment_email(
+    event_code: str,
+    segment_id: str,
+    send_immediately: bool = True
+):
+    """Send approved marketing email"""
+    
+    segment = db.event_audience_segments.find_one({
+        "_id": ObjectId(segment_id),
+        "event_code": event_code
+    })
+    
+    if not segment:
+        raise HTTPException(status_code=404, detail="Segment not found")
+    
+    marketing_content = segment.get('marketing_content', {})
+    if marketing_content.get('status') != "Approved":
+        raise HTTPException(status_code=400, detail="Segment not approved yet")
+    
+    # TODO: Integrate with email service (SendGrid, AWS SES, etc.)
+    # For now, just mark as sent
+    
+    db.event_audience_segments.update_one(
+        {"_id": ObjectId(segment_id)},
+        {"$set": {
+            "marketing_content.status": "Sent",
+            "last_updated": datetime.utcnow()
+        }}
+    )
+    
+    return {
+        "status": "success",
+        "message": f"Email sent to {segment.get('user_count', 0)} users",
+        "segment_id": segment_id,
+        "emails_sent": segment.get('user_count', 0),
+        "emails_failed": 0
+    }
+
+
+# ===== SENTIMENT =====
+@app.post("/api/events/{event_code}/sentiment/analyze", tags=["Sentiment"])
+async def analyze_event_sentiment(event_code: str, background_tasks: BackgroundTasks):
+    """Analyze sentiment for event comments"""
+    
+    def run_task():
+        service = SentimentAnalysisService(event_code)
+        service.analyze_event_comments()
+    
+    background_tasks.add_task(run_task)
+    
+    return {
+        "status": "started",
+        "message": f"Sentiment analysis started for event {event_code}"
+    }
+
+
+@app.get("/api/events/{event_code}/sentiment/summary", tags=["Sentiment"])
+async def get_sentiment_summary(event_code: str):
+    """Get sentiment summary for an event"""
+    
+    summary = db.event_sentiment_summary.find_one({"event_code": event_code})
+    
+    if not summary:
+        raise HTTPException(status_code=404, detail="No sentiment data for this event")
+    
+    return serialize_doc(summary)
+
+
+@app.get("/api/events/{event_code}/sentiment/results", tags=["Sentiment"])
+async def get_sentiment_results(
+    event_code: str,
+    sentiment_label: Optional[str] = None,
+    skip: int = 0,
+    limit: int = 100
+):
+    """Get detailed sentiment results"""
+    
+    query = {"event_code": event_code}
+    if sentiment_label:
+        query["sentiment_label"] = sentiment_label
+    
+    total = db.sentiment_results.count_documents(query)
+    results = list(
+        db.sentiment_results.find(query)
+        .sort("analyzed_at", -1)
+        .skip(skip)
+        .limit(limit)
+    )
+    
+    return {
+        "total": total,
+        "results": [serialize_doc(r) for r in results]
+    }
+
+
+# ===== GENAI =====
+@app.post("/api/events/{event_code}/genai/generate-emails", tags=["GenAI"])
+async def generate_event_emails(event_code: str, background_tasks: BackgroundTasks):
+    """Generate marketing emails for all segments"""
+    
+    def run_task():
+        service = GenerativeAIService(event_code)
+        service.generate_emails_for_all_segments()
+    
+    background_tasks.add_task(run_task)
+    
+    return {
+        "status": "started",
+        "message": "Email generation started"
+    }
+
+
+@app.post("/api/events/{event_code}/genai/generate-insights", tags=["GenAI"])
+async def generate_event_insights(event_code: str, background_tasks: BackgroundTasks):
+    """Generate AI insights from negative feedback"""
+    
+    def run_task():
+        service = GenerativeAIService(event_code)
+        service.update_sentiment_summary_with_insights()
+    
+    background_tasks.add_task(run_task)
+    
+    return {
+        "status": "started",
+        "message": "Insight generation started"
+    }
+
+
+# ===== MONITORING =====
+@app.get("/api/monitoring/pipelines/{pipeline}/metrics", tags=["Monitoring"])
+async def get_pipeline_metrics(
+    pipeline: str,
+    event_code: Optional[str] = None,
+    days: int = 7
+):
+    """Get performance metrics"""
+    # TODO: Implement based on monitoring.py
+    return {
+        "pipeline": pipeline,
+        "event_code": event_code,
+        "message": "Metrics endpoint - implement as needed"
+    }
+
+
+# ===== ADMIN =====
+@app.post("/api/admin/indexes/create", tags=["Admin"])
+async def create_indexes():
+    """Create MongoDB indexes"""
+    from scripts.create_indexes import create_all_indexes
+    
+    try:
+        create_all_indexes()
+        return {"status": "success", "message": "Indexes created"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+# ===== ROOT =====
+@app.get("/")
+async def root():
+    """API root"""
+    return {
+        "name": "Audience Segmentation AI - Event-Centric",
+        "version": "2.0.0",
+        "docs": "/api/docs",
+        "health": "/health"
+    }
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "app:app",
+        host="0.0.0.0",
+        port=7860,
+        reload=True
+    )

config.py

@@ -31,9 +31,8 @@ class Settings(BaseSettings):
     # AI Models
     SENTIMENT_MODEL: str = "wonrax/phobert-base-vietnamese-sentiment"
     
-    # Vistral LLM (Auto-download via Transformers)
+    # Vistral LLM (Via HuggingFace Inference API)
     LLM_MODEL_NAME: str = os.getenv("LLM_MODEL_NAME", "Viet-Mistral/Vistral-7B-Chat")
-    LLM_CACHE_DIR: str = os.getenv("LLM_CACHE_DIR", "./models/cache")
     
     # Clustering
     N_CLUSTERS: int = 5

services/genai_service.py

@@ -1,13 +1,12 @@
 """
 Event-Centric Generative AI Service
 Author: AI Generated
-Created: 2025-11-24 (Refactored with Transformers)
-Purpose: Generate marketing content and insights using Vistral-7B-Chat
+Created: 2025-11-24 (Using HuggingFace Inference API)
+Purpose: Generate marketing content using Vistral-7B-Chat via API
 """
 
-import torch
-from transformers import AutoModelForCausalLM, AutoTokenizer
-from typing import Dict, List, Optional
+from huggingface_hub import InferenceClient
+from typing import Dict, List
 from datetime import datetime
 from bson import ObjectId
 
@@ -15,12 +14,12 @@ from database import db
 from config import settings
 from models.event_models import EventSentimentSummary, AIInsights, MarketingContent
 from services.monitoring import monitor
-from services.model_registry import registry
 
 
 class GenerativeAIService:
     """
-    Event-centric GenAI using Vistral-7B-Chat via Transformers.
+    Event-centric GenAI using Vistral-7B-Chat via HuggingFace Inference API.
+    Much faster and lighter than loading model locally.
     """
     
     def __init__(self, event_code: str):
@@ -32,10 +31,7 @@ class GenerativeAIService:
         """
         self.event_code = event_code
         self.model_name = settings.LLM_MODEL_NAME
-        self.cache_dir = settings.LLM_CACHE_DIR
-        self.tokenizer = None
-        self.model = None
-        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.client = None
         
         # System prompt theo official docs
         self.system_prompt = (
@@ -45,99 +41,63 @@ class GenerativeAIService:
             "phân biệt chủng tộc, phân biệt giới tính, độc hại, nguy hiểm hoặc bất hợp pháp nào."
         )
     
-    def load_model(self):
-        """
-        Load Vistral-7B-Chat model using Transformers.
-        Auto-downloads from HuggingFace Hub.
-        """
-        try:
-            print(f"🔄 Loading Vistral-7B-Chat: {self.model_name}")
-            print(f"   Device: {self.device}")
-            print(f"   Cache: {self.cache_dir}")
+    def get_client(self):
+        """Get or create InferenceClient"""
+        if not self.client:
+            print(f"🔄 Initializing HuggingFace Inference API")
+            print(f"   Model: {self.model_name}")
             
-            # Load tokenizer
-            self.tokenizer = AutoTokenizer.from_pretrained(
-                self.model_name,
-                cache_dir=self.cache_dir,
+            self.client = InferenceClient(
+                model=self.model_name,
                 token=settings.HF_TOKEN if settings.HF_TOKEN else None
             )
-            
-            # Load model with appropriate dtype
-            dtype = torch.float16 if self.device == "cuda" else torch.float32
-            
-            print(f"   Loading model with dtype={dtype}...")
-            self.model = AutoModelForCausalLM.from_pretrained(
-                self.model_name,
-                torch_dtype=dtype,
-                device_map="auto" if self.device == "cuda" else None,
-                cache_dir=self.cache_dir,
-                token=settings.HF_TOKEN if settings.HF_TOKEN else None,
-                use_cache=True,
-                low_cpu_mem_usage=True
-            )
-            
-            # Move to device if CPU
-            if self.device == "cpu":
-                self.model = self.model.to(self.device)
-            
-            self.model.eval()
-            print("✓ Model loaded successfully!")
-            
-        except Exception as e:
-            print(f"❌ Error loading model: {str(e)}")
-            print(f"⚠️ GenAI features will be disabled.")
-            print(f"   Segmentation and Sentiment analysis will still work.")
-            self.model = None
-            self.tokenizer = None
+            print("✓ Inference client ready!")
+        
+        return self.client
     
     def generate_text(self, prompt: str, max_new_tokens: int = 512) -> str:
         """
-        Generate text using Vistral with chat template.
-        """
-        if not self.model or not self.tokenizer:
-            self.load_model()
-        
-        if not self.model:
-            return ""
-        
-        # Build conversation with system prompt
-        conversation = [
-            {"role": "system", "content": self.system_prompt},
-            {"role": "user", "content": prompt}
-        ]
-        
-        # Apply chat template
-        input_ids = self.tokenizer.apply_chat_template(
-            conversation,
-            return_tensors="pt"
-        ).to(self.model.device)
+        Generate text using Vistral via HuggingFace Inference API.
         
-        # Generate
-        with torch.no_grad():
-            output_ids = self.model.generate(
-                input_ids=input_ids,
-                max_new_tokens=max_new_tokens,
-                do_sample=True,
-                top_p=0.95,
-                top_k=40,
+        Args:
+            prompt: User prompt
+            max_new_tokens: Max tokens to generate
+            
+        Returns:
+            Generated text
+        """
+        try:
+            client = self.get_client()
+            
+            # Build messages with system prompt
+            messages = [
+                {"role": "system", "content": self.system_prompt},
+                {"role": "user", "content": prompt}
+            ]
+            
+            # Call Inference API
+            response = client.chat_completion(
+                messages=messages,
+                max_tokens=max_new_tokens,
                 temperature=0.7,
-                repetition_penalty=1.05,
+                top_p=0.95,
             )
+            
+            # Extract generated text
+            generated = response.choices[0].message.content
+            return generated.strip()
         
-        # Decode (only new tokens)
-        generated_text = self.tokenizer.batch_decode(
-            output_ids[:, input_ids.size(1):],
-            skip_special_tokens=True
-        )[0].strip()
-        
-        return generated_text
+        except Exception as e:
+            print(f"❌ Error calling Inference API: {str(e)}")
+            print(f"⚠️ Returning empty response")
+            return ""
     
     def generate_email_for_segment(self, segment: Dict) -> MarketingContent:
         """
         Generate personalized email for a segment.
         """
         # Get event info
-        event = db.event_versions.find_one({"_id": self.event_code})
+        event = db.event_versions.find_one({"_id": ObjectId(self.event_code)})
         event_name = event.get("EventName", "Sự kiện") if event else "Sự kiện"
         
         # Build prompt
@@ -161,6 +121,14 @@ BODY:
         
         generated = self.generate_text(prompt, max_new_tokens=400)
         
+        if not generated:
+            return MarketingContent(
+                email_subject=f"Ưu đãi đặc biệt cho {segment['segment_name']}",
+                email_body="Nội dung email sẽ được tạo khi API khả dụng.",
+                status="Draft",
+                generated_at=datetime.utcnow()
+            )
+        
         # Parse response
         lines = generated.split('\n')
         subject = ""
@@ -199,13 +167,6 @@ BODY:
         print("=" * 60)
         
         try:
-            if not self.model:
-                self.load_model()
-            
-            if not self.model:
-                print("⚠️ Model not available, skipping email generation")
-                return
-            
             # Find segments without marketing content
             segments = list(db.event_audience_segments.find({
                 "event_code": self.event_code,
@@ -294,6 +255,14 @@ BODY:
         comments = list(db.post_social_media.aggregate(pipeline))
         negative_texts = [c.get('comment_text', '') for c in comments if c.get('comment_text')]
         
+        if not negative_texts:
+            return AIInsights(
+                summary="Không thể truy xuất nội dung feedback tiêu cực.",
+                top_issues=[],
+                improvement_suggestions=[],
+                predicted_nps=60.0
+            )
+        
         # Build prompt
         comments_sample = "\n".join([f"- {text[:100]}" for text in negative_texts[:15]])
         
@@ -331,32 +300,32 @@ NPS: [số]
         suggestions = []
         predicted_nps = 60.0
         
-        lines = generated.split('\n')
-        current_section = None
-        
-        for line in lines:
-            line = line.strip()
-            if "TOP_ISSUES" in line or "VẤN ĐỀ" in line:
-                current_section = "issues"
-            elif "SUGGESTIONS" in line or "ĐỀ XUẤT" in line:
-                current_section = "suggestions"
-            elif "NPS" in line:
-                try:
-                    # Extract number from line
-                    import re
-                    numbers = re.findall(r'\d+', line)
-                    if numbers:
-                        predicted_nps = float(numbers[0])
-                except:
-                    pass
-            elif current_section == "issues" and (line.startswith("-") or line[0].isdigit()):
-                issue = line.lstrip("0123456789.-) ").strip()
-                if issue and len(issue) > 5:
-                    top_issues.append(issue)
-            elif current_section == "suggestions" and line.startswith("-"):
-                suggestion = line.lstrip("- ").strip()
-                if suggestion and len(suggestion) > 5:
-                    suggestions.append(suggestion)
+        if generated:
+            lines = generated.split('\n')
+            current_section = None
+            
+            for line in lines:
+                line = line.strip()
+                if "TOP_ISSUES" in line or "VẤN ĐỀ" in line:
+                    current_section = "issues"
+                elif "SUGGESTIONS" in line or "ĐỀ XUẤT" in line:
+                    current_section = "suggestions"
+                elif "NPS" in line:
+                    try:
+                        import re
+                        numbers = re.findall(r'\d+', line)
+                        if numbers:
+                            predicted_nps = float(numbers[0])
+                    except:
+                        pass
+                elif current_section == "issues" and (line.startswith("-") or line[0].isdigit()):
+                    issue = line.lstrip("0123456789.-) ").strip()
+                    if issue and len(issue) > 5:
+                        top_issues.append(issue)
+                elif current_section == "suggestions" and line.startswith("-"):
+                    suggestion = line.lstrip("- ").strip()
+                    if suggestion and len(suggestion) > 5:
+                        suggestions.append(suggestion)
         
         # Create summary
         total_comments = db.sentiment_results.count_documents({"event_code": self.event_code})