426 lines
16 KiB
Python
426 lines
16 KiB
Python
#!/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()
|