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

"""
数据大屏服务

提供企业级和团队级数据大屏功能：
- 学习数据概览
- 部门/团队对比
- 趋势分析
- 实时动态
"""

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

from app.core.logger import get_logger
from app.models.user import User
from app.models.course import Course, CourseMaterial
from app.models.exam import Exam
from app.models.practice import PracticeSession
from app.models.training import TrainingSession, TrainingReport
from app.models.level import UserLevel, ExpHistory, UserBadge
from app.models.position import Position
from app.models.position_member import PositionMember

logger = get_logger(__name__)


class DashboardService:
    """数据大屏服务"""
    
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_enterprise_overview(self, enterprise_id: Optional[int] = None) -> Dict[str, Any]:
        """
        获取企业级数据概览
        
        Args:
            enterprise_id: 企业ID（可选，用于多租户）
            
        Returns:
            企业级数据概览
        """
        try:
            today = date.today()
            week_ago = today - timedelta(days=7)
            month_ago = today - timedelta(days=30)
            
            # 基础统计
            # 1. 总学员数
            result = await self.db.execute(
                select(func.count(User.id))
                .where(User.is_deleted == False, User.role == 'trainee')
            )
            total_users = result.scalar() or 0
            
            # 2. 今日活跃用户（有经验值记录）
            try:
                result = await self.db.execute(
                    select(func.count(func.distinct(ExpHistory.user_id)))
                    .where(func.date(ExpHistory.created_at) == today)
                )
                today_active = result.scalar() or 0
            except Exception as e:
                logger.warning(f"获取今日活跃用户失败: {e}")
                today_active = 0
            
            # 3. 本周活跃用户
            try:
                result = await self.db.execute(
                    select(func.count(func.distinct(ExpHistory.user_id)))
                    .where(ExpHistory.created_at >= datetime.combine(week_ago, datetime.min.time()))
                )
                week_active = result.scalar() or 0
            except Exception as e:
                logger.warning(f"获取本周活跃用户失败: {e}")
                week_active = 0
            
            # 4. 本月活跃用户
            try:
                result = await self.db.execute(
                    select(func.count(func.distinct(ExpHistory.user_id)))
                    .where(ExpHistory.created_at >= datetime.combine(month_ago, datetime.min.time()))
                )
                month_active = result.scalar() or 0
            except Exception as e:
                logger.warning(f"获取本月活跃用户失败: {e}")
                month_active = 0
            
            # 5. 总学习时长（小时）
            practice_hours = 0
            training_hours = 0
            try:
                result = await self.db.execute(
                    select(func.coalesce(func.sum(PracticeSession.duration_seconds), 0))
                    .where(PracticeSession.status == 'completed')
                )
                practice_hours = (result.scalar() or 0) / 3600
            except Exception as e:
                logger.warning(f"获取陪练时长失败: {e}")
            
            try:
                result = await self.db.execute(
                    select(func.coalesce(func.sum(TrainingSession.duration_seconds), 0))
                    .where(TrainingSession.status == 'completed')  # 修复: 使用小写
                )
                training_hours = (result.scalar() or 0) / 3600
            except Exception as e:
                logger.warning(f"获取培训时长失败: {e}")
            
            total_hours = round(practice_hours + training_hours, 1)
            
            # 6. 考试统计
            exam_count = 0
            exam_passed = 0
            exam_avg_score = 0
            try:
                result = await self.db.execute(
                    select(
                        func.count(Exam.id),
                        func.count(case((Exam.is_passed == True, 1))),
                        func.avg(Exam.score)
                    )
                    .where(Exam.status == 'submitted')
                )
                exam_row = result.first()
                if exam_row:
                    exam_count = exam_row[0] or 0
                    exam_passed = exam_row[1] or 0
                    exam_avg_score = round(exam_row[2] or 0, 1)
            except Exception as e:
                logger.warning(f"获取考试统计失败: {e}")
            
            exam_pass_rate = round(exam_passed / exam_count * 100, 1) if exam_count > 0 else 0
            
            # 7. 满分人数
            perfect_users = 0
            try:
                result = await self.db.execute(
                    select(func.count(func.distinct(Exam.user_id)))
                    .where(
                        Exam.status == 'submitted', 
                        Exam.score.isnot(None),
                        Exam.total_score.isnot(None),
                        Exam.score >= Exam.total_score
                    )
                )
                perfect_users = result.scalar() or 0
            except Exception as e:
                logger.warning(f"获取满分人数失败: {e}")
            
            # 8. 签到率（今日签到人数/总用户数）
            today_checkin = 0
            try:
                result = await self.db.execute(
                    select(func.count(UserLevel.id))
                    .where(UserLevel.last_login_date == today)
                )
                today_checkin = result.scalar() or 0
            except Exception as e:
                logger.warning(f"获取签到率失败: {e}")
            
            checkin_rate = round(today_checkin / total_users * 100, 1) if total_users > 0 else 0
            
            return {
                "overview": {
                    "total_users": total_users,
                    "today_active": today_active,
                    "week_active": week_active,
                    "month_active": month_active,
                    "total_hours": total_hours,
                    "checkin_rate": checkin_rate,
                },
                "exam": {
                    "total_count": exam_count,
                    "pass_rate": exam_pass_rate,
                    "avg_score": exam_avg_score,
                    "perfect_users": perfect_users,
                },
                "updated_at": datetime.now().isoformat()
            }
        except Exception as e:
            logger.error(f"获取企业概览失败: {e}")
            # 返回默认数据而不是抛出异常
            return {
                "overview": {
                    "total_users": 0,
                    "today_active": 0,
                    "week_active": 0,
                    "month_active": 0,
                    "total_hours": 0,
                    "checkin_rate": 0,
                },
                "exam": {
                    "total_count": 0,
                    "pass_rate": 0,
                    "avg_score": 0,
                    "perfect_users": 0,
                },
                "updated_at": datetime.now().isoformat()
            }
    
    async def get_department_comparison(self) -> List[Dict[str, Any]]:
        """
        获取部门/团队学习对比数据
        
        Returns:
            部门对比列表
        """
        # 获取所有岗位及其成员的学习数据
        result = await self.db.execute(
            select(Position)
            .where(Position.is_deleted == False)
            .order_by(Position.name)
        )
        positions = result.scalars().all()
        
        departments = []
        for pos in positions:
            # 获取该岗位的成员数
            result = await self.db.execute(
                select(func.count(PositionMember.id))
                .where(PositionMember.position_id == pos.id)
            )
            member_count = result.scalar() or 0
            
            if member_count == 0:
                continue
            
            # 获取成员ID列表
            result = await self.db.execute(
                select(PositionMember.user_id)
                .where(PositionMember.position_id == pos.id)
            )
            member_ids = [row[0] for row in result.all()]
            
            # 统计该岗位成员的学习数据
            # 考试通过率
            result = await self.db.execute(
                select(
                    func.count(Exam.id),
                    func.count(case((Exam.is_passed == True, 1)))
                )
                .where(
                    Exam.user_id.in_(member_ids),
                    Exam.status == 'submitted'
                )
            )
            exam_row = result.first()
            exam_total = exam_row[0] or 0
            exam_passed = exam_row[1] or 0
            pass_rate = round(exam_passed / exam_total * 100, 1) if exam_total > 0 else 0
            
            # 平均学习时长
            result = await self.db.execute(
                select(func.coalesce(func.sum(PracticeSession.duration_seconds), 0))
                .where(
                    PracticeSession.user_id.in_(member_ids),
                    PracticeSession.status == 'completed'
                )
            )
            total_seconds = result.scalar() or 0
            avg_hours = round(total_seconds / 3600 / member_count, 1) if member_count > 0 else 0
            
            # 平均等级
            result = await self.db.execute(
                select(func.avg(UserLevel.level))
                .where(UserLevel.user_id.in_(member_ids))
            )
            avg_level = round(result.scalar() or 1, 1)
            
            departments.append({
                "id": pos.id,
                "name": pos.name,
                "member_count": member_count,
                "pass_rate": pass_rate,
                "avg_hours": avg_hours,
                "avg_level": avg_level,
            })
        
        # 按通过率排序
        departments.sort(key=lambda x: x["pass_rate"], reverse=True)
        
        return departments
    
    async def get_learning_trend(self, days: int = 7) -> Dict[str, Any]:
        """
        获取学习趋势数据
        
        Args:
            days: 统计天数
            
        Returns:
            趋势数据
        """
        today = date.today()
        dates = [(today - timedelta(days=i)) for i in range(days-1, -1, -1)]
        
        trend_data = []
        for d in dates:
            # 当日活跃用户
            result = await self.db.execute(
                select(func.count(func.distinct(ExpHistory.user_id)))
                .where(func.date(ExpHistory.created_at) == d)
            )
            active_users = result.scalar() or 0
            
            # 当日新增学习时长
            result = await self.db.execute(
                select(func.coalesce(func.sum(PracticeSession.duration_seconds), 0))
                .where(
                    func.date(PracticeSession.created_at) == d,
                    PracticeSession.status == 'completed'
                )
            )
            hours = round((result.scalar() or 0) / 3600, 1)
            
            # 当日考试次数
            result = await self.db.execute(
                select(func.count(Exam.id))
                .where(
                    func.date(Exam.created_at) == d,
                    Exam.status == 'submitted'
                )
            )
            exams = result.scalar() or 0
            
            trend_data.append({
                "date": d.isoformat(),
                "active_users": active_users,
                "learning_hours": hours,
                "exam_count": exams,
            })
        
        return {
            "dates": [d.isoformat() for d in dates],
            "trend": trend_data
        }
    
    async def get_level_distribution(self) -> Dict[str, Any]:
        """
        获取等级分布数据
        
        Returns:
            等级分布
        """
        result = await self.db.execute(
            select(UserLevel.level, func.count(UserLevel.id))
            .group_by(UserLevel.level)
            .order_by(UserLevel.level)
        )
        rows = result.all()
        
        distribution = {row[0]: row[1] for row in rows}
        
        # 补全1-10级
        for i in range(1, 11):
            if i not in distribution:
                distribution[i] = 0
        
        return {
            "levels": list(range(1, 11)),
            "counts": [distribution.get(i, 0) for i in range(1, 11)]
        }
    
    async def get_realtime_activities(self, limit: int = 20) -> List[Dict[str, Any]]:
        """
        获取实时动态
        
        Args:
            limit: 数量限制
            
        Returns:
            实时动态列表
        """
        activities = []
        
        try:
            # 获取最近的经验值记录
            result = await self.db.execute(
                select(ExpHistory, User)
                .join(User, ExpHistory.user_id == User.id)
                .order_by(ExpHistory.created_at.desc())
                .limit(limit)
            )
            rows = result.all()
            
            for exp, user in rows:
                activity_type = "学习"
                description = exp.description or ""
                if "考试" in description:
                    activity_type = "考试"
                elif "签到" in description:
                    activity_type = "签到"
                elif "陪练" in description:
                    activity_type = "陪练"
                elif "奖章" in description:
                    activity_type = "奖章"
                
                activities.append({
                    "id": exp.id,
                    "user_id": user.id,
                    "user_name": user.full_name or user.username,
                    "type": activity_type,
                    "description": description,
                    "exp_amount": exp.exp_change,  # 修复: exp_change 而非 exp_amount
                    "created_at": exp.created_at.isoformat() if exp.created_at else None,
                })
        except Exception as e:
            logger.error(f"获取实时动态失败: {e}")
            # 返回空列表而不是抛出异常
        
        return activities
    
    async def get_team_dashboard(self, team_leader_id: int) -> Dict[str, Any]:
        """
        获取团队级数据大屏
        
        Args:
            team_leader_id: 团队负责人ID
            
        Returns:
            团队数据
        """
        # 获取团队负责人管理的岗位
        result = await self.db.execute(
            select(Position)
            .where(
                Position.is_deleted == False,
                or_(
                    Position.manager_id == team_leader_id,
                    Position.created_by == team_leader_id
                )
            )
        )
        positions = result.scalars().all()
        position_ids = [p.id for p in positions]
        
        if not position_ids:
            return {
                "members": [],
                "overview": {
                    "total_members": 0,
                    "avg_level": 0,
                    "avg_exp": 0,
                    "total_badges": 0,
                },
                "pending_tasks": []
            }
        
        # 获取团队成员
        result = await self.db.execute(
            select(PositionMember.user_id)
            .where(PositionMember.position_id.in_(position_ids))
        )
        member_ids = [row[0] for row in result.all()]
        
        if not member_ids:
            return {
                "members": [],
                "overview": {
                    "total_members": 0,
                    "avg_level": 0,
                    "avg_exp": 0,
                    "total_badges": 0,
                },
                "pending_tasks": []
            }
        
        # 获取成员详细信息
        result = await self.db.execute(
            select(User, UserLevel)
            .outerjoin(UserLevel, User.id == UserLevel.user_id)
            .where(User.id.in_(member_ids))
            .order_by(UserLevel.total_exp.desc().nullslast())
        )
        rows = result.all()
        
        members = []
        total_exp = 0
        total_level = 0
        
        for user, level in rows:
            user_level = level.level if level else 1
            user_exp = level.total_exp if level else 0
            total_level += user_level
            total_exp += user_exp
            
            # 获取用户奖章数
            result = await self.db.execute(
                select(func.count(UserBadge.id))
                .where(UserBadge.user_id == user.id)
            )
            badge_count = result.scalar() or 0
            
            members.append({
                "id": user.id,
                "username": user.username,
                "full_name": user.full_name,
                "avatar_url": user.avatar_url,
                "level": user_level,
                "total_exp": user_exp,
                "badge_count": badge_count,
            })
        
        total_members = len(members)
        
        # 获取团队总奖章数
        result = await self.db.execute(
            select(func.count(UserBadge.id))
            .where(UserBadge.user_id.in_(member_ids))
        )
        total_badges = result.scalar() or 0
        
        return {
            "members": members,
            "overview": {
                "total_members": total_members,
                "avg_level": round(total_level / total_members, 1) if total_members > 0 else 0,
                "avg_exp": round(total_exp / total_members) if total_members > 0 else 0,
                "total_badges": total_badges,
            },
            "positions": [{"id": p.id, "name": p.name} for p in positions]
        }
    
    async def get_course_ranking(self, limit: int = 10) -> List[Dict[str, Any]]:
        """
        获取课程热度排行
        
        Args:
            limit: 数量限制
            
        Returns:
            课程排行列表
        """
        # 这里简化实现，实际应该统计课程学习次数
        result = await self.db.execute(
            select(Course)
            .where(Course.is_deleted == False, Course.is_published == True)
            .order_by(Course.created_at.desc())
            .limit(limit)
        )
        courses = result.scalars().all()
        
        ranking = []
        for i, course in enumerate(courses, 1):
            ranking.append({
                "rank": i,
                "id": course.id,
                "name": course.name,
                "description": course.description,
                # 这里可以添加实际的学习人数统计
                "learners": 0,
            })
        
        return ranking