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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
学生操行分历史数据迁移 - 主控制脚本

功能说明：
- 协调整个迁移流程的执行
- 按顺序执行所有迁移脚本
- 记录整体迁移过程和结果
- 支持断点续传（可从指定步骤开始）

执行方式：
python import_conduct_score_history.py [--start-from STEP] [--validate-only]

参数说明：
- --start-from: 从指定步骤开始执行 (1-5)
- --validate-only: 仅执行验证，不执行迁移

迁移步骤：
1. Excel导入临时表 (01_import_excel_to_temp.py)
2. 数据解析处理 (02_parse_score_data.py)
3. 数据关联处理 (03_link_student_data.py)
4. 正式导入处理 (04_import_to_production.py)
5. 验证与清理 (05_validate_and_cleanup.py)
"""

from __future__ import annotations

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

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

class MigrationController:
    """迁移控制器"""

    def __init__(self, start_from: int = 1, validate_only: bool = False):
        self.start_from = start_from
        self.validate_only = validate_only
        self.steps = [
            {
                'name': 'Excel导入临时表',
                'script': '01_import_excel_to_temp.py',
                'description': '读取操行.xls文件，导入到temp_conduct_score_import表'
            },
            {
                'name': '数据解析处理',
                'script': '02_parse_score_data.py',
                'description': '解析学年、学期、月度分数，生成月度记录'
            },
            {
                'name': '数据关联处理',
                'script': '03_link_student_data.py',
                'description': '通过学号关联学生、班级、班主任信息'
            },
            {
                'name': '正式导入处理',
                'script': '04_import_to_production.py',
                'description': '将验证后的数据导入到xg_conduct_score表'
            },
            {
                'name': '验证与清理',
                'script': '05_validate_and_cleanup.py',
                'description': '验证迁移结果，生成报告'
            }
        ]

    def check_prerequisites(self) -> bool:
        """检查前置条件"""
        logger.info("检查前置条件...")

        # 检查Python脚本是否存在
        for step in self.steps:
            script_path = step['script']
            if not Path(script_path).exists():
                logger.error(f"脚本文件不存在: {script_path}")
                return False

        # 检查Excel文件是否存在（仅在执行第一步时检查）
        if self.start_from == 1 and not self.validate_only:
            excel_file = "操行.xls"
            if not Path(excel_file).exists():
                logger.error(f"Excel文件不存在: {excel_file}")
                return False

        # 检查logs目录
        Path("logs").mkdir(exist_ok=True)

        logger.info("前置条件检查通过")
        return True

    def run_step(self, step_info: Dict[str, Any]) -> tuple[bool, str]:
        """执行单个步骤"""
        step_name = step_info['name']
        script = step_info['script']
        description = step_info['description']

        logger.info(f"开始执行步骤: {step_name}")
        logger.info(f"脚本: {script}")
        logger.info(f"说明: {description}")

        start_time = time.time()

        try:
            # 构建命令
            cmd = [sys.executable, script]

            # 如果是验证步骤，添加--cleanup参数（由用户决定是否清理）
            # 这里默认不清理，让用户手动决定

            # 执行脚本
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                encoding='utf-8',
                timeout=3600  # 1小时超时
            )

            elapsed_time = time.time() - start_time

            # 记录执行结果
            if result.returncode == 0:
                logger.info(f"步骤耗时: {step_elapsed:.1f}秒")
                return True, result.stdout + result.stderr
            else:
                logger.error(f"步骤耗时: {step_elapsed:.1f}秒")
                logger.error("标准输出:")
                logger.error(result.stdout)
                logger.error("标准错误:")
                logger.error(result.stderr)
                return False, result.stdout + result.stderr

        except subprocess.TimeoutExpired:
            elapsed_time = time.time() - start_time
            logger.error(f"步骤执行超时: {elapsed_time:.1f}秒")
            return False, "执行超时"

        except Exception as e:
            elapsed_time = time.time() - start_time
            logger.error(f"步骤执行出错: {elapsed_time:.1f}秒")
            return False, str(e)

    def run_migration(self) -> bool:
        """执行迁移流程"""
        logger.info("="*80)
        logger.info("学生操行分历史数据迁移开始")
        logger.info("="*80)
        logger.info(f"开始步骤: {self.start_from}")
        logger.info(f"验证模式: {self.validate_only}")
        logger.info(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        logger.info("")

        # 显示执行计划
        logger.info("执行计划:")
        for i, step in enumerate(self.steps, 1):
            status = "⏭️ 跳过" if i < self.start_from else "▶️ 执行"
            if self.validate_only and i < 5:
                status = "⏭️ 跳过"
            logger.info(f"  {i}. {status} {step['name']}")
        logger.info("")

        total_start_time = time.time()
        results = []

        # 执行步骤
        for i, step in enumerate(self.steps, 1):
            step_start_time = time.time()

            # 检查是否需要执行此步骤
            if i < self.start_from:
                logger.info(f"跳过步骤 {i}: {step['name']}")
                continue

            if self.validate_only and i < 5:
                logger.info(f"验证模式跳过步骤 {i}: {step['name']}")
                continue

            # 执行步骤
            success, output = self.run_step(step)

            step_elapsed = time.time() - step_start_time
            results.append({
                'step': i,
                'name': step['name'],
                'success': success,
                'duration': step_elapsed,
                'output': output
            })

            # 如果步骤失败，停止执行
            if not success:
                logger.error(f"步骤 {i} 执行失败，停止迁移流程")
                break

            logger.info("")

        # 输出总结报告
        total_elapsed = time.time() - total_start_time
        self.print_summary_report(results, total_elapsed)

        # 检查整体结果
        all_success = all(result['success'] for result in results)
        return all_success

    def print_summary_report(self, results: List[Dict[str, Any]], total_time: float) -> None:
        """打印总结报告"""
        logger.info("="*80)
        logger.info("迁移执行总结报告")
        logger.info("="*80)

        logger.info("各步骤执行结果:")
        for result in results:
            status = "✅ 成功" if result['success'] else "❌ 失败"
            duration = ".1f"
            logger.info("6")

        logger.info("")
        logger.info(f"迁移总耗时: {total_time:.1f}秒")
        # 统计信息
        successful_steps = sum(1 for r in results if r['success'])
        total_steps = len(results)

        logger.info(f"成功步骤: {successful_steps}/{total_steps}")

        if successful_steps == total_steps:
            logger.info("🎉 迁移流程完全成功！")
        else:
            logger.info("⚠️ 迁移流程部分失败，请检查上述错误信息")

        logger.info("="*80)

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='学生操行分历史数据迁移 - 主控制脚本')
    parser.add_argument('--start-from', type=int, choices=range(1, 6), default=1,
                       help='从指定步骤开始执行 (1-5)')
    parser.add_argument('--validate-only', action='store_true',
                       help='仅执行验证步骤（步骤5），跳过前面的迁移步骤')

    args = parser.parse_args()

    # 如果是验证模式，从步骤5开始
    if args.validate_only:
        args.start_from = 5

    # 创建控制器并执行
    controller = MigrationController(
        start_from=args.start_from,
        validate_only=args.validate_only
    )

    # 检查前置条件
    if not controller.check_prerequisites():
        logger.error("前置条件检查失败，请解决上述问题后重试")
        sys.exit(1)

    # 执行迁移
    success = controller.run_migration()

    if success:
        logger.info("✅ 学生操行分历史数据迁移完成")
        sys.exit(0)
    else:
        logger.error("❌ 学生操行分历史数据迁移失败")
        sys.exit(1)

if __name__ == "__main__":
    main()