012-kaopeilian/backend/scripts/fix_exam_scores.py

#!/usr/bin/env python3
"""
修复历史考试的小数分数问题

将历史考试中的小数分数（如 0.9090909090909091）转换为整数分数
使用智能整数分配算法，确保所有题目分数之和等于总分

使用方法：
    # 在后端容器中执行
    cd /app
    python scripts/fix_exam_scores.py --dry-run  # 预览模式，不实际修改
    python scripts/fix_exam_scores.py            # 实际执行修复
"""

import sys
import json
import argparse
from decimal import Decimal

# 添加项目路径
sys.path.insert(0, '/app')

def distribute_integer_scores(total_score: float, question_count: int) -> list:
    """
    整数分配分数
    
    将总分分配为整数，前面的题目分数可能比后面的多1分
    
    示例：
        100分 / 6题 = [17, 17, 17, 17, 16, 16]
        100分 / 11题 = [10, 10, 10, 10, 10, 10, 10, 10, 10, 5, 5]
    """
    total = int(total_score)
    count = question_count
    
    # 基础分数（向下取整）
    base_score = total // count
    # 需要额外加1分的题目数量
    extra_count = total % count
    
    # 生成分数列表
    scores = []
    for i in range(count):
        if i < extra_count:
            scores.append(base_score + 1)
        else:
            scores.append(base_score)
    
    return scores


def is_decimal_score(score) -> bool:
    """检查分数是否是小数（非整数）"""
    if score is None:
        return False
    try:
        score_float = float(score)
        return score_float != int(score_float)
    except (ValueError, TypeError):
        return False


def fix_exam_scores(dry_run: bool = True, db_url: str = None):
    """
    修复考试分数
    
    Args:
        dry_run: 如果为 True，只预览不实际修改
        db_url: 数据库连接字符串，如果为 None 则从环境变量读取
    """
    import os
    from sqlalchemy import create_engine, text
    from sqlalchemy.orm import sessionmaker
    
    # 获取数据库连接
    if db_url is None:
        db_url = os.environ.get('DATABASE_URL')
        if not db_url:
            # 尝试从配置文件读取
            try:
                from app.core.config import settings
                db_url = settings.DATABASE_URL
            except:
                print("错误：无法获取数据库连接字符串")
                print("请设置 DATABASE_URL 环境变量或确保在正确的目录下运行")
                sys.exit(1)
    
    # 将 mysql+aiomysql:// 转换为 mysql+pymysql://
    if 'aiomysql' in db_url:
        db_url = db_url.replace('aiomysql', 'pymysql')
    
    print(f"连接数据库...")
    engine = create_engine(db_url, echo=False)
    Session = sessionmaker(bind=engine)
    session = Session()
    
    try:
        # 查询所有考试记录
        result = session.execute(text("""
            SELECT id, user_id, course_id, exam_name, question_count, total_score, questions 
            FROM exams 
            WHERE questions IS NOT NULL 
            ORDER BY id DESC
        """))
        
        exams = result.fetchall()
        print(f"找到 {len(exams)} 条考试记录")
        
        fixed_count = 0
        skipped_count = 0
        error_count = 0
        
        for exam in exams:
            exam_id, user_id, course_id, exam_name, question_count, total_score, questions_json = exam
            
            try:
                # 解析 questions JSON
                if isinstance(questions_json, str):
                    questions = json.loads(questions_json)
                else:
                    questions = questions_json
                
                if not questions or not isinstance(questions, list):
                    skipped_count += 1
                    continue
                
                # 检查是否有小数分数
                has_decimal = False
                for q in questions:
                    if 'score' in q and is_decimal_score(q['score']):
                        has_decimal = True
                        break
                
                if not has_decimal:
                    skipped_count += 1
                    continue
                
                # 计算新的整数分数
                actual_count = len(questions)
                actual_total = total_score or 100
                new_scores = distribute_integer_scores(actual_total, actual_count)
                
                # 更新每道题的分数
                old_scores = [q.get('score', 0) for q in questions]
                for i, q in enumerate(questions):
                    q['score'] = new_scores[i]
                
                # 验证总分
                new_total = sum(new_scores)
                if abs(new_total - actual_total) > 0.01:
                    print(f"  警告：exam_id={exam_id} 分数总和不匹配: {new_total} != {actual_total}")
                    error_count += 1
                    continue
                
                if dry_run:
                    print(f"[预览] exam_id={exam_id}, 课程={course_id}, 题数={actual_count}, 总分={actual_total}")
                    print(f"        旧分数: {old_scores[:5]}..." if len(old_scores) > 5 else f"        旧分数: {old_scores}")
                    print(f"        新分数: {new_scores[:5]}..." if len(new_scores) > 5 else f"        新分数: {new_scores}")
                else:
                    # 实际更新数据库
                    new_json = json.dumps(questions, ensure_ascii=False)
                    session.execute(text("""
                        UPDATE exams SET questions = :questions WHERE id = :exam_id
                    """), {"questions": new_json, "exam_id": exam_id})
                    print(f"[已修复] exam_id={exam_id}, 题数={actual_count}, 分数: {new_scores}")
                
                fixed_count += 1
                
            except Exception as e:
                print(f"  错误：处理 exam_id={exam_id} 时出错: {e}")
                error_count += 1
                continue
        
        if not dry_run:
            session.commit()
            print("\n已提交数据库更改")
        
        print(f"\n=== 统计 ===")
        print(f"需要修复: {fixed_count}")
        print(f"已跳过（无小数）: {skipped_count}")
        print(f"错误: {error_count}")
        
        if dry_run:
            print("\n这是预览模式，未实际修改数据库。")
            print("如需实际执行，请去掉 --dry-run 参数重新运行。")
        
    except Exception as e:
        print(f"执行失败: {e}")
        session.rollback()
        raise
    finally:
        session.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="修复历史考试的小数分数问题")
    parser.add_argument("--dry-run", action="store_true", help="预览模式，不实际修改数据库")
    parser.add_argument("--db-url", type=str, help="数据库连接字符串")
    args = parser.parse_args()
    
    fix_exam_scores(dry_run=args.dry_run, db_url=args.db_url)