012-kaopeilian/backend/scripts/check_database_status.py

"""
检查数据库现状脚本
用于探索users、teams、exams、practice_sessions表的数据情况
"""
import asyncio
import sys
from pathlib import Path

# 添加项目路径
sys.path.append(str(Path(__file__).resolve().parent.parent))

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import AsyncSessionLocal
from app.core.logger import logger
from app.models.user import User, Team, UserTeam
from app.models.exam import Exam
from app.models.practice import PracticeSession
from app.models.position_member import PositionMember


async def check_database(db: AsyncSession):
    """检查数据库数据"""
    
    print("=" * 60)
    print("数据库状态检查")
    print("=" * 60)
    
    # 1. 检查用户数据
    print("\n【用户表 (users)】")
    result = await db.execute(select(func.count()).select_from(User))
    total_users = result.scalar() or 0
    print(f"总用户数: {total_users}")
    
    if total_users > 0:
        # 按角色统计
        result = await db.execute(
            select(User.role, func.count()).group_by(User.role)
        )
        print("按角色统计:")
        for role, count in result.all():
            print(f"  - {role}: {count}")
        
        # 显示部分用户
        result = await db.execute(
            select(User).limit(5)
        )
        users = result.scalars().all()
        print("\n前5个用户:")
        for user in users:
            print(f"  - ID:{user.id}, 用户名:{user.username}, 姓名:{user.full_name}, 角色:{user.role}")
    
    # 2. 检查团队数据
    print("\n【团队表 (teams)】")
    result = await db.execute(select(func.count()).select_from(Team))
    total_teams = result.scalar() or 0
    print(f"总团队数: {total_teams}")
    
    if total_teams > 0:
        result = await db.execute(select(Team).limit(5))
        teams = result.scalars().all()
        print("团队列表:")
        for team in teams:
            print(f"  - ID:{team.id}, 名称:{team.name}, 代码:{team.code}, 类型:{team.team_type}")
    
    # 3. 检查用户团队关联
    print("\n【用户团队关联表 (user_teams)】")
    result = await db.execute(select(func.count()).select_from(UserTeam))
    total_relations = result.scalar() or 0
    print(f"总关联记录数: {total_relations}")
    
    if total_relations > 0:
        # 查询前5条关联记录
        result = await db.execute(
            select(UserTeam, User.full_name, Team.name)
            .join(User, UserTeam.user_id == User.id)
            .join(Team, UserTeam.team_id == Team.id)
            .limit(5)
        )
        print("关联记录示例:")
        for user_team, user_name, team_name in result.all():
            print(f"  - 用户:{user_name}, 团队:{team_name}, 角色:{user_team.role}")
    
    # 4. 检查岗位成员
    print("\n【岗位成员表 (position_members)】")
    result = await db.execute(
        select(func.count()).select_from(PositionMember).where(
            PositionMember.is_deleted == False
        )
    )
    total_position_members = result.scalar() or 0
    print(f"总岗位成员数: {total_position_members}")
    
    # 5. 检查考试记录
    print("\n【考试表 (exams)】")
    result = await db.execute(select(func.count()).select_from(Exam))
    total_exams = result.scalar() or 0
    print(f"总考试记录数: {total_exams}")
    
    if total_exams > 0:
        # 按状态统计
        result = await db.execute(
            select(Exam.status, func.count()).group_by(Exam.status)
        )
        print("按状态统计:")
        for status, count in result.all():
            print(f"  - {status}: {count}")
        
        # 显示最近5条考试记录
        result = await db.execute(
            select(Exam, User.full_name)
            .join(User, Exam.user_id == User.id)
            .order_by(Exam.created_at.desc())
            .limit(5)
        )
        print("\n最近5条考试记录:")
        for exam, user_name in result.all():
            print(f"  - 用户:{user_name}, 课程ID:{exam.course_id}, 状态:{exam.status}, "
                  f"分数:{exam.round1_score}, 时间:{exam.created_at}")
    
    # 6. 检查陪练记录
    print("\n【陪练会话表 (practice_sessions)】")
    result = await db.execute(select(func.count()).select_from(PracticeSession))
    total_sessions = result.scalar() or 0
    print(f"总陪练记录数: {total_sessions}")
    
    if total_sessions > 0:
        # 按状态统计
        result = await db.execute(
            select(PracticeSession.status, func.count()).group_by(PracticeSession.status)
        )
        print("按状态统计:")
        for status, count in result.all():
            print(f"  - {status}: {count}")
        
        # 显示最近5条陪练记录
        result = await db.execute(
            select(PracticeSession, User.full_name)
            .join(User, PracticeSession.user_id == User.id)
            .order_by(PracticeSession.start_time.desc())
            .limit(5)
        )
        print("\n最近5条陪练记录:")
        for session, user_name in result.all():
            print(f"  - 用户:{user_name}, 场景:{session.scene_name}, 状态:{session.status}, "
                  f"时长:{session.duration_seconds}秒, 时间:{session.start_time}")
    
    # 总结
    print("\n" + "=" * 60)
    print("检查总结:")
    print("=" * 60)
    if total_users == 0:
        print("❌ 数据库为空，需要注入测试数据")
    elif total_teams == 0 or total_relations == 0:
        print("⚠️  有用户但无团队数据，需要创建团队")
    elif total_exams == 0:
        print("⚠️  有用户和团队但无考试记录，需要创建考试数据")
    elif total_sessions == 0:
        print("⚠️  有用户和团队但无陪练记录，需要创建陪练数据")
    else:
        print("✅ 数据库有基本数据，可以直接使用")
    print("=" * 60)


async def main():
    """主函数"""
    async with AsyncSessionLocal() as db:
        try:
            await check_database(db)
        except Exception as e:
            logger.error(f"检查数据库失败: {str(e)}", exc_info=True)
            raise


if __name__ == "__main__":
    asyncio.run(main())