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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
学生操行分历史数据迁移 - 脚本2：数据解析处理（Python 3.6兼容版）

功能说明：
- 解析学年格式：2015-2016 → 2015
- 解析学期格式：第一学期→1，第二学期→2
- 解析月度分数：一月:90,九月:85 → 多条月度记录
- 结果存储到temp_conduct_score_parsed中间表

修复内容：
- 添加数据库连接重试机制
- 添加连接健康检查和自动重连
- 定期检查连接状态，避免超时
- 改进错误处理和日志记录
- 支持断点续传

兼容性说明：
- 完全兼容Python 3.6
- 移除类型注解
- 使用.format()替代f-string
- 简化字符串格式化表达式

执行方式：
python 02_parse_score_data_py36_compatible.py

注意事项：
- 依赖脚本1已执行完成
- 支持断点续传（已解析的数据会跳过）
- 自动处理数据库连接问题
"""

# 完全兼容Python 3.4+的版本，不使用任何高级类型注解
import argparse
import logging
import sys
import time

import pymysql
from db_config import DB_CONFIG

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('02_parse_score_data.log', encoding='utf-8'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# 常量配置
CONSTANTS = {
    "BATCH_SIZE": 1000,  # 每批处理1000条记录
    "MAX_RETRIES": 3,    # 最大重试次数
    "RETRY_DELAY": 5,    # 重试间隔（秒）
}

# 数据库表配置
TABLE_CONFIG = {
    "temp_import": "temp_conduct_score_import",      # 临时导入表
    "temp_parsed": "temp_conduct_score_parsed",      # 解析结果表
}


class ConductScoreParser:
    """学生操行分数据解析器（Python 3.6兼容版）"""

    def __init__(self, db_config):
        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:
                    # 简化f-string表达式，避免Python 3.6兼容性问题
                    attempt_info = "数据库连接失败 (尝试 {}/{})".format(attempt + 1, self.max_retries + 1)
                    logger.warning("{}: {}".format(attempt_info, e))
                    logger.info("等待 {} 秒后重试...".format(retry_delay))
                    time.sleep(retry_delay)
                    retry_delay = min(retry_delay * 2, 60)  # 指数退避，最多60秒
                else:
                    logger.error("数据库连接失败，已达到最大重试次数: {}".format(e))
                    raise

    def disconnect_db(self):
        """断开数据库连接"""
        if self.connection:
            try:
                self.connection.close()
                logger.info("数据库连接已断开")
            except Exception as e:
                logger.warning("断开连接时出现警告: {}".format(e))
            finally:
                self.connection = None

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

        try:
            # 执行简单查询检查连接
            with self.connection.cursor() as cursor:
                cursor.execute("SELECT 1")
                cursor.fetchone()
            self.last_health_check = time.time()
            return True
        except Exception as e:
            logger.warning("连接健康检查失败: {}".format(e))
            return False

    def ensure_connection(self):
        """确保数据库连接有效，必要时重连"""
        current_time = time.time()

        # 检查是否需要定期健康检查
        if current_time - self.last_health_check > self.health_check_interval:
            if not self.check_connection_health():
                logger.info("连接健康检查失败，尝试重连...")
                self.disconnect_db()
                self.connect_db()
        elif not self.connection:
            # 连接不存在，直接连接
            self.connect_db()

    def get_pending_data(self):
        """获取待解析的数据"""
        max_retries = 3
        retry_delay = 2

        for attempt in range(max_retries):
            try:
                self.ensure_connection()
                with self.connection.cursor() as cursor:
                    # 获取未解析或解析失败的数据
                    sql = """
                    SELECT temp_id, student_no, academic_year, semester, conduct_score
                    FROM {}
                    WHERE import_status IN ('pending', 'error')
                    ORDER BY temp_id
                    LIMIT {}
                    """.format(self.temp_import_table, self.batch_size)

                    cursor.execute(sql)
                    columns = [desc[0] for desc in cursor.description]
                    rows = cursor.fetchall()

                    # 转换为字典列表
                    data = []
                    for row in rows:
                        data.append(dict(zip(columns, row)))

                    logger.info("获取到 {} 条待解析记录".format(len(data)))
                    return data

            except Exception as e:
                if attempt < max_retries - 1:
                    attempt_info = "获取待解析数据失败 (尝试 {}/{})".format(attempt + 1, max_retries)
                    logger.warning("{}: {}".format(attempt_info, e))
                    time.sleep(retry_delay)
                    retry_delay *= 2
                else:
                    logger.error("获取待解析数据失败，已达到最大重试次数: {}".format(e))
                    return []

        return []

    def parse_academic_year(self, year_str):
        """解析学年字符串，返回(年份, 错误信息)"""
        if not year_str or not isinstance(year_str, str):
            return None, "学年为空或格式错误: {}".format(year_str)

        year_str = year_str.strip()

        # 尝试匹配XXXX-XXXX格式
        if '-' in year_str:
            parts = year_str.split('-')
            if len(parts) == 2:
                try:
                    start_year = int(parts[0].strip())
                    end_year = int(parts[1].strip())

                    # 检查年份跨度是否合理（应该相差1年）
                    if end_year - start_year == 1:
                        # 检查年份范围是否合理
                        if 2000 <= start_year <= 2030:
                            return start_year, None
                        else:
                            return None, "学年范围不合理: {}".format(year_str)
                    else:
                        return None, "学年跨度不正确: {}".format(year_str)
                except ValueError:
                    return None, "学年格式不正确，应为XXXX-XXXX格式: {}".format(year_str)
            else:
                return None, "学年格式不正确，应为XXXX-XXXX格式: {}".format(year_str)
        else:
            # 尝试直接转换为整数
            try:
                year = int(year_str)
                if 2000 <= year <= 2030:
                    return year, None
                else:
                    return None, "学年范围不合理: {}".format(year_str)
            except ValueError:
                return None, "学年格式不正确: {}".format(year_str)

    def parse_semester(self, semester_str):
        """解析学期字符串，返回(学期编号, 错误信息)"""
        if not semester_str or not isinstance(semester_str, str):
            return None, "学期为空或格式错误: {}".format(semester_str)

        semester = semester_str.strip()

        # 映射学期名称到编号
        semester_mapping = {
            "第一学期": 1,
            "第二学期": 2,
            "上学期": 1,
            "下学期": 2,
            "春季学期": 1,
            "秋季学期": 2
        }

        if semester in semester_mapping:
            return semester_mapping[semester], None
        else:
            return None, "学期格式不正确，应为'第一学期'或'第二学期': {}".format(semester)

    def parse_monthly_scores(self, score_str):
        """解析月度分数字符串，返回(分数列表, 错误信息)"""
        if not score_str or not isinstance(score_str, str):
            return [], "平时分数为空或格式错误: {}".format(score_str)

        score_str = score_str.strip()
        monthly_scores = []
        error_msgs = []

        # 按逗号分割多个月份的分数
        parts = [part.strip() for part in score_str.split(',') if part.strip()]

        # 月度映射
        month_mapping = {
            "一月": 1, "二月": 2, "三月": 3, "四月": 4, "五月": 5, "六月": 6,
            "七月": 7, "八月": 8, "九月": 9, "十月": 10, "十一月": 11, "十二月": 12,
            "1月": 1, "2月": 2, "3月": 3, "4月": 4, "5月": 5, "6月": 6,
            "7月": 7, "8月": 8, "9月": 9, "10月": 10, "11月": 11, "12月": 12
        }

        for part in parts:
            if ':' not in part:
                error_msgs.append("分数格式错误: {}".format(part))
                continue

            month_part, score_part = part.split(':', 1)
            month_name = month_part.strip()
            score_value = score_part.strip()

            # 解析月份
            if month_name not in month_mapping:
                error_msgs.append("月份名称不正确: {}".format(month_name))
                continue

            month_num = month_mapping[month_name]

            # 解析分数
            try:
                score = float(score_value)
                if 0 <= score <= 100:
                    monthly_scores.append((month_num, score))
                else:
                    error_msgs.append("分数范围不正确(0-100): {}".format(score))
            except ValueError:
                error_msgs.append("分数格式不是数字: {}".format(score_value))

        if error_msgs:
            return monthly_scores, "; ".join(error_msgs)
        else:
            return monthly_scores, None

    def insert_parsed_data(self, parsed_records):
        """插入解析后的数据"""
        if not parsed_records:
            return 0, 0

        max_retries = 3
        retry_delay = 2
        total_inserted = 0
        total_failed = 0

        for attempt in range(max_retries):
            try:
                self.ensure_connection()
                with self.connection.cursor() as cursor:
                    # 批量插入解析结果
                    sql = """
                    INSERT INTO {} (
                        temp_id, student_no, academic_year, semester,
                        score_month, conduct_score, parse_status, error_message
                    ) VALUES {}
                    """.format(
                        self.temp_parsed_table,
                        ", ".join(["(%s, %s, %s, %s, %s, %s, %s, %s)"] * len(parsed_records))
                    )

                    # 准备参数
                    params = []
                    for record in parsed_records:
                        params.extend([
                            record['temp_id'],
                            record['student_no'],
                            record['academic_year'],
                            record['semester'],
                            record['score_month'],
                            record['conduct_score'],
                            record['parse_status'],
                            record['error_message']
                        ])

                    cursor.execute(sql, params)
                    self.connection.commit()

                    total_inserted = len(parsed_records)
                    logger.info("成功插入 {} 条解析记录".format(total_inserted))
                    return total_inserted, 0

            except Exception as e:
                total_failed = len(parsed_records)
                if attempt < max_retries - 1:
                    attempt_info = "插入解析数据失败 (尝试 {}/{})".format(attempt + 1, max_retries)
                    logger.warning("{}: {}".format(attempt_info, e))
                    time.sleep(retry_delay)
                    retry_delay *= 2
                else:
                    logger.error("插入解析数据失败，已达到最大重试次数: {}".format(e))
                    return 0, total_failed

        return total_inserted, total_failed

    def update_import_status(self, temp_ids, status, error_msg=None):
        """更新导入状态"""
        if not temp_ids:
            return

        max_retries = 3
        retry_delay = 2

        for attempt in range(max_retries):
            try:
                self.ensure_connection()
                with self.connection.cursor() as cursor:
                    # 批量更新状态
                    placeholders = ", ".join(["%s"] * len(temp_ids))

                    if error_msg:
                        sql = """
                        UPDATE {} SET import_status = %s, error_message = %s, updated_at = NOW()
                        WHERE temp_id IN ({})
                        """.format(self.temp_import_table, placeholders)
                        cursor.execute(sql, [status, error_msg] + temp_ids)
                    else:
                        sql = """
                        UPDATE {} SET import_status = %s, error_message = NULL, updated_at = NOW()
                        WHERE temp_id IN ({})
                        """.format(self.temp_import_table, placeholders)
                        cursor.execute(sql, [status] + temp_ids)

                    self.connection.commit()
                    affected_rows = cursor.rowcount
                    logger.info("更新 {} 条记录状态为: {}".format(affected_rows, status))
                    return

            except Exception as e:
                if attempt < max_retries - 1:
                    attempt_info = "更新导入状态失败 (尝试 {}/{})".format(attempt + 1, max_retries)
                    logger.warning("{}: {}".format(attempt_info, e))
                    time.sleep(retry_delay)
                    retry_delay *= 2
                else:
                    logger.error("更新导入状态失败，已达到最大重试次数: {}".format(e))

    def process_batch(self, batch_data):
        """处理一批数据"""
        processed = 0
        success = 0
        error = 0
        parsed_records = []

        for record in batch_data:
            processed += 1
            temp_id = record['temp_id']
            student_no = record['student_no']

            try:
                # 解析学年
                academic_year, year_error = self.parse_academic_year(record['academic_year'])
                if year_error:
                    logger.warning("学号{}: {}".format(student_no, year_error))
                    error += 1
                    continue

                # 解析学期
                semester, semester_error = self.parse_semester(record['semester'])
                if semester_error:
                    logger.warning("学号{}: {}".format(student_no, semester_error))
                    error += 1
                    continue

                # 解析月度分数
                monthly_scores, score_error = self.parse_monthly_scores(record['conduct_score'])
                if score_error:
                    logger.warning("学号{}: {}".format(student_no, score_error))
                    error += 1
                    continue

                # 生成解析记录
                for month, score in monthly_scores:
                    parsed_records.append({
                        'temp_id': temp_id,
                        'student_no': student_no,
                        'academic_year': academic_year,
                        'semester': semester,
                        'score_month': month,
                        'conduct_score': score,
                        'parse_status': 'success',
                        'error_message': None
                    })

                success += 1

            except Exception as e:
                logger.error("处理记录时出现异常 (学号: {}): {}".format(student_no, e))
                error += 1

        return {
            'processed': processed,
            'success': success,
            'error': error,
            'parsed_records': parsed_records
        }

    def run(self):
        """主运行函数"""
        start_time = time.time()
        logger.info("开始学生操行分数据解析...")
        logger.info("批量大小: {}, 重试次数: {}".format(self.batch_size, self.max_retries))

        try:
            # 连接数据库
            self.connect_db()

            total_processed = 0
            total_success = 0
            total_error = 0
            total_parsed_records = 0
            batch_count = 0

            while True:
                batch_count += 1

                # 获取待处理数据
                batch_data = self.get_pending_data()
                if not batch_data:
                    logger.info("没有更多待处理数据，解析完成")
                    break

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

                # 处理当前批次
                logger.info("处理批次 {}: {}条记录".format(batch_count, len(batch_data)))
                result = self.process_batch(batch_data)

                # 插入解析结果
                if result['parsed_records']:
                    inserted, failed = self.insert_parsed_data(result['parsed_records'])
                    if failed > 0:
                        logger.error("批次 {} 插入失败 {} 条记录".format(batch_count, failed))

                # 更新导入状态
                temp_ids = [record['temp_id'] for record in batch_data]
                if result['success'] > 0:
                    success_ids = temp_ids[:result['success']]
                    self.update_import_status(success_ids, 'completed')

                if result['error'] > 0:
                    error_ids = temp_ids[result['success']:result['success'] + result['error']]
                    self.update_import_status(error_ids, 'error', '解析失败')

                # 统计信息
                logger.info("批次 {} 结果: 处理{}, 成功{}, 失败{}, 生成{}条月度记录".format(
                    batch_count, result['processed'], result['success'],
                    result['error'], len(result['parsed_records'])
                ))

                # 累计统计
                total_processed += result['processed']
                total_success += result['success']
                total_error += result['error']
                total_parsed_records += len(result['parsed_records'])

            # 输出最终统计
            elapsed_time = time.time() - start_time
            logger.info("数据解析完成!")
            logger.info("  处理总记录: {}".format(total_processed))
            logger.info("  解析成功: {}".format(total_success))
            logger.info("  解析失败: {}".format(total_error))
            logger.info("  生成月度记录: {}".format(total_parsed_records))
            logger.info("解析总耗时: {:.1f}秒".format(elapsed_time))

            return True

        except KeyboardInterrupt:
            logger.info("用户中断解析过程")
            return False
        except Exception as e:
            elapsed_time = time.time() - start_time
            logger.error("解析过程出现错误: {}".format(e))
            return False
        finally:
            self.disconnect_db()
            elapsed_time = time.time() - start_time
            logger.info("解析总耗时: {:.1f}秒".format(elapsed_time))


def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='学生操行分数据解析工具')
    parser.add_argument('--config', help='数据库配置文件路径')
    parser.add_argument('--batch-size', type=int, default=1000, help='批处理大小')
    parser.add_argument('--max-retries', type=int, default=3, help='最大重试次数')

    args = parser.parse_args()

    # 更新配置
    if args.batch_size:
        CONSTANTS["BATCH_SIZE"] = args.batch_size

    # 创建解析器并运行
    parser = ConductScoreParser(DB_CONFIG)
    success = parser.run()

    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()