012-kaopeilian/backend/app/services/level_service.py

"""
等级服务

提供用户等级管理功能：
- 经验值获取与计算
- 等级升级判断
- 每日签到
- 排行榜查询
"""

from datetime import datetime, date, timedelta
from typing import Optional, List, Dict, Any, Tuple
from sqlalchemy import select, func, and_, desc
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.logger import get_logger
from app.models.level import (
    UserLevel, ExpHistory, LevelConfig, ExpType
)
from app.models.user import User

logger = get_logger(__name__)


# 经验值配置
EXP_CONFIG = {
    # 每日签到
    "login_base": 10,           # 基础签到经验
    "login_streak_7": 20,       # 连续7天额外奖励
    "login_streak_30": 50,      # 连续30天额外奖励
    
    # 考试
    "exam_pass": 50,            # 通过考试
    "exam_excellent": 30,       # 90分以上额外
    "exam_perfect": 50,         # 满分额外
    
    # 练习
    "practice_complete": 20,    # 完成练习
    "practice_good": 10,        # 80分以上额外
    
    # 陪练
    "training_complete": 30,    # 完成陪练
    "training_good": 15,        # 80分以上额外
    
    # 任务
    "task_complete": 40,        # 完成任务
}


class LevelService:
    """等级服务"""
    
    def __init__(self, db: AsyncSession):
        self.db = db
        self._level_configs: Optional[List[LevelConfig]] = None
    
    async def _get_level_configs(self) -> List[LevelConfig]:
        """获取等级配置（带缓存）"""
        if self._level_configs is None:
            result = await self.db.execute(
                select(LevelConfig).order_by(LevelConfig.level)
            )
            self._level_configs = list(result.scalars().all())
        return self._level_configs
    
    async def get_or_create_user_level(self, user_id: int) -> UserLevel:
        """
        获取或创建用户等级记录
        
        Args:
            user_id: 用户ID
            
        Returns:
            UserLevel 对象
        """
        result = await self.db.execute(
            select(UserLevel).where(UserLevel.user_id == user_id)
        )
        user_level = result.scalar_one_or_none()
        
        if not user_level:
            user_level = UserLevel(
                user_id=user_id,
                level=1,
                exp=0,
                total_exp=0,
                login_streak=0,
                max_login_streak=0
            )
            self.db.add(user_level)
            await self.db.flush()
            logger.info(f"为用户 {user_id} 创建等级记录")
        
        return user_level
    
    async def get_user_level_info(self, user_id: int) -> Dict[str, Any]:
        """
        获取用户等级详细信息
        
        Args:
            user_id: 用户ID
            
        Returns:
            包含等级、经验值、称号等信息的字典
        """
        user_level = await self.get_or_create_user_level(user_id)
        level_configs = await self._get_level_configs()
        
        # 获取当前等级配置
        current_config = next(
            (c for c in level_configs if c.level == user_level.level),
            level_configs[0] if level_configs else None
        )
        
        # 获取下一等级配置
        next_config = next(
            (c for c in level_configs if c.level == user_level.level + 1),
            None
        )
        
        # 计算升级所需经验
        exp_to_next_level = 0
        next_level_total_exp = 0
        if next_config:
            next_level_total_exp = next_config.total_exp_required
            exp_to_next_level = next_level_total_exp - user_level.total_exp
            if exp_to_next_level < 0:
                exp_to_next_level = 0
        
        return {
            "user_id": user_id,
            "level": user_level.level,
            "exp": user_level.exp,
            "total_exp": user_level.total_exp,
            "title": current_config.title if current_config else "初学者",
            "color": current_config.color if current_config else "#909399",
            "login_streak": user_level.login_streak,
            "max_login_streak": user_level.max_login_streak,
            "last_checkin_at": user_level.last_checkin_at.isoformat() if user_level.last_checkin_at else None,
            "next_level_exp": next_level_total_exp,
            "exp_to_next_level": exp_to_next_level,
            "is_max_level": next_config is None,
        }
    
    async def add_exp(
        self,
        user_id: int,
        exp_amount: int,
        exp_type: str,
        description: str,
        source_id: Optional[int] = None
    ) -> Tuple[UserLevel, bool, Optional[int]]:
        """
        增加用户经验值
        
        Args:
            user_id: 用户ID
            exp_amount: 经验值数量
            exp_type: 经验值类型
            description: 描述
            source_id: 来源ID（可选）
            
        Returns:
            (用户等级对象, 是否升级, 新等级)
        """
        if exp_amount <= 0:
            logger.warning(f"尝试增加非正数经验值: {exp_amount}")
            return await self.get_or_create_user_level(user_id), False, None
        
        user_level = await self.get_or_create_user_level(user_id)
        level_before = user_level.level
        
        # 增加经验值
        user_level.exp += exp_amount
        user_level.total_exp += exp_amount
        
        # 检查是否升级
        level_configs = await self._get_level_configs()
        leveled_up = False
        new_level = None
        
        for config in level_configs:
            if config.level > user_level.level and user_level.total_exp >= config.total_exp_required:
                user_level.level = config.level
                leveled_up = True
                new_level = config.level
        
        # 记录经验值历史
        exp_history = ExpHistory(
            user_id=user_id,
            exp_change=exp_amount,
            exp_type=exp_type,
            source_id=source_id,
            description=description,
            level_before=level_before,
            level_after=user_level.level
        )
        self.db.add(exp_history)
        
        await self.db.flush()
        
        if leveled_up:
            logger.info(f"用户 {user_id} 升级: {level_before} -> {new_level}")
        
        logger.info(f"用户 {user_id} 获得 {exp_amount} 经验值: {description}")
        
        return user_level, leveled_up, new_level
    
    async def daily_checkin(self, user_id: int) -> Dict[str, Any]:
        """
        每日签到
        
        Args:
            user_id: 用户ID
            
        Returns:
            签到结果
        """
        user_level = await self.get_or_create_user_level(user_id)
        today = date.today()
        
        # 检查今天是否已签到
        if user_level.last_login_date == today:
            return {
                "success": False,
                "message": "今天已经签到过了",
                "exp_gained": 0,
                "login_streak": user_level.login_streak,
                "already_checked_in": True
            }
        
        # 计算连续登录
        yesterday = today - timedelta(days=1)
        if user_level.last_login_date == yesterday:
            # 连续登录
            user_level.login_streak += 1
        else:
            # 中断了，重新计算
            user_level.login_streak = 1
        
        # 更新最长连续登录记录
        if user_level.login_streak > user_level.max_login_streak:
            user_level.max_login_streak = user_level.login_streak
        
        user_level.last_login_date = today
        user_level.last_checkin_at = datetime.now()
        
        # 计算签到经验
        exp_gained = EXP_CONFIG["login_base"]
        bonus_exp = 0
        bonus_reason = []
        
        # 连续登录奖励
        if user_level.login_streak >= 30 and user_level.login_streak % 30 == 0:
            bonus_exp += EXP_CONFIG["login_streak_30"]
            bonus_reason.append(f"连续{user_level.login_streak}天")
        elif user_level.login_streak >= 7 and user_level.login_streak % 7 == 0:
            bonus_exp += EXP_CONFIG["login_streak_7"]
            bonus_reason.append(f"连续{user_level.login_streak}天")
        
        total_exp = exp_gained + bonus_exp
        
        # 添加经验值
        description = f"每日签到"
        if bonus_reason:
            description += f"（{', '.join(bonus_reason)}奖励）"
        
        _, leveled_up, new_level = await self.add_exp(
            user_id=user_id,
            exp_amount=total_exp,
            exp_type=ExpType.LOGIN,
            description=description
        )
        
        await self.db.commit()
        
        return {
            "success": True,
            "message": "签到成功",
            "exp_gained": total_exp,
            "base_exp": exp_gained,
            "bonus_exp": bonus_exp,
            "login_streak": user_level.login_streak,
            "leveled_up": leveled_up,
            "new_level": new_level,
            "already_checked_in": False
        }
    
    async def get_exp_history(
        self,
        user_id: int,
        limit: int = 50,
        offset: int = 0,
        exp_type: Optional[str] = None
    ) -> Tuple[List[Dict[str, Any]], int]:
        """
        获取经验值历史
        
        Args:
            user_id: 用户ID
            limit: 限制数量
            offset: 偏移量
            exp_type: 类型筛选
            
        Returns:
            (历史记录列表, 总数)
        """
        # 构建查询
        query = select(ExpHistory).where(ExpHistory.user_id == user_id)
        
        if exp_type:
            query = query.where(ExpHistory.exp_type == exp_type)
        
        # 获取总数
        count_query = select(func.count(ExpHistory.id)).where(ExpHistory.user_id == user_id)
        if exp_type:
            count_query = count_query.where(ExpHistory.exp_type == exp_type)
        total_result = await self.db.execute(count_query)
        total = total_result.scalar() or 0
        
        # 获取记录
        query = query.order_by(desc(ExpHistory.created_at)).limit(limit).offset(offset)
        result = await self.db.execute(query)
        records = result.scalars().all()
        
        history = [
            {
                "id": r.id,
                "exp_change": r.exp_change,
                "exp_type": r.exp_type,
                "description": r.description,
                "level_before": r.level_before,
                "level_after": r.level_after,
                "created_at": r.created_at.isoformat() if r.created_at else None
            }
            for r in records
        ]
        
        return history, total
    
    async def get_leaderboard(
        self,
        limit: int = 50,
        offset: int = 0
    ) -> Tuple[List[Dict[str, Any]], int]:
        """
        获取等级排行榜
        
        Args:
            limit: 限制数量
            offset: 偏移量
            
        Returns:
            (排行榜列表, 总数)
        """
        # 获取总数
        count_query = select(func.count(UserLevel.id))
        total_result = await self.db.execute(count_query)
        total = total_result.scalar() or 0
        
        # 获取排行榜（按等级和总经验值排序）
        query = (
            select(UserLevel, User)
            .join(User, UserLevel.user_id == User.id)
            .where(User.is_deleted == False)
            .order_by(desc(UserLevel.level), desc(UserLevel.total_exp))
            .limit(limit)
            .offset(offset)
        )
        result = await self.db.execute(query)
        rows = result.all()
        
        # 获取等级配置
        level_configs = await self._get_level_configs()
        config_map = {c.level: c for c in level_configs}
        
        leaderboard = []
        for i, (user_level, user) in enumerate(rows):
            config = config_map.get(user_level.level)
            leaderboard.append({
                "rank": offset + i + 1,
                "user_id": user.id,
                "username": user.username,
                "full_name": user.full_name,
                "avatar_url": user.avatar_url,
                "level": user_level.level,
                "title": config.title if config else "初学者",
                "color": config.color if config else "#909399",
                "total_exp": user_level.total_exp,
                "login_streak": user_level.login_streak,
            })
        
        return leaderboard, total
    
    async def get_user_rank(self, user_id: int) -> Optional[int]:
        """
        获取用户排名
        
        Args:
            user_id: 用户ID
            
        Returns:
            排名（从1开始）
        """
        user_level = await self.get_or_create_user_level(user_id)
        
        # 计算排在该用户前面的人数
        query = select(func.count(UserLevel.id)).where(
            and_(
                UserLevel.user_id != user_id,
                (
                    (UserLevel.level > user_level.level) |
                    (
                        (UserLevel.level == user_level.level) &
                        (UserLevel.total_exp > user_level.total_exp)
                    )
                )
            )
        )
        result = await self.db.execute(query)
        count = result.scalar() or 0
        
        return count + 1
    
    async def add_exam_exp(
        self,
        user_id: int,
        exam_id: int,
        score: float,
        is_passed: bool
    ) -> Optional[Dict[str, Any]]:
        """
        考试通过获得经验值
        
        Args:
            user_id: 用户ID
            exam_id: 考试ID
            score: 得分
            is_passed: 是否通过
            
        Returns:
            经验值变化信息
        """
        if not is_passed:
            return None
        
        exp_gained = EXP_CONFIG["exam_pass"]
        bonus = []
        
        if score >= 100:
            exp_gained += EXP_CONFIG["exam_perfect"]
            bonus.append("满分")
        elif score >= 90:
            exp_gained += EXP_CONFIG["exam_excellent"]
            bonus.append("优秀")
        
        description = f"通过考试"
        if bonus:
            description += f"（{', '.join(bonus)}奖励）"
        
        _, leveled_up, new_level = await self.add_exp(
            user_id=user_id,
            exp_amount=exp_gained,
            exp_type=ExpType.EXAM,
            description=description,
            source_id=exam_id
        )
        
        return {
            "exp_gained": exp_gained,
            "leveled_up": leveled_up,
            "new_level": new_level
        }
    
    async def add_practice_exp(
        self,
        user_id: int,
        session_id: int,
        score: Optional[float] = None
    ) -> Dict[str, Any]:
        """
        完成练习获得经验值
        
        Args:
            user_id: 用户ID
            session_id: 练习会话ID
            score: 得分（可选）
            
        Returns:
            经验值变化信息
        """
        exp_gained = EXP_CONFIG["practice_complete"]
        bonus = []
        
        if score is not None and score >= 80:
            exp_gained += EXP_CONFIG["practice_good"]
            bonus.append("高分")
        
        description = f"完成练习"
        if bonus:
            description += f"（{', '.join(bonus)}奖励）"
        
        _, leveled_up, new_level = await self.add_exp(
            user_id=user_id,
            exp_amount=exp_gained,
            exp_type=ExpType.PRACTICE,
            description=description,
            source_id=session_id
        )
        
        return {
            "exp_gained": exp_gained,
            "leveled_up": leveled_up,
            "new_level": new_level
        }
    
    async def add_training_exp(
        self,
        user_id: int,
        session_id: int,
        score: Optional[float] = None
    ) -> Dict[str, Any]:
        """
        完成陪练获得经验值
        
        Args:
            user_id: 用户ID
            session_id: 陪练会话ID
            score: 得分（可选）
            
        Returns:
            经验值变化信息
        """
        exp_gained = EXP_CONFIG["training_complete"]
        bonus = []
        
        if score is not None and score >= 80:
            exp_gained += EXP_CONFIG["training_good"]
            bonus.append("高分")
        
        description = f"完成陪练"
        if bonus:
            description += f"（{', '.join(bonus)}奖励）"
        
        _, leveled_up, new_level = await self.add_exp(
            user_id=user_id,
            exp_amount=exp_gained,
            exp_type=ExpType.TRAINING,
            description=description,
            source_id=session_id
        )
        
        return {
            "exp_gained": exp_gained,
            "leveled_up": leveled_up,
            "new_level": new_level
        }
    
    async def add_task_exp(
        self,
        user_id: int,
        task_id: int
    ) -> Dict[str, Any]:
        """
        完成任务获得经验值
        
        Args:
            user_id: 用户ID
            task_id: 任务ID
            
        Returns:
            经验值变化信息
        """
        exp_gained = EXP_CONFIG["task_complete"]
        
        _, leveled_up, new_level = await self.add_exp(
            user_id=user_id,
            exp_amount=exp_gained,
            exp_type=ExpType.TASK,
            description="完成任务",
            source_id=task_id
        )
        
        return {
            "exp_gained": exp_gained,
            "leveled_up": leveled_up,
            "new_level": new_level
        }