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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
学生操行分历史数据迁移 - 脚本3：数据关联处理

功能说明：
- 根据学号关联xg_student_msg表，获取学生ID
- 根据班级ID获取班主任信息
- 更新temp_conduct_score_parsed表中的关联字段

执行方式：
python 03_link_student_data.py

注意事项：
- 依赖脚本2已执行完成
- 学号不存在的学生记录会标记为错误状态
"""

from __future__ import annotations

import argparse
import logging
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

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler("logs/03_link_data.log", mode='a', encoding='utf-8')
    ]
)
logger = logging.getLogger("link_data")

class DataLinker:
    """数据关联处理器"""

    def __init__(self):
        self.db_config = DB_CONFIG
        self.temp_parsed_table = TABLE_CONFIG["temp_parsed"]
        self.students_table = TABLE_CONFIG["students"]
        self.classes_table = TABLE_CONFIG["classes"]
        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 get_pending_data(self) -> List[Dict[str, Any]]:
        """获取待关联的数据"""
        try:
            with self.connection.cursor(DictCursor) as cursor:
                sql = f"""
                SELECT id, temp_id, student_no, academic_year, semester, month, score
                FROM {self.temp_parsed_table}
                WHERE processing_status = 'parsed'
                ORDER BY id
                LIMIT {self.batch_size}
                """

                cursor.execute(sql)
                return cursor.fetchall()

        except Exception as e:
            logger.error(f"获取待关联数据失败: {e}")
            return []

    def link_student_info(self, student_nos: List[str]) -> Dict[str, Dict[str, Any]]:
        """批量关联学生信息"""
        if not student_nos:
            return {}

        try:
            with self.connection.cursor(DictCursor) as cursor:
                # 构建IN查询
                placeholders = ','.join(['%s'] * len(student_nos))
                sql = f"""
                SELECT
                    s.id as student_id,
                    s.student_no,
                    s.class_id,
                    c.class_teacher_id
                FROM {self.students_table} s
                LEFT JOIN {self.classes_table} c ON s.class_id = c.id
                WHERE s.student_no IN ({placeholders})
                    AND s.deleted = 0
                    AND (c.id IS NULL OR c.deleted = 0)
                """

                cursor.execute(sql, student_nos)
                results = cursor.fetchall()

                # 转换为字典格式
                student_info = {}
                for row in results:
                    student_info[row['student_no']] = {
                        'student_id': row['student_id'],
                        'class_id': row['class_id'],
                        'class_teacher_id': row['class_teacher_id']
                    }

                return student_info

        except Exception as e:
            logger.error(f"关联学生信息失败: {e}")
            return {}

    def update_parsed_data(self, update_data: List[Dict[str, Any]]) -> tuple[int, int]:
        """更新解析表中的关联信息"""
        if not update_data:
            return 0, 0

        try:
            with self.connection.cursor() as cursor:
                # 分别处理成功和失败的数据
                success_updates = []
                error_updates = []

                for item in update_data:
                    if item.get('student_id'):
                        success_updates.append(item)
                    else:
                        error_updates.append(item)

                # 更新成功的记录
                success_count = 0
                if success_updates:
                    sql = f"""
                    UPDATE {self.temp_parsed_table}
                    SET student_id = %s, class_id = %s, class_teacher_id = %s,
                        processing_status = 'linked', update_time = NOW()
                    WHERE id = %s
                    """

                    values = [(item['student_id'], item['class_id'], item['class_teacher_id'], item['id'])
                             for item in success_updates]

                    cursor.executemany(sql, values)
                    success_count = len(success_updates)

                # 更新失败的记录
                error_count = 0
                if error_updates:
                    sql = f"""
                    UPDATE {self.temp_parsed_table}
                    SET processing_status = 'error', error_msg = %s, update_time = NOW()
                    WHERE id = %s
                    """

                    values = [("学号不存在或学生已被删除", item['id']) for item in error_updates]
                    cursor.executemany(sql, values)
                    error_count = len(error_updates)

                self.connection.commit()

                return success_count, error_count

        except Exception as e:
            logger.error(f"更新关联数据失败: {e}")
            self.connection.rollback()
            return 0, len(update_data)

    def process_batch(self, batch_data: List[Dict[str, Any]]) -> Dict[str, Any]:
        """处理一批数据"""
        if not batch_data:
            return {'processed': 0, 'linked': 0, 'not_found': 0}

        # 提取学号列表
        student_nos = list(set(row['student_no'] for row in batch_data))

        # 关联学生信息
        student_info = self.link_student_info(student_nos)

        # 构建更新数据
        update_data = []
        not_found_nos = set()

        for row in batch_data:
            student_no = row['student_no']
            info = student_info.get(student_no)

            if info:
                update_data.append({
                    'id': row['id'],
                    'student_id': info['student_id'],
                    'class_id': info['class_id'],
                    'class_teacher_id': info['class_teacher_id']
                })
            else:
                update_data.append({
                    'id': row['id'],
                    'student_id': None,
                    'class_id': None,
                    'class_teacher_id': None
                })
                not_found_nos.add(student_no)

        # 输出未找到的学号
        if not_found_nos:
            logger.warning(f"未找到以下学号的学生信息 ({len(not_found_nos)}个):")
            for student_no in sorted(list(not_found_nos)[:10]):  # 只显示前10个
                logger.warning(f"  {student_no}")
            if len(not_found_nos) > 10:
                logger.warning(f"  ... 还有{len(not_found_nos) - 10}个")

        # 更新数据库
        linked, not_linked = self.update_parsed_data(update_data)

        return {
            'processed': len(batch_data),
            'linked': linked,
            'not_found': not_linked,
            'unique_students': len(student_nos),
            'found_students': len(student_info)
        }

    def get_statistics(self) -> Dict[str, Any]:
        """获取处理统计"""
        try:
            with self.connection.cursor(DictCursor) as cursor:
                sql = f"""
                SELECT
                    COUNT(*) as total,
                    SUM(CASE WHEN processing_status = 'parsed' THEN 1 ELSE 0 END) as pending,
                    SUM(CASE WHEN processing_status = 'linked' THEN 1 ELSE 0 END) as linked,
                    SUM(CASE WHEN processing_status = 'error' THEN 1 ELSE 0 END) as error
                FROM {self.temp_parsed_table}
                """

                cursor.execute(sql)
                result = cursor.fetchone()

                return {
                    'total_records': result['total'] or 0,
                    'pending': result['pending'] or 0,
                    'linked': result['linked'] or 0,
                    'error': result['error'] or 0
                }

        except Exception as e:
            logger.error(f"获取统计信息失败: {e}")
            return {'total_records': 0, 'pending': 0, 'linked': 0, 'error': 0}

    def link_student_data_efficient(self) -> Tuple[int, int]:
        """高效的关联学生数据 - 使用单次UPDATE JOIN操作"""
        try:
            with self.connection.cursor() as cursor:
                # 使用UPDATE JOIN一次性关联所有数据
                sql = f"""
                UPDATE {self.temp_parsed_table} t
                LEFT JOIN {self.students_table} s ON t.student_no = s.student_no
                    AND s.deleted = 0
                LEFT JOIN {self.classes_table} c ON s.class_id = c.id
                    AND c.deleted = 0
                SET
                    t.student_id = s.id,
                    t.class_id = s.class_id,
                    t.class_teacher_id = c.class_teacher_id,
                    t.processing_status = CASE
                        WHEN s.id IS NOT NULL THEN 'linked'
                        ELSE 'error'
                    END,
                    t.error_msg = CASE
                        WHEN s.id IS NULL THEN '学号不存在或学生已被删除'
                        ELSE NULL
                    END,
                    t.update_time = NOW()
                WHERE t.processing_status = 'parsed'
                """

                affected_rows = cursor.execute(sql)
                self.connection.commit()

                # 统计成功和失败的数量
                success_count = 0
                error_count = 0

                if affected_rows > 0:
                    # 获取更新后的统计
                    with self.connection.cursor() as count_cursor:
                        count_sql = f"""
                        SELECT
                            SUM(CASE WHEN processing_status = 'linked' THEN 1 ELSE 0 END) as linked,
                            SUM(CASE WHEN processing_status = 'error' THEN 1 ELSE 0 END) as error
                        FROM {self.temp_parsed_table}
                        WHERE processing_status IN ('linked', 'error')
                        """
                        count_cursor.execute(count_sql)
                        result = count_cursor.fetchone()
                        success_count = result[0] or 0
                        error_count = result[1] or 0

                logger.info(f"UPDATE JOIN操作完成，影响行数: {affected_rows}")
                return success_count, error_count

        except Exception as e:
            logger.error(f"高效关联操作失败: {e}")
            self.connection.rollback()
            return 0, 0

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

        try:
            self.connect_db()

            # 显示初始统计
            initial_stats = self.get_statistics()
            logger.info("数据关联处理开始")
            logger.info(f"当前状态: 总计{initial_stats['total_records']}条记录，"
                       f"待关联{initial_stats['pending']}条，已关联{initial_stats['linked']}条，"
                       f"错误{initial_stats['error']}条")

            # 使用高效的UPDATE JOIN方式处理所有数据
            logger.info("开始执行高效关联操作...")
            linked, error = self.link_student_data_efficient()

            # 显示最终统计
            final_stats = self.get_statistics()
            logger.info("="*50)
            logger.info("数据关联统计结果:")
            logger.info(f"  关联成功: {linked}")
            logger.info(f"  学号未找到: {error}")
            logger.info(f"  最终状态:")
            logger.info(f"    - 已关联: {final_stats['linked']}")
            logger.info(f"    - 关联失败: {final_stats['error']}")
            logger.info("="*50)

            # 如果有未找到的学号，输出详细信息
            if error > 0:
                logger.warning(f"共有{error}条记录的学号未找到对应学生信息")
                logger.warning("这些记录将无法导入到正式表，请检查学生数据")

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

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

    # 执行关联
    linker = DataLinker()
    success = linker.run()

    if success:
        logger.info("✅ 数据关联处理完成")
        sys.exit(0)
    else:
        logger.warning("⚠️ 数据关联处理完成，但存在未关联的记录")
        sys.exit(0)  # 不算错误，只是警告

if __name__ == "__main__":
    main()