提交

2026-01-04 10:00:08 +08:00
1 parent 48f15745be
commit e421030ada
@@ -0,0 +1,494 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+学生操行分历史数据迁移 - 脚本2：数据解析处理（修复版）
+
+功能说明：
+- 解析学年格式：2015-2016 → 2015
+- 解析学期格式：第一学期→1，第二学期→2
+- 解析月度分数：一月:90,九月:85 → 多条月度记录
+- 结果存储到temp_conduct_score_parsed中间表
+
+修复内容：
+- 添加数据库连接重试机制
+- 添加连接健康检查和自动重连
+- 定期检查连接状态，避免超时
+- 改进错误处理和日志记录
+- 支持断点续传
+
+执行方式：
+python 02_parse_score_data_fixed.py
+
+注意事项：
+- 依赖脚本1已执行完成
+- 支持断点续传（已解析的数据会跳过）
+- 自动处理数据库连接问题
+"""
+
+# from __future__ import annotations  # Python 3.7+ only
+
+import argparse
+import logging
+import re
+import sys
+import time
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple
+
+import pymysql
+from pymysql.cursors import DictCursor
+
+from db_config import DB_CONFIG, TABLE_CONFIG, CONSTANTS, MONTH_MAPPING, SEMESTER_MAPPING
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(message)s",
+    handlers=[
+        logging.StreamHandler(sys.stdout),
+        logging.FileHandler("logs/02_parse_data_fixed.log", mode='a', encoding='utf-8')
+    ]
+)
+logger = logging.getLogger("parse_data_fixed")
+
+class DataParserFixed:
+    """数据解析处理器（修复版）"""
+
+    def __init__(self):
+        self.db_config = DB_CONFIG
+        self.temp_import_table = TABLE_CONFIG["temp_import"]
+        self.temp_parsed_table = TABLE_CONFIG["temp_parsed"]
+        self.batch_size = CONSTANTS["BATCH_SIZE"]
+        self.max_retries = self.db_config.get('max_retries', 3)
+        self.connection = None
+        self.last_health_check = time.time()
+        self.health_check_interval = 300  # 5分钟检查一次连接健康状态
+
+    def connect_db(self):
+        """连接数据库（带重试机制）"""
+        retry_delay = 5  # 每次重试间隔5秒
+
+        for attempt in range(self.max_retries + 1):
+            try:
+                self.connection = pymysql.connect(**self.db_config)
+                logger.info("数据库连接成功")
+                self.last_health_check = time.time()
+                return
+            except Exception as e:
+                if attempt < self.max_retries:
+                    logger.warning(f"数据库连接失败 (尝试 {attempt + 1}/{self.max_retries + 1}): {e}")
+                    logger.info(f"等待 {retry_delay} 秒后重试...")
+                    time.sleep(retry_delay)
+                    retry_delay = min(retry_delay * 2, 60)  # 指数退避，最多60秒
+                else:
+                    logger.error(f"数据库连接失败，已达到最大重试次数: {e}")
+                    raise
+
+    def disconnect_db(self):
+        """断开数据库连接"""
+        if self.connection:
+            try:
+                self.connection.close()
+                logger.info("数据库连接已断开")
+            except Exception as e:
+                logger.warning(f"断开连接时出现警告: {e}")
+
+    def check_connection_health(self):
+        """检查数据库连接健康状态"""
+        if not self.connection:
+            return False
+
+        try:
+            # 发送一个简单的查询来测试连接
+            with self.connection.cursor() as cursor:
+                cursor.execute("SELECT 1 as health_check")
+                result = cursor.fetchone()
+                if result and result[0] == 1:
+                    self.last_health_check = time.time()
+                    return True
+                else:
+                    logger.warning("连接健康检查返回异常结果")
+                    return False
+        except Exception as e:
+            logger.warning(f"连接健康检查失败: {e}")
+            return False
+
+    def ensure_connection(self):
+        """确保数据库连接可用，如果断开则重连"""
+        current_time = time.time()
+
+        # 如果距离上次健康检查超过指定时间，也进行检查
+        if current_time - self.last_health_check > self.health_check_interval:
+            logger.info("执行定时连接健康检查...")
+
+        if not self.check_connection_health():
+            logger.info("检测到连接断开或不健康，正在重新连接...")
+            self.disconnect_db()
+            self.connect_db()
+
+    def get_pending_data(self):
+        """获取待解析的数据"""
+        # 确保连接可用
+        self.ensure_connection()
+
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                with self.connection.cursor(DictCursor) as cursor:
+                    sql = f"""
+                    SELECT id, student_no, academic_year_str, semester_str, monthly_scores
+                    FROM {self.temp_import_table}
+                    WHERE import_status = 'pending'
+                    ORDER BY id
+                    LIMIT {self.batch_size}
+                    """
+
+                    cursor.execute(sql)
+                    return cursor.fetchall()
+
+            except Exception as e:
+                logger.warning(f"获取待解析数据失败 (尝试 {attempt + 1}/{max_retries}): {e}")
+                if attempt < max_retries - 1:
+                    self.ensure_connection()  # 重新连接后重试
+                else:
+                    logger.error(f"获取待解析数据失败，已达到最大重试次数: {e}")
+                    return []
+
+    def parse_academic_year(self, year_str):
+        """解析学年格式：2015-2016 → 2015"""
+        if not year_str or not isinstance(year_str, str):
+            return None, f"学年为空或格式错误: {year_str}"
+
+        # 匹配XXXX-XXXX格式
+        match = re.match(r'^(\d{4})-(\d{4})$', year_str.strip())
+        if not match:
+            return None, f"学年格式不正确，应为XXXX-XXXX格式: {year_str}"
+
+        start_year = int(match.group(1))
+        end_year = int(match.group(2))
+
+        # 验证年份合理性
+        if end_year != start_year + 1:
+            return None, f"学年跨度不正确: {year_str}"
+
+        if start_year < 2000 or start_year > 2030:
+            return None, f"学年范围不合理: {year_str}"
+
+        return start_year, None
+
+    def parse_semester(self, semester_str):
+        """解析学期格式：第一学期→1，第二学期→2"""
+        if not semester_str or not isinstance(semester_str, str):
+            return None, f"学期为空或格式错误: {semester_str}"
+
+        semester = semester_str.strip()
+        if semester in SEMESTER_MAPPING:
+            return SEMESTER_MAPPING[semester], None
+        else:
+            return None, f"学期格式不正确，应为'第一学期'或'第二学期': {semester}"
+
+    def parse_monthly_scores(self, score_str):
+        """解析月度分数：一月:90,九月:85 → [(1, 90.0), (9, 85.0), ...]"""
+        if not score_str or not isinstance(score_str, str):
+            return [], f"平时分数为空或格式错误: {score_str}"
+
+        score_str = score_str.strip()
+        if not score_str:
+            return [], "平时分数为空"
+
+        parsed_scores = []
+        error_msgs = []
+
+        # 分割多个月份的分数
+        score_parts = score_str.split(',')
+        for part in score_parts:
+            part = part.strip()
+            if not part:
+                continue
+
+            # 匹配"月份:分数"格式
+            match = re.match(r'^(.+?):(.+)$', part)
+            if not match:
+                error_msgs.append(f"分数格式错误: {part}")
+                continue
+
+            month_name = match.group(1).strip()
+            score_value = match.group(2).strip()
+
+            # 验证月份名称
+            if month_name not in MONTH_MAPPING:
+                error_msgs.append(f"月份名称不正确: {month_name}")
+                continue
+
+            month_num = MONTH_MAPPING[month_name]
+
+            # 验证分数（取消范围限制）
+            try:
+                score = float(score_value)
+                parsed_scores.append((month_num, score))
+            except ValueError:
+                error_msgs.append(f"分数格式不是数字: {score_value}")
+                continue
+
+        if error_msgs:
+            return parsed_scores, "; ".join(error_msgs)
+        elif not parsed_scores:
+            return [], "未解析到任何有效分数"
+        else:
+            return parsed_scores, None
+
+    def insert_parsed_data(self, parsed_records):
+        """插入解析后的数据（带重试机制）"""
+        if not parsed_records:
+            return 0, 0
+
+        # 确保连接可用
+        self.ensure_connection()
+
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                with self.connection.cursor() as cursor:
+                    sql = f"""
+                    INSERT INTO {self.temp_parsed_table} (
+                        temp_id, student_no, academic_year_str, semester_str,
+                        academic_year, semester, month, score, processing_status
+                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, 'parsed')
+                    """
+
+                    values = []
+                    for record in parsed_records:
+                        values.append((
+                            record['temp_id'],
+                            record['student_no'],
+                            record['academic_year_str'],
+                            record['semester_str'],
+                            record['academic_year'],
+                            record['semester'],
+                            record['month'],
+                            record['score']
+                        ))
+
+                    cursor.executemany(sql, values)
+                    self.connection.commit()
+
+                    return len(parsed_records), 0
+
+            except Exception as e:
+                logger.warning(f"插入解析数据失败 (尝试 {attempt + 1}/{max_retries}): {e}")
+                if attempt < max_retries - 1:
+                    self.ensure_connection()  # 重新连接后重试
+                    time.sleep(2)  # 等待2秒再重试
+                else:
+                    logger.error(f"插入解析数据失败，已达到最大重试次数: {e}")
+                    try:
+                        self.connection.rollback()
+                    except:
+                        pass
+                    return 0, len(parsed_records)
+
+    def update_import_status(self, temp_ids, status, error_msg=None):
+        """更新导入表的状态（带重试机制）"""
+        # 确保连接可用
+        self.ensure_connection()
+
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                with self.connection.cursor() as cursor:
+                    if error_msg:
+                        sql = f"""
+                        UPDATE {self.temp_import_table}
+                        SET import_status = %s, error_msg = %s, update_time = NOW()
+                        WHERE id IN ({','.join(['%s'] * len(temp_ids))})
+                        """
+                        cursor.execute(sql, [status, error_msg] + temp_ids)
+                    else:
+                        sql = f"""
+                        UPDATE {self.temp_import_table}
+                        SET import_status = %s, update_time = NOW()
+                        WHERE id IN ({','.join(['%s'] * len(temp_ids))})
+                        """
+                        cursor.execute(sql, [status] + temp_ids)
+
+                    self.connection.commit()
+                    return
+
+            except Exception as e:
+                logger.warning(f"更新导入状态失败 (尝试 {attempt + 1}/{max_retries}): {e}")
+                if attempt < max_retries - 1:
+                    self.ensure_connection()  # 重新连接后重试
+                    time.sleep(1)
+                else:
+                    logger.error(f"更新导入状态失败，已达到最大重试次数: {e}")
+                    try:
+                        self.connection.rollback()
+                    except:
+                        pass
+
+    def process_batch(self, batch_data):
+        """处理一批数据"""
+        parsed_records = []
+        success_ids = []
+        error_ids = []
+
+        for row in batch_data:
+            temp_id = row['id']
+            student_no = row['student_no']
+            academic_year_str = row['academic_year_str']
+            semester_str = row['semester_str']
+            monthly_scores = row['monthly_scores']
+
+            # 解析学年
+            academic_year, year_error = self.parse_academic_year(academic_year_str)
+            if year_error:
+                logger.warning(f"学号{student_no}: {year_error}")
+                error_ids.append(temp_id)
+                continue
+
+            # 解析学期
+            semester, semester_error = self.parse_semester(semester_str)
+            if semester_error:
+                logger.warning(f"学号{student_no}: {semester_error}")
+                error_ids.append(temp_id)
+                continue
+
+            # 解析月度分数
+            monthly_data, score_error = self.parse_monthly_scores(monthly_scores)
+            if score_error:
+                logger.warning(f"学号{student_no}: {score_error}")
+
+            # 如果解析出分数，创建记录
+            if monthly_data:
+                for month, score in monthly_data:
+                    parsed_records.append({
+                        'temp_id': temp_id,
+                        'student_no': student_no,
+                        'academic_year_str': academic_year_str,
+                        'semester_str': semester_str,
+                        'academic_year': academic_year,
+                        'semester': semester,
+                        'month': month,
+                        'score': score
+                    })
+                success_ids.append(temp_id)
+            else:
+                # 没有解析出有效分数，但学年学期正确，也算成功（只是没有月度数据）
+                if not score_error:
+                    success_ids.append(temp_id)
+                else:
+                    error_ids.append(temp_id)
+
+        # 插入解析后的数据
+        inserted, failed = self.insert_parsed_data(parsed_records)
+
+        # 更新状态
+        if success_ids:
+            self.update_import_status(success_ids, 'parsed')
+        if error_ids:
+            self.update_import_status(error_ids, 'error', '数据解析失败')
+
+        return {
+            'processed': len(batch_data),
+            'success': len(success_ids),
+            'error': len(error_ids),
+            'parsed_records': inserted,
+            'inserted': inserted,
+            'insert_failed': failed
+        }
+
+    def run(self):
+        """执行解析流程"""
+        start_time = time.time()
+        total_processed = 0
+        total_success = 0
+        total_error = 0
+        total_parsed_records = 0
+
+        try:
+            self.connect_db()
+
+            logger.info("开始数据解析处理（修复版）")
+            logger.info(f"批量大小: {self.batch_size}, 重试次数: {self.max_retries}")
+
+            batch_count = 0
+            while True:
+                batch_count += 1
+
+                # 每处理10批数据检查一次连接健康状态
+                if batch_count % 10 == 0:
+                    logger.info(f"已处理 {batch_count} 批数据，执行定期连接健康检查...")
+                    self.ensure_connection()
+
+                # 获取待处理数据
+                batch_data = self.get_pending_data()
+
+                if not batch_data:
+                    logger.info("没有更多待处理数据")
+                    break
+
+                logger.info(f"处理批次 {batch_count}: {len(batch_data)}条记录")
+
+                # 处理这一批数据
+                result = self.process_batch(batch_data)
+
+                total_processed += result['processed']
+                total_success += result['success']
+                total_error += result['error']
+                total_parsed_records += result['parsed_records']
+
+                logger.info(f"批次 {batch_count} 结果: 处理{result['processed']}, 成功{result['success']}, "
+                           f"失败{result['error']}, 生成{result['parsed_records']}条月度记录")
+
+                # 如果这一批都是错误，说明可能有问题，停止处理
+                if result['success'] == 0 and result['processed'] > 0:
+                    logger.warning("当前批次全部处理失败，可能存在数据质量问题，请检查日志")
+                    break
+
+            # 输出统计信息
+            logger.info("="*60)
+            logger.info("数据解析统计结果（修复版）:")
+            logger.info(f"  处理总记录: {total_processed}")
+            logger.info(f"  解析成功: {total_success}")
+            logger.info(f"  解析失败: {total_error}")
+            logger.info(f"  生成月度记录: {total_parsed_records}")
+            elapsed_time = time.time() - start_time
+            logger.info(f"解析总耗时: {elapsed_time:.1f}秒")
+            logger.info("="*60)
+
+            return total_error == 0
+
+        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='学生操行分历史数据 - 数据解析处理（修复版）')
+
+    logger.info("="*70)
+    logger.info("学生操行分历史数据迁移 - 脚本2：数据解析处理（修复版）")
+    logger.info("="*70)
+
+    # 创建logs目录
+    import os
+    os.makedirs("logs", exist_ok=True)
+
+    # 执行解析
+    parser = DataParserFixed()
+    success = parser.run()
+
+    if success:
+        logger.info("✅ 数据解析处理完成（修复版）")
+        sys.exit(0)
+    else:
+        logger.error("❌ 数据解析处理失败（修复版）")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()