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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
学生操行分历史数据迁移 - 脚本1：Excel数据导入临时表

功能说明：
- 读取操行.xls文件
- 数据清洗和验证
- 导入到temp_conduct_score_import临时表

执行方式：
python 01_import_excel_to_temp.py

注意事项：
- 执行前确保Excel文件存在
- 执行前确保临时表已创建
- 支持断点续传（已导入的数据会跳过）
"""

from __future__ import annotations

import argparse
import logging
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd
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/01_import_excel.log", mode='a', encoding='utf-8')
    ]
)
logger = logging.getLogger("import_excel")

class ExcelImporter:
    """Excel导入处理器"""

    def __init__(self, excel_file: str):
        self.excel_file = excel_file
        self.db_config = DB_CONFIG
        self.temp_table = TABLE_CONFIG["temp_import"]
        self.batch_size = CONSTANTS["BATCH_SIZE"]
        self.connection = None

    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 read_excel_data(self) -> pd.DataFrame:
        """读取Excel文件"""
        try:
            logger.info(f"开始读取Excel文件: {self.excel_file}")

            # 检查文件是否存在
            if not Path(self.excel_file).exists():
                raise FileNotFoundError(f"Excel文件不存在: {self.excel_file}")

            # 读取Excel文件
            df = pd.read_excel(self.excel_file)

            logger.info(f"Excel文件读取成功，共{len(df)}行数据")
            logger.info(f"列名: {list(df.columns)}")

            return df

        except Exception as e:
            logger.error(f"读取Excel文件失败: {e}")
            raise

    def validate_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """数据验证和清洗"""
        logger.info("开始数据验证和清洗")

        # 检查必需列
        required_columns = ['学号', '学年', '学期', '平时分数']
        missing_columns = [col for col in required_columns if col not in df.columns]
        if missing_columns:
            raise ValueError(f"Excel文件缺少必需列: {missing_columns}")

        # 数据清洗
        original_count = len(df)

        # 删除空行（学号为空的行）
        df = df.dropna(subset=['学号'])
        logger.info(f"删除空学号行: {original_count - len(df)}行")

        # 去除学号前后的空格
        df.loc[:, '学号'] = df['学号'].astype(str).str.strip()

        # 删除重复的学号+学年+学期记录（保留第一条）
        before_dedupe = len(df)
        df = df.drop_duplicates(subset=['学号', '学年', '学期'], keep='first')
        logger.info(f"删除重复记录: {before_dedupe - len(df)}行")

        # 验证学年格式
        df.loc[:, 'academic_year_valid'] = df['学年'].astype(str).str.match(r'^\d{4}-\d{4}$')
        invalid_years = df[~df['academic_year_valid']]
        if len(invalid_years) > 0:
            logger.warning(f"发现{len(invalid_years)}条学年格式不正确的记录:")
            for idx, row in invalid_years.head(5).iterrows():
                logger.warning(f"  行{idx+2}: 学号={row['学号']}, 学年={row['学年']}")

        # 验证学期格式
        valid_semesters = ['第一学期', '第二学期']
        df.loc[:, 'semester_valid'] = df['学期'].isin(valid_semesters)
        invalid_semesters = df[~df['semester_valid']]
        if len(invalid_semesters) > 0:
            logger.warning(f"发现{len(invalid_semesters)}条学期格式不正确的记录:")
            for idx, row in invalid_semesters.head(5).iterrows():
                logger.warning(f"  行{idx+2}: 学号={row['学号']}, 学期={row['学期']}")

        # 验证平时分数格式（非空且包含冒号）
        df.loc[:, 'monthly_scores_valid'] = df['平时分数'].astype(str).str.contains(':')
        invalid_scores = df[~df['monthly_scores_valid']]
        if len(invalid_scores) > 0:
            logger.warning(f"发现{len(invalid_scores)}条平时分数格式不正确的记录:")
            for idx, row in invalid_scores.head(5).iterrows():
                logger.warning(f"  行{idx+2}: 学号={row['学号']}, 平时分数={row['平时分数']}")

        # 验证数值范围
        df.loc[:, 'final_score_valid'] = True
        final_scores = pd.to_numeric(df['最终分'], errors='coerce')

        # 检查最终分范围 (decimal(5,2) 最大999.99)
        invalid_final_scores = final_scores[(final_scores < 0) | (final_scores > 999.99)]
        if len(invalid_final_scores) > 0:
            df.loc[:, 'final_score_valid'] = ~final_scores.isin(invalid_final_scores)
            logger.warning(f"发现{len(invalid_final_scores)}条最终分数值超出范围(0-999.99):")
            for idx in invalid_final_scores.head(5).index:
                logger.warning(f"  行{idx+2}: 学号={df.loc[idx, '学号']}, 最终分={final_scores.loc[idx]}")

        # 统计验证结果
        valid_count = len(df[df['academic_year_valid'] & df['semester_valid'] & df['monthly_scores_valid'] & df['final_score_valid']])
        logger.info(f"数据验证完成: 总计{len(df)}行，验证通过{valid_count}行")

        # 处理超出范围的数值
        df.loc[:, '最终分'] = final_scores.clip(0, 999.99)  # 将超出范围的值截断到有效范围

        # 清理临时列
        df = df.drop(columns=['academic_year_valid', 'semester_valid', 'monthly_scores_valid', 'final_score_valid'])

        return df

    def check_existing_data(self, student_nos: List[str]) -> set:
        """检查已存在的学号"""
        try:
            with self.connection.cursor() as cursor:
                # 分批查询已存在的学号
                existing_nos = set()
                batch_size = 1000

                for i in range(0, len(student_nos), batch_size):
                    batch = student_nos[i:i+batch_size]
                    placeholders = ','.join(['%s'] * len(batch))

                    sql = f"""
                    SELECT DISTINCT student_no
                    FROM {self.temp_table}
                    WHERE student_no IN ({placeholders})
                    """

                    cursor.execute(sql, batch)
                    results = cursor.fetchall()
                    existing_nos.update([row[0] for row in results])

                logger.info(f"检查到{len(existing_nos)}个学号已存在，将跳过导入")
                return existing_nos

        except Exception as e:
            logger.error(f"检查已存在数据失败: {e}")
            return set()

    def insert_batch(self, batch_data: List[Dict[str, Any]]) -> tuple[int, int]:
        """批量插入数据"""
        if not batch_data:
            return 0, 0

        try:
            with self.connection.cursor() as cursor:
                sql = """
                INSERT INTO {table} (
                    student_no, student_name, gender, ethnicity,
                    academic_year_str, semester_str, monthly_scores,
                    final_score, grade_level, department, major,
                    class_name, grade, import_status
                ) VALUES (
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'pending'
                )
                """.format(table=self.temp_table)

                values = []
                for row in batch_data:
                    # 处理NaN值和数值范围，转换为None或合适的值
                    def safe_value(key, default='', is_numeric=False):
                        val = row.get(key, default)
                        if pd.isna(val):
                            return None
                        if is_numeric and isinstance(val, (int, float)):
                            # 对于数值字段，确保在有效范围内
                            if key == '最终分':
                                return max(0, min(999.99, float(val)))
                        return val

                    values.append((
                        safe_value('学号', ''),
                        safe_value('姓名', ''),
                        safe_value('性别', ''),
                        safe_value('民族', ''),
                        safe_value('学年', ''),
                        safe_value('学期', ''),
                        safe_value('平时分数', ''),
                        safe_value('最终分', is_numeric=True),  # 数值字段，NaN转换为None
                        safe_value('等级', ''),
                        safe_value('院系', ''),
                        safe_value('专业', ''),
                        safe_value('班级', ''),
                        safe_value('年级', '')
                    ))

                cursor.executemany(sql, values)
                self.connection.commit()

                return len(batch_data), 0

        except Exception as e:
            logger.error(f"批量插入失败: {e}")
            self.connection.rollback()
            return 0, len(batch_data)

    def import_data(self, df: pd.DataFrame) -> Dict[str, Any]:
        """导入数据到临时表"""
        logger.info("开始导入数据到临时表")

        # 获取所有学号
        student_nos = df['学号'].tolist()

        # 检查已存在的数据
        existing_nos = self.check_existing_data(student_nos)

        # 过滤掉已存在的数据
        df_to_import = df[~df['学号'].isin(existing_nos)]
        skipped_count = len(df) - len(df_to_import)

        logger.info(f"准备导入{len(df_to_import)}行数据，跳过{skipped_count}行已存在数据")

        if len(df_to_import) == 0:
            logger.info("没有需要导入的新数据")
            return {
                'total': len(df),
                'skipped': skipped_count,
                'imported': 0,
                'failed': 0
            }

        # 批量导入
        total_imported = 0
        total_failed = 0

        for i in range(0, len(df_to_import), self.batch_size):
            batch_df = df_to_import.iloc[i:i+self.batch_size]
            batch_data = batch_df.to_dict('records')

            imported, failed = self.insert_batch(batch_data)
            total_imported += imported
            total_failed += failed

            logger.info(f"处理进度: {min(i+self.batch_size, len(df_to_import))}/{len(df_to_import)} "
                       f"(成功:{total_imported}, 失败:{total_failed})")

        result = {
            'total': len(df),
            'skipped': skipped_count,
            'imported': total_imported,
            'failed': total_failed
        }

        logger.info(f"数据导入完成: {result}")
        return result

    def run(self) -> bool:
        """执行导入流程"""
        start_time = time.time()

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

            # 读取Excel数据
            df = self.read_excel_data()

            # 数据验证和清洗
            df = self.validate_data(df)

            # 导入数据
            result = self.import_data(df)

            # 输出统计信息
            logger.info("="*50)
            logger.info("导入统计结果:")
            logger.info(f"  总记录数: {result['total']}")
            logger.info(f"  跳过已存在: {result['skipped']}")
            logger.info(f"  成功导入: {result['imported']}")
            logger.info(f"  导入失败: {result['failed']}")
            logger.info(f"导入总耗时: {elapsed_time:.1f}秒")
            logger.info("="*50)

            return result['failed'] == 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='学生操行分历史数据 - Excel导入临时表')
    parser.add_argument('--excel', default=CONSTANTS["EXCEL_FILE"],
                       help=f'Excel文件路径 (默认: {CONSTANTS["EXCEL_FILE"]})')

    args = parser.parse_args()

    logger.info("="*60)
    logger.info("学生操行分历史数据迁移 - 脚本1：Excel导入临时表")
    logger.info("="*60)

    # 创建logs目录
    Path("logs").mkdir(exist_ok=True)

    # 执行导入
    importer = ExcelImporter(args.excel)
    success = importer.run()

    if success:
        logger.info("✅ Excel数据导入完成")
        sys.exit(0)
    else:
        logger.error("❌ Excel数据导入失败")
        sys.exit(1)

if __name__ == "__main__":
    main()