all-project/学校学生操行历史数据迁移/05_validate_and_cleanup.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
学生操行分历史数据迁移 - 脚本5：验证与清理

功能说明：
- 验证迁移结果的完整性和正确性
- 生成详细的迁移报告
- 可选择清理临时表和日志

执行方式：
python 05_validate_and_cleanup.py [--cleanup]

注意事项：
- 默认只做验证，不清理数据
- 使用--cleanup参数才会清理临时表
"""

from __future__ import annotations

import argparse
import csv
import logging
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

import pymysql
from pymysql.cursors import DictCursor

from db_config import DB_CONFIG, TABLE_CONFIG, CONSTANTS

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler("logs/05_validate_cleanup.log", mode='a', encoding='utf-8')
    ]
)
logger = logging.getLogger("validate_cleanup")

class MigrationValidator:
    """迁移验证和清理处理器"""

    def __init__(self, cleanup: bool = False):
        self.db_config = DB_CONFIG
        self.temp_import_table = TABLE_CONFIG["temp_import"]
        self.temp_parsed_table = TABLE_CONFIG["temp_parsed"]
        self.production_table = TABLE_CONFIG["production"]
        self.students_table = TABLE_CONFIG["students"]
        self.classes_table = TABLE_CONFIG["classes"]
        self.cleanup = cleanup
        self.connection = None
        self.report_data = {}

    def connect_db(self) -> None:
        """连接数据库"""
        try:
            self.connection = pymysql.connect(**self.db_config)
            logger.info("数据库连接成功")
        except Exception as e:
            logger.error(f"数据库连接失败: {e}")
            raise

    def disconnect_db(self) -> None:
        """断开数据库连接"""
        if self.connection:
            self.connection.close()
            logger.info("数据库连接已断开")

    def get_migration_statistics(self) -> Dict[str, Any]:
        """获取迁移统计数据"""
        try:
            with self.connection.cursor(DictCursor) as cursor:
                stats = {}

                # Excel导入表统计
                sql_import = f"""
                SELECT
                    COUNT(*) as total_import,
                    SUM(CASE WHEN import_status = 'pending' THEN 1 ELSE 0 END) as pending_import,
                    SUM(CASE WHEN import_status = 'parsed' THEN 1 ELSE 0 END) as parsed_import,
                    SUM(CASE WHEN import_status = 'error' THEN 1 ELSE 0 END) as error_import
                FROM {self.temp_import_table}
                """
                cursor.execute(sql_import)
                stats.update(cursor.fetchone())

                # 解析表统计
                sql_parsed = f"""
                SELECT
                    COUNT(*) as total_parsed,
                    SUM(CASE WHEN processing_status = 'parsed' THEN 1 ELSE 0 END) as pending_parsed,
                    SUM(CASE WHEN processing_status = 'linked' THEN 1 ELSE 0 END) as linked_parsed,
                    SUM(CASE WHEN processing_status = 'imported' THEN 1 ELSE 0 END) as imported_parsed,
                    SUM(CASE WHEN processing_status = 'duplicate' THEN 1 ELSE 0 END) as duplicate_parsed,
                    SUM(CASE WHEN processing_status = 'error' THEN 1 ELSE 0 END) as error_parsed
                FROM {self.temp_parsed_table}
                """
                cursor.execute(sql_parsed)
                stats.update(cursor.fetchone())

                # 正式表统计
                sql_production = f"""
                SELECT COUNT(*) as total_production
                FROM {self.production_table}
                WHERE deleted = 0
                """
                cursor.execute(sql_production)
                production_result = cursor.fetchone()
                stats['total_production'] = production_result['total_production']

                # 数据质量检查
                sql_quality = f"""
                SELECT
                    COUNT(*) as invalid_scores,
                    SUM(CASE WHEN score < 0 OR score > 100 THEN 1 ELSE 0 END) as out_of_range_scores
                FROM {self.production_table}
                WHERE deleted = 0
                """
                cursor.execute(sql_quality)
                quality_result = cursor.fetchone()
                stats.update(quality_result)

                return stats

        except Exception as e:
            logger.error(f"获取统计数据失败: {e}")
            return {}

    def validate_data_integrity(self) -> Dict[str, Any]:
        """验证数据完整性"""
        issues = []

        try:
            with self.connection.cursor(DictCursor) as cursor:
                # 检查孤立记录（学生不存在）
                sql_orphan_students = f"""
                SELECT COUNT(*) as count
                FROM {self.production_table} cs
                LEFT JOIN {self.students_table} s ON cs.student_id = s.id
                WHERE cs.deleted = 0 AND (s.id IS NULL OR s.deleted = 1)
                """
                cursor.execute(sql_orphan_students)
                orphan_students = cursor.fetchone()['count']
                if orphan_students > 0:
                    issues.append(f"发现{orphan_students}条学生ID不存在的记录")

                # 检查孤立记录（班级不存在）
                sql_orphan_classes = f"""
                SELECT COUNT(*) as count
                FROM {self.production_table} cs
                LEFT JOIN {self.classes_table} c ON cs.class_id = c.id
                WHERE cs.deleted = 0 AND cs.class_id IS NOT NULL AND (c.id IS NULL OR c.deleted = 1)
                """
                cursor.execute(sql_orphan_classes)
                orphan_classes = cursor.fetchone()['count']
                if orphan_classes > 0:
                    issues.append(f"发现{orphan_classes}条班级ID不存在的记录")

                # 检查重复记录
                sql_duplicates = f"""
                SELECT COUNT(*) as count
                FROM (
                    SELECT student_id, academic_year, semester, month, COUNT(*) as cnt
                    FROM {self.production_table}
                    WHERE deleted = 0
                    GROUP BY student_id, academic_year, semester, month
                    HAVING cnt > 1
                ) t
                """
                cursor.execute(sql_duplicates)
                duplicates = cursor.fetchone()['count']
                if duplicates > 0:
                    issues.append(f"发现{duplicates}组重复记录（同一学生同月有多条记录）")

                # 检查异常分数
                sql_invalid_scores = f"""
                SELECT COUNT(*) as count
                FROM {self.production_table}
                WHERE deleted = 0 AND (score < 0 OR score > 100 OR score IS NULL)
                """
                cursor.execute(sql_invalid_scores)
                invalid_scores = cursor.fetchone()['count']
                if invalid_scores > 0:
                    issues.append(f"发现{invalid_scores}条分数不在0-100范围内的记录")

                return {
                    'issues': issues,
                    'orphan_students': orphan_students,
                    'orphan_classes': orphan_classes,
                    'duplicates': duplicates,
                    'invalid_scores': invalid_scores
                }

        except Exception as e:
            logger.error(f"数据完整性验证失败: {e}")
            return {'issues': [f"验证过程出错: {str(e)}"], 'error': True}

    def generate_detailed_report(self) -> str:
        """生成详细报告"""
        stats = self.get_migration_statistics()
        integrity = self.validate_data_integrity()

        report_lines = []
        report_lines.append("="*80)
        report_lines.append("学生操行分历史数据迁移报告")
        report_lines.append("="*80)
        report_lines.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report_lines.append("")

        # 总体统计
        report_lines.append("📊 总体统计")
        report_lines.append("-" * 40)
        report_lines.append(f"Excel原始记录: {stats.get('total_import', 0)}")
        report_lines.append(f"解析后月度记录: {stats.get('total_parsed', 0)}")
        report_lines.append(f"正式表总记录: {stats.get('total_production', 0)}")
        report_lines.append("")

        # 各阶段处理情况
        report_lines.append("🔄 处理阶段详情")
        report_lines.append("-" * 40)
        report_lines.append(f"Excel导入 - 待处理: {stats.get('pending_import', 0)}")
        report_lines.append(f"Excel导入 - 已解析: {stats.get('parsed_import', 0)}")
        report_lines.append(f"Excel导入 - 解析失败: {stats.get('error_import', 0)}")
        report_lines.append("")
        report_lines.append(f"数据解析 - 待关联: {stats.get('pending_parsed', 0)}")
        report_lines.append(f"数据解析 - 已关联: {stats.get('linked_parsed', 0)}")
        report_lines.append(f"数据解析 - 已导入: {stats.get('imported_parsed', 0)}")
        report_lines.append(f"数据解析 - 重复跳过: {stats.get('duplicate_parsed', 0)}")
        report_lines.append(f"数据解析 - 处理失败: {stats.get('error_parsed', 0)}")
        report_lines.append("")

        # 数据质量检查
        report_lines.append("✅ 数据质量检查")
        report_lines.append("-" * 40)
        report_lines.append(f"分数范围异常: {stats.get('out_of_range_scores', 0)}")
        report_lines.append("")

        # 数据完整性验证
        report_lines.append("🔍 数据完整性验证")
        report_lines.append("-" * 40)
        if integrity.get('issues'):
            for issue in integrity['issues']:
                report_lines.append(f"❌ {issue}")
        else:
            report_lines.append("✅ 所有完整性检查通过")

        report_lines.append("")
        report_lines.append(f"孤立学生记录: {integrity.get('orphan_students', 0)}")
        report_lines.append(f"孤立班级记录: {integrity.get('orphan_classes', 0)}")
        report_lines.append(f"重复记录组数: {integrity.get('duplicates', 0)}")
        report_lines.append(f"无效分数记录: {integrity.get('invalid_scores', 0)}")
        report_lines.append("")

        # 计算成功率
        total_import = stats.get('total_import', 0)
        total_production = stats.get('total_production', 0)
        success_rate = (total_production / total_import * 100) if total_import > 0 else 0

        report_lines.append("📈 迁移成功率")
        report_lines.append("-" * 40)
        report_lines.append(".1f")
        report_lines.append("")

        # 结论
        report_lines.append("🎯 迁移结论")
        report_lines.append("-" * 40)
        if not integrity.get('issues') and success_rate >= 95:
            report_lines.append("✅ 迁移成功：数据完整，质量良好")
        elif success_rate >= 80:
            report_lines.append("⚠️ 迁移基本成功：存在少量问题，建议检查")
        else:
            report_lines.append("❌ 迁移失败：存在严重问题，需要重新处理")

        report_lines.append("="*80)

        return "\n".join(report_lines)

    def export_error_details(self) -> None:
        """导出错误详情到CSV文件"""
        try:
            with self.connection.cursor(DictCursor) as cursor:
                # 导出解析失败的记录
                sql_parse_errors = f"""
                SELECT ti.student_no, ti.academic_year_str, ti.semester_str,
                       ti.monthly_scores, ti.error_msg as import_error
                FROM {self.temp_import_table} ti
                WHERE ti.import_status = 'error'
                """

                cursor.execute(sql_parse_errors)
                parse_errors = cursor.fetchall()

                if parse_errors:
                    with open('migration_parse_errors.csv', 'w', newline='', encoding='utf-8-sig') as f:
                        if parse_errors:
                            writer = csv.DictWriter(f, fieldnames=parse_errors[0].keys())
                            writer.writeheader()
                            writer.writerows(parse_errors)
                    logger.info(f"导出{len(parse_errors)}条解析错误记录到 migration_parse_errors.csv")

                # 导出关联失败的记录
                sql_link_errors = f"""
                SELECT tp.student_no, tp.academic_year, tp.semester, tp.month,
                       tp.score, tp.error_msg as link_error
                FROM {self.temp_parsed_table} tp
                WHERE tp.processing_status = 'error'
                """

                cursor.execute(sql_link_errors)
                link_errors = cursor.fetchall()

                if link_errors:
                    with open('migration_link_errors.csv', 'w', newline='', encoding='utf-8-sig') as f:
                        if link_errors:
                            writer = csv.DictWriter(f, fieldnames=link_errors[0].keys())
                            writer.writeheader()
                            writer.writerows(link_errors)
                    logger.info(f"导出{len(link_errors)}条关联错误记录到 migration_link_errors.csv")

        except Exception as e:
            logger.error(f"导出错误详情失败: {e}")

    def cleanup_temp_tables(self) -> bool:
        """清理临时表"""
        if not self.cleanup:
            logger.info("跳过清理操作（使用--cleanup参数启用）")
            return True

        try:
            logger.warning("开始清理临时表数据...")
            with self.connection.cursor() as cursor:
                # 删除临时表
                cursor.execute(f"DROP TABLE IF EXISTS {self.temp_parsed_table}")
                cursor.execute(f"DROP TABLE IF EXISTS {self.temp_import_table}")

                # 可选：删除迁移日志表
                cursor.execute("DROP TABLE IF EXISTS temp_conduct_score_migration_log")

                self.connection.commit()

            logger.info("临时表清理完成")
            return True

        except Exception as e:
            logger.error(f"清理临时表失败: {e}")
            self.connection.rollback()
            return False

    def run(self) -> bool:
        """执行验证和清理流程"""
        start_time = time.time()

        try:
            self.connect_db()

            logger.info("开始迁移验证和清理")

            # 生成报告
            report = self.generate_detailed_report()
            print("\n" + report + "\n")

            # 保存报告到文件
            report_file = f"migration_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
            with open(report_file, 'w', encoding='utf-8') as f:
                f.write(report)
            logger.info(f"详细报告已保存到: {report_file}")

            # 导出错误详情
            self.export_error_details()

            # 清理临时表（如果指定了cleanup参数）
            cleanup_success = self.cleanup_temp_tables()

            return cleanup_success

        except Exception as e:
            logger.error(f"验证和清理过程出现错误: {e}")
            return False

        finally:
            self.disconnect_db()
            elapsed_time = time.time() - start_time
            logger.info(f"验证总耗时: {elapsed_time:.1f}秒")
def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='学生操行分历史数据 - 验证与清理')
    parser.add_argument('--cleanup', action='store_true',
                       help='清理临时表数据（默认只验证不清理）')

    args = parser.parse_args()

    logger.info("="*60)
    logger.info("学生操行分历史数据迁移 - 脚本5：验证与清理")
    if args.cleanup:
        logger.warning("⚠️ 已启用清理模式，将删除临时表数据")
    else:
        logger.info("ℹ️ 验证模式：只检查数据，不清理临时表")
    logger.info("="*60)

    # 创建logs目录
    import os
    os.makedirs("logs", exist_ok=True)

    # 执行验证和清理
    validator = MigrationValidator(cleanup=args.cleanup)
    success = validator.run()

    if success:
        if args.cleanup:
            logger.info("✅ 验证和清理处理完成")
        else:
            logger.info("✅ 验证处理完成")
        sys.exit(0)
    else:
        logger.error("❌ 验证和清理处理失败")
        sys.exit(1)

if __name__ == "__main__":
    main()