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