all-project/学校学生操行历史数据迁移/02_parse_score_data.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.py

注意事项：
- 依赖脚本1已执行完成
- 支持断点续传（已解析的数据会跳过）
"""

from __future__ import annotations

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.log", mode='a', encoding='utf-8')
    ]
)
logger = logging.getLogger("parse_data")

class DataParser:
    """数据解析处理器"""

    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.connection = None

    def connect_db(self) -> None:
        """连接数据库（带重试机制）"""
        max_retries = self.db_config.get('max_retries', 3)
        retry_delay = 5  # 每次重试间隔5秒

        for attempt in range(max_retries + 1):
            try:
                self.connection = pymysql.connect(**self.db_config)
                logger.info("数据库连接成功")
                return
            except Exception as e:
                if attempt < max_retries:
                    logger.warning(f"数据库连接失败 (尝试 {attempt + 1}/{max_retries + 1}): {e}")
                    logger.info(f"等待 {retry_delay} 秒后重试...")
                    time.sleep(retry_delay)
                    retry_delay *= 2  # 指数退避
                else:
                    logger.error(f"数据库连接失败，已达到最大重试次数: {e}")
                    raise

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

    def check_connection_health(self) -> bool:
        """检查数据库连接健康状态"""
        if not self.connection:
            return False

        try:
            with self.connection.cursor() as cursor:
                cursor.execute("SELECT 1")
                cursor.fetchone()
            return True
        except Exception as e:
            logger.warning(f"连接健康检查失败: {e}")
            return False

    def ensure_connection(self) -> None:
        """确保数据库连接可用，如果断开则重连"""
        if not self.check_connection_health():
            logger.info("检测到连接断开，正在重新连接...")
            self.disconnect_db()
            self.connect_db()

    def get_pending_data(self) -> List[Dict[str, Any]]:
        """获取待解析的数据"""
        # 确保连接可用
        self.ensure_connection()

        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.error(f"获取待解析数据失败: {e}")
            return []

    def parse_academic_year(self, year_str: str) -> Tuple[Optional[int], Optional[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: str) -> Tuple[Optional[int], Optional[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: str) -> Tuple[List[Tuple[int, float]], Optional[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)
                # 取消分数范围限制，允许任何数值分数
            except ValueError:
                error_msgs.append(f"分数格式不是数字: {score_value}")
                continue

            parsed_scores.append((month_num, score))

        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: List[Dict[str, Any]]) -> tuple[int, int]:
        """插入解析后的数据"""
        if not parsed_records:
            return 0, 0

        # 确保连接可用
        self.ensure_connection()

        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.error(f"插入解析数据失败: {e}")
            self.connection.rollback()
            return 0, len(parsed_records)

    def update_import_status(self, temp_ids: List[int], status: str, error_msg: str = None) -> None:
        """更新导入表的状态"""
        # 确保连接可用
        self.ensure_connection()

        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()

        except Exception as e:
            logger.error(f"更新导入状态失败: {e}")
            self.connection.rollback()

    def process_batch(self, batch_data: List[Dict[str, Any]]) -> Dict[str, Any]:
        """处理一批数据"""
        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) -> bool:
        """执行解析流程"""
        start_time = time.time()
        total_processed = 0
        total_success = 0
        total_error = 0
        total_parsed_records = 0

        try:
            self.connect_db()

            logger.info("开始数据解析处理")

            batch_count = 0
            while True:
                batch_count += 1

                # 每处理10批数据检查一次连接健康状态
                if batch_count % 10 == 0:
                    logger.info("执行定期连接健康检查...")
                    self.ensure_connection()

                # 获取待处理数据
                batch_data = self.get_pending_data()

                if not batch_data:
                    logger.info("没有更多待处理数据")
                    break

                logger.info(f"处理批次: {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"批次结果: 处理{result['processed']}, 成功{result['success']}, "
                           f"失败{result['error']}, 生成{result['parsed_records']}条月度记录")

                # 如果这一批都是错误，说明可能有问题，停止处理
                if result['success'] == 0 and result['processed'] > 0:
                    logger.warning("当前批次全部处理失败，可能存在数据质量问题，请检查日志")
                    break

            # 输出统计信息
            logger.info("="*50)
            logger.info("数据解析统计结果:")
            logger.info(f"  处理总记录: {total_processed}")
            logger.info(f"  解析成功: {total_success}")
            logger.info(f"  解析失败: {total_error}")
            logger.info(f"  生成月度记录: {total_parsed_records}")
            logger.info(f"解析总耗时: {elapsed_time:.1f}秒")
            logger.info("="*50)

            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("="*60)
    logger.info("学生操行分历史数据迁移 - 脚本2：数据解析处理")
    logger.info("="*60)

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

    # 执行解析
    parser = DataParser()
    success = parser.run()

    if success:
        logger.info("✅ 数据解析处理完成")
        sys.exit(0)
    else:
        logger.error("❌ 数据解析处理失败")
        sys.exit(1)

if __name__ == "__main__":
    main()