441 lines
15 KiB
Python
441 lines
15 KiB
Python
#!/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()
|