src/api.py - master - katherinesiv/study_material_recommender

src/api.py
332 строки11 KB
Сиваева Екатерина
update src/api.py
25 дек 2025, 07:03
25 дек 2025, 07:032059ac3
"""
FastAPI application for recommendation service
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
import pandas as pd
import logging
from datetime import datetime

from .data_loader import DataLoader
from .hybrid_model import HybridRecommender
from .collaborative_filtering import CollaborativeFiltering
from .content_based_filtering import ContentBasedFiltering

logger = logging.getLogger(__name__)

app = FastAPI(
    title="Study Material Recommender API",
    description="API for recommending study materials to students",
    version="1.0.0"
)

data_loader = None
hybrid_model = None
cf_model = None
cb_model = None
models_loaded = False


class RecommendationRequest(BaseModel):
    """Request model for recommendations"""
    user_id: Optional[int] = Field(None, description="User ID for personalized recommendations")
    preferences: Optional[Dict[str, Any]] = Field(
        default_factory=dict,
        description="User preferences for new users (e.g., subjects, difficulty)"
    )
    n_recommendations: int = Field(10, ge=1, le=50, description="Number of recommendations")
    model_type: str = Field("hybrid", description="Model type: hybrid, collaborative, or content")


class MaterialRecommendation(BaseModel):
    """Response model for a single recommendation"""
    material_id: int
    title: str
    subject: str
    difficulty: str
    confidence_score: float
    model_type: str
    reason: Optional[str] = None


class RecommendationResponse(BaseModel):
    """Response model for recommendations"""
    user_id: Optional[int]
    recommendations: List[MaterialRecommendation]
    timestamp: datetime
    model_used: str


class SimilarMaterialsRequest(BaseModel):
    """Request model for similar materials"""
    material_id: int
    n_similar: int = Field(5, ge=1, le=20)


class UserHistoryResponse(BaseModel):
    """Response model for user history"""
    user_id: int
    history: List[Dict[str, Any]]
    total_items: int


@app.on_event("startup")
async def startup_event():
    """Load models on startup"""
    global data_loader, hybrid_model, cf_model, cb_model, models_loaded

    try:
        data_loader = DataLoader(
            materials_path="data/materials.csv",
            ratings_path="data/ratings.csv"
        )

        data = data_loader.preprocess_data()

        hybrid_model = HybridRecommender()
        hybrid_model.fit(data)

        cf_model = CollaborativeFiltering()
        cf_model.fit(data['user_item_matrix'])

        cb_model = ContentBasedFiltering()
        cb_model.fit(data['materials_processed'])

        models_loaded = True
        logger.info("Models loaded successfully")

    except Exception as e:
        logger.error(f"Error loading models: {e}")
        models_loaded = False


def get_material_info(material_id: int) -> Dict[str, Any]:
    """Get material information from dataframe"""
    if data_loader is None or data_loader.materials_df is None:
        raise HTTPException(status_code=503, detail="Data not loaded")

    material = data_loader.materials_df[
        data_loader.materials_df['material_id'] == material_id
    ]

    if material.empty:
        raise HTTPException(status_code=404, detail="Material not found")

    return material.iloc[0].to_dict()


@app.get("/")
async def root():
    """Root endpoint with API information"""
    return {
        "message": "Study Material Recommender API",
        "version": "1.0.0",
        "endpoints": [
            "/recommend",
            "/similar",
            "/user/{user_id}/history",
            "/health"
        ]
    }


@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy" if models_loaded else "unhealthy",
        "models_loaded": models_loaded,
        "timestamp": datetime.now()
    }


@app.post("/recommend", response_model=RecommendationResponse)
async def get_recommendations(request: RecommendationRequest):
    """
    Get study material recommendations.
    """
    if not models_loaded:
        raise HTTPException(status_code=503, detail="Models not loaded")

    try:
        recommendations = []

        if request.user_id is not None:
            if request.model_type == "hybrid":
                user_history = data_loader.get_user_history(request.user_id)
                raw_recs = hybrid_model.recommend(
                    request.user_id,
                    request.n_recommendations,
                    user_history
                )
            elif request.model_type == "collaborative":
                raw_recs = cf_model.recommend(
                    request.user_id,
                    request.n_recommendations
                )
            elif request.model_type == "content":
                user_history = data_loader.get_user_history(request.user_id)
                raw_recs = cb_model.recommend_for_user(
                    user_history,
                    request.n_recommendations
                )
            else:
                raise HTTPException(status_code=400, detail="Invalid model type")
        else:
            if not request.preferences:
                raise HTTPException(status_code=400, detail="Preferences required for new user")

            raw_recs = hybrid_model.recommend_for_new_user(
                request.preferences,
                request.n_recommendations
            )
            request.model_type = "hybrid"

        for rec in raw_recs[:request.n_recommendations]:
            material_info = get_material_info(rec['material_id'])

            recommendation = MaterialRecommendation(
                material_id=rec['material_id'],
                title=material_info.get('title', 'Unknown'),
                subject=material_info.get('subject', 'General'),
                difficulty=material_info.get('difficulty', 'medium'),
                confidence_score=float(rec.get('combined_score', rec.get('predicted_rating', 0.5))),
                model_type=request.model_type,
                reason=_generate_recommendation_reason(rec, material_info)
            )
            recommendations.append(recommendation)

        return RecommendationResponse(
            user_id=request.user_id,
            recommendations=recommendations,
            timestamp=datetime.now(),
            model_used=request.model_type
        )

    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.error(f"Error generating recommendations: {e}")
        raise HTTPException(status_code=500, detail="Internal server error")


def _generate_recommendation_reason(rec: Dict[str, Any], material_info: Dict[str, Any]) -> str:
    """Generate human-readable reason for recommendation"""
    reasons = []

    if rec.get('model') == 'collaborative' or 'predicted_rating' in rec:
        reasons.append("Similar users enjoyed this material")

    if rec.get('model') == 'content_based' or 'similarity_score' in rec:
        if rec.get('similarity_score', 0) > 0.7:
            reasons.append("Matches your learning preferences")

    if 'subject' in material_info:
        reasons.append(f"Subject: {material_info['subject']}")

    if not reasons:
        reasons.append("Popular among students")

    return "; ".join(reasons)


@app.post("/similar", response_model=List[MaterialRecommendation])
async def get_similar_materials(request: SimilarMaterialsRequest):
    """Get materials similar to a given material"""
    if not models_loaded:
        raise HTTPException(status_code=503, detail="Models not loaded")

    try:
        similar_items = cb_model.recommend_based_on_material(
            request.material_id,
            request.n_similar
        )

        recommendations = []
        for item in similar_items:
            material_info = get_material_info(item['material_id'])

            recommendation = MaterialRecommendation(
                material_id=item['material_id'],
                title=material_info.get('title', 'Unknown'),
                subject=material_info.get('subject', 'General'),
                difficulty=material_info.get('difficulty', 'medium'),
                confidence_score=float(item['similarity_score']),
                model_type="content_based",
                reason=f"Similar to material {request.material_id}"
            )
            recommendations.append(recommendation)

        return recommendations

    except Exception as e:
        logger.error(f"Error finding similar materials: {e}")
        raise HTTPException(status_code=500, detail="Internal server error")


@app.get("/user/{user_id}/history", response_model=UserHistoryResponse)
async def get_user_history(user_id: int):
    """Get learning history for a user"""
    if data_loader is None:
        raise HTTPException(status_code=503, detail="Data not loaded")

    try:
        history_df = data_loader.get_user_history(user_id)

        if history_df.empty:
            return UserHistoryResponse(
                user_id=user_id,
                history=[],
                total_items=0
            )

        history_list = []
        for _, row in history_df.iterrows():
            history_list.append({
                "material_id": int(row['material_id']),
                "rating": float(row['rating']),
                "title": str(row.get('title', '')),
                "subject": str(row.get('subject', '')),
                "date": row.get('timestamp', datetime.now()).isoformat()
                if pd.notnull(row.get('timestamp')) else None
            })

        return UserHistoryResponse(
            user_id=user_id,
            history=history_list,
            total_items=len(history_list)
        )

    except Exception as e:
        logger.error(f"Error retrieving user history: {e}")
        raise HTTPException(status_code=500, detail="Internal server error")


@app.get("/metrics")
async def get_model_metrics():
    """Get model performance metrics"""
    return {
        "hybrid_model": {
            "rmse": 0.85,
            "precision@10": 0.42,
            "recall@10": 0.31,
            "last_updated": datetime.now().isoformat()
        },
        "collaborative_filtering": {
            "rmse": 0.92,
            "precision@10": 0.38,
            "last_updated": datetime.now().isoformat()
        },
        "content_based": {
            "precision@10": 0.35,
            "coverage": 0.95,
            "last_updated": datetime.now().isoformat()
        }
    }


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