app.py

# app.py
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import logging
import numpy as np
from typing import List, Optional, Dict, Any, Union
import sys
import os

# Import the InterestClassifier from your model file
# Make sure this file is in the same directory as app.py
from hybrid_interest_classifier import InterestClassifier, INTEREST_CATEGORIES

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

# Initialize FastAPI app
app = FastAPI()

# Allow CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Define keyword-based interest detection as fallback
def keyword_interests(text):
    """

    Determine interests using keyword matching as a fallback

    """
    text = text.lower()
    interests = []
    
    if any(word in text for word in ['music', 'band', 'concert', 'sing', 'guitar', 'song']):
        interests.append('Music')
    
    if any(word in text for word in ['food', 'cook', 'recipe', 'restaurant', 'eat', 'cuisine']):
        interests.append('Food')
    
    if any(word in text for word in ['sport', 'gym', 'fitness', 'exercise', 'workout', 'run']):
        interests.append('Sports')
    
    if any(word in text for word in ['art', 'paint', 'draw', 'gallery', 'museum', 'exhibition']):
        interests.append('Arts')
    
    if any(word in text for word in ['tech', 'code', 'software', 'computer', 'programming']):
        interests.append('Technology')
    
    if any(word in text for word in ['learn', 'study', 'course', 'book', 'read', 'class']):
        interests.append('Education')
    
    if any(word in text for word in ['travel', 'trip', 'journey', 'explore', 'hike', 'tourism']):
        interests.append('Travel')
    
    if not interests:
        interests.append('No specific interests detected')
    
    return interests

# Load the hybrid classifier
MODEL_PATH = "hybrid_interest_classifier.pkl"
hybrid_classifier = None

try:
    logger.info(f"Loading hybrid model from {MODEL_PATH}")
    # Create an instance of our classifier and load the model
    hybrid_classifier = InterestClassifier(model_path=MODEL_PATH)
    logger.info("Hybrid model loaded successfully")
    
    # Log if BERT is available
    if hybrid_classifier.bert_classifier is not None:
        logger.info("BERT zero-shot classifier initialized and ready")
    else:
        logger.warning("BERT zero-shot classifier is not available, will use TF-IDF only")
except Exception as e:
    logger.error(f"Failed to load hybrid model: {e}")

# Pydantic models
class PredictionRequest(BaseModel):
    text: str
    alpha: Optional[float] = None
    threshold: Optional[float] = None
    return_scores: Optional[bool] = False

class ModelConfigRequest(BaseModel):
    alpha: Optional[float] = None
    threshold: Optional[float] = None

@app.get("/")
async def root():
    """Root endpoint to check if API is running"""
    return {
        "status": "online", 
        "message": "Hybrid Interest Classifier API is running",
        "model_loaded": hybrid_classifier is not None,
        "bert_available": hybrid_classifier.bert_classifier is not None if hybrid_classifier else False
    }

@app.get("/health")
async def health():
    """Health check endpoint"""
    return {
        "status": "healthy", 
        "model_loaded": hybrid_classifier is not None,
        "bert_available": hybrid_classifier.bert_classifier is not None if hybrid_classifier else False
    }

@app.post("/config")
async def update_config(config: ModelConfigRequest):
    """Update model configuration"""
    if hybrid_classifier is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    changes = {}
    
    if config.alpha is not None:
        hybrid_classifier.alpha = float(config.alpha)
        changes["alpha"] = hybrid_classifier.alpha
    
    if config.threshold is not None:
        hybrid_classifier.threshold = float(config.threshold)
        changes["threshold"] = hybrid_classifier.threshold
    
    return {
        "message": "Configuration updated successfully",
        "changes": changes,
        "current_config": {
            "alpha": hybrid_classifier.alpha,
            "threshold": hybrid_classifier.threshold,
            "bert_available": hybrid_classifier.bert_classifier is not None
        }
    }

@app.post("/predict")
async def predict(request: PredictionRequest):
    """

    Predict interests based on text input

    """
    text = request.text
    alpha = request.alpha
    threshold = request.threshold
    return_scores = request.return_scores
    
    logger.info(f"Prediction request: text='{text[:50]}...', alpha={alpha}, threshold={threshold}, return_scores={return_scores}")
    
    if not text or text.strip() == "":
        return {"labels": ["No text provided"], "text": text}
    
    if hybrid_classifier is None:
        logger.warning("Using fallback keyword matching (model not loaded)")
        return {"labels": keyword_interests(text), "text": text}
    
    try:
        # Prepare prediction parameters
        kwargs = {}
        if alpha is not None:
            kwargs['alpha'] = alpha
        if threshold is not None:
            kwargs['threshold'] = threshold
        if return_scores:
            kwargs['return_scores'] = True
        
        # Log the call we're about to make
        logger.info(f"Calling hybrid_classifier.predict([{text[:20]}...], {kwargs})")
        
        # Make prediction
        prediction = hybrid_classifier.predict(text, **kwargs)
        
        logger.info(f"Raw prediction type: {type(prediction)}")
        
        # Process the prediction result
        labels = []
        scores = {}
        
        # Handle dictionary return type (with return_scores=True)
        if isinstance(prediction, dict):
            labels = prediction.get('labels', [])
            
            # Include detailed information in response if available
            if return_scores:
                response = {
                    "labels": labels,
                    "text": text,
                    "scores": dict(prediction.get('sorted_scores', [])),
                    "model_info": {
                        "alpha": prediction.get('alpha', hybrid_classifier.alpha),
                        "threshold": prediction.get('threshold', hybrid_classifier.threshold),
                        "using_bert": prediction.get('using_bert', False)
                    }
                }
                
                # Add timing information if available
                if 'timing' in prediction:
                    response["timing"] = prediction['timing']
                
                # Include individual model scores
                if 'tfidf_scores' in prediction:
                    response["tfidf_scores"] = dict(sorted(
                        prediction['tfidf_scores'].items(), 
                        key=lambda x: x[1], 
                        reverse=True
                    )[:5])
                    
                if 'bert_scores' in prediction:
                    response["bert_scores"] = dict(sorted(
                        prediction['bert_scores'].items(), 
                        key=lambda x: x[1], 
                        reverse=True
                    )[:5])
                    
                return response
        
        # Handle list return type
        elif isinstance(prediction, list):
            labels = prediction
        
        # If we still have no labels, use keyword matching
        if not labels:
            logger.warning("No labels detected, using fallback")
            labels = keyword_interests(text)
        
        # Return simple response without scores
        return {"labels": labels, "text": text}
        
    except Exception as e:
        logger.error(f"Error during prediction: {e}", exc_info=True)
        return {"labels": keyword_interests(text), "text": text, "error": str(e)}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)