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

#!/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()