utils/scoring.py

from typing import List, Dict, Any
import math
import logging
import time
from config.settings import get_settings

logger = logging.getLogger(__name__)

def rerank_mentors(
    similar_mentors: List[Dict[str, Any]],
    mentee_data: Dict[str, Any],
    final_count: int = 8
) -> List[Dict[str, Any]]:
    start_time = time.perf_counter()
    logger.info(f"[SCORING] Starting final scoring for {len(similar_mentors)} mentors, final_count={final_count}")
    
    settings = get_settings()
    scored_mentors = []
    
    scoring_start = time.perf_counter()
    for mentor in similar_mentors:
        metadata = mentor.get("metadata", {})
        
        reranker_score = mentor.get("reranker_score")
        if reranker_score is None:
            reranker_score = mentor.get("score", 0.0)
        
        semantic_score = reranker_score * settings.SEMANTIC_WEIGHT
        
        rating = metadata.get("rating", 0.0)
        rating_score = _calculate_rating_score(rating)
        
        rule_based_score = rating_score * settings.RULE_BASED_WEIGHT
        
        final_score = semantic_score + rule_based_score
        
        original_score = mentor.get("score", 0.0)
        
        reason = _generate_reason(
            reranker_score,
            metadata,
            mentee_data,
            rating_score
        )
        
        scored_mentors.append({
            "mentor_id": mentor["mentor_id"],
            "score": final_score,
            "semantic_similarity": original_score,
            "reranker_score": reranker_score,
            "metadata": metadata,
            "reason": reason
        })
    
    scoring_time = time.perf_counter() - scoring_start
    logger.info(f"[SCORING] Scoring completed in {scoring_time:.3f}s")
    
    sort_start = time.perf_counter()
    scored_mentors.sort(key=lambda x: x["score"], reverse=True)
    sort_time = time.perf_counter() - sort_start
    logger.info(f"[SCORING] Sorting completed in {sort_time:.3f}s")
    
    result = scored_mentors[:final_count]
    
    for idx, mentor in enumerate(result, 1):
        metadata = mentor.get("metadata", {})
        logger.info(
            f"[SCORING] Final rank #{idx}: mentor_id={mentor['mentor_id']}, "
            f"rating={metadata.get('rating', 0.0)}, "
            f"pinecone_score={mentor.get('semantic_similarity', 0.0):.4f}, "
            f"reranker_score={mentor.get('reranker_score', 0.0):.4f}, "
            f"final_score={mentor['score']:.4f}, "
            f"semantic_weight={settings.SEMANTIC_WEIGHT}, "
            f"rule_based_weight={settings.RULE_BASED_WEIGHT}"
        )
    
    total_time = time.perf_counter() - start_time
    logger.info(f"[SCORING] Final scoring completed in {total_time:.3f}s: {len(result)} mentors returned")
    
    return result

def _calculate_rating_score(rating: float) -> float:
    if rating <= 0:
        return 0.0
    return min(rating / 5.0, 1.0)


def _generate_reason(
    reranker_score: float,
    metadata: Dict[str, Any],
    mentee_data: Dict[str, Any],
    rating_score: float
) -> str:
    reasons = []
    
    if reranker_score >= 0.8:
        reasons.append("Chuyên môn rất phù hợp")
    elif reranker_score >= 0.6:
        reasons.append("Phù hợp với mục tiêu của bạn")
    
    rating = metadata.get("rating", 0.0)
    if rating >= 4.5:
        reasons.append("Đánh giá xuất sắc")
    elif rating >= 4.0:
        reasons.append("Đánh giá cao")
    
    if not reasons:
        reasons.append("Phù hợp tổng thể")
    
    return "; ".join(reasons[:3])