all-project/学校历史数据迁移/08-insert_students_from_his.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
从 his_xssj 表把历史学生数据同步到 xg 系统表

主要写入表：
 - xg_student_msg
 - sys_user
 - sys_user_role

特性：
 - 同步前提示是否清空 xg_student_msg（需用户确认）
 - 跳过 班级编号 为 '#YDQX#' 或 空字符串 的记录
 - 批量插入并分批提交，包含简单的重试机制
 - 通过 sys_dept 表查找班级/专业/学院信息；通过 xg_class_msg 获取班级关联字段
 - 字段映射/清洗遵循项目要求（见任务描述）

注意：此脚本依赖同目录下的 `db_config.py` 提供 `DB_CONFIG` 与 `TABLE_MAPPING`
"""
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_MAPPING

# 常量配置（可根据需要改为从外部配置）
STUDENT_ROLE_ID = 1938120227459309569
DEFAULT_TENANT_ID = "000000"
DEFAULT_CREATE_BY = 1
DEFAULT_UPDATE_BY = 1
DEFAULT_PASSWORD_HASH = "$2a$10$Vm51QDgTteXRQKpERT2okumJh1sVm5YpvkiVQdaih/rgaOtHrm6yi"
BATCH_SIZE = 200
DB_RETRY_MAX = 3

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("sync_his_xssj")


def get_db_connection():
    return pymysql.connect(
        **DB_CONFIG,
        cursorclass=DictCursor,
        autocommit=False,
        connect_timeout=30,
        read_timeout=60,
        write_timeout=60,
    )


def parse_enrollment_date(raw_value: Optional[str]) -> Optional[datetime]:
    if not raw_value:
        return None
    text = str(raw_value).strip()
    if not text:
        return None
    # 常见格式 "9/1/2009 0:0:0" 或 "2009-09-01 00:00:00"
    fmts = ["%m/%d/%Y %H:%M:%S", "%m/%d/%Y %H:%M", "%m/%d/%Y", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"]
    for fmt in fmts:
        try:
            return datetime.strptime(text, fmt)
        except Exception:
            continue
    # 作为最后手段尝试截取日期部分
    try:
        return datetime.strptime(text.split()[0], "%m/%d/%Y")
    except Exception:
        logger.warning("无法解析入学日期 -> raw='%s'", raw_value)
        return None


def normalize_phone(raw_value: Optional[str], max_len: int = 11) -> Optional[str]:
    if not raw_value:
        return None
    s = str(raw_value).strip()
    return s[:max_len] if s else None


def map_gender(text: Optional[str]) -> str:
    if not text:
        return "0"
    t = str(text).strip()
    if t == "女":
        return "1"
    return "0"


def map_sys_user_status(his_status: Optional[str]) -> str:
    """sys_user.status: 毕业、结业、分流、停学、退学、休学 -> 1 ; 复学、在读 -> 0"""
    if not his_status:
        return "0"
    s = str(his_status).strip()
    if s in ("毕业", "结业", "分流", "停学", "退学", "休学"):
        return "1"
    return "0"


def map_student_status(his_status: Optional[str]) -> str:
    """xg_student_msg.student_status 映射"""
    if not his_status:
        return "1"
    mapping = {
        "毕业": "5",
        "结业": "7",
        "分流": "9",
        "复学": "1",
        "停学": "8",
        "退学": "4",
        "在读": "1",
        "休学": "2",
    }
    return mapping.get(str(his_status).strip(), "1")


def find_class_dept(connection, class_code: Optional[str], class_name: Optional[str]) -> Tuple[Optional[int], Optional[str], Optional[int], Optional[str], Optional[int], Optional[str]]:
    """
    根据 his_xssj 的班级编号与班级名称，查找 sys_dept 中对应的班级记录，
    并向上查找专业(zy-)与学院(xy-)节点，返回 (class_id,class_name, major_id,major_name, college_id,college_name)
    """
    if not class_code or not class_name:
        return None, None, None, None, None, None

    dept_category = f"bj-{class_code}"
    with connection.cursor() as cursor:
        cursor.execute(
            "SELECT dept_id, dept_name, parent_id, dept_category FROM `sys_dept` WHERE dept_category = %s AND dept_name = %s LIMIT 1",
            (dept_category, class_name),
        )
        row = cursor.fetchone()
        if not row:
            logger.warning("未在 sys_dept 中找到班级 -> dept_category=%s, dept_name=%s", dept_category, class_name)
            return None, None, None, None, None, None

        class_id = row["dept_id"]
        class_name_ret = row["dept_name"]
        parent_id = row.get("parent_id")

        major_id = None
        major_name = None
        college_id = None
        college_name = None

        # 向上遍历父级查找 zy- 与 xy- 前缀
        current_parent = parent_id
        while current_parent:
            cursor.execute("SELECT dept_id, dept_name, parent_id, dept_category FROM `sys_dept` WHERE dept_id = %s LIMIT 1", (current_parent,))
            parent_row = cursor.fetchone()
            if not parent_row:
                break
            cat = parent_row.get("dept_category") or ""
            if cat.startswith("zy-") and major_id is None:
                major_id = parent_row["dept_id"]
                major_name = parent_row["dept_name"]
            if cat.startswith("xy-") and college_id is None:
                college_id = parent_row["dept_id"]
                college_name = parent_row["dept_name"]
            current_parent = parent_row.get("parent_id")

        return class_id, class_name_ret, major_id, major_name, college_id, college_name


def load_class_msg(connection, class_id: Optional[int]) -> Dict[str, Any]:
    """
    从 xg_class_msg 表读取与班级相关的字段，用于补齐 xg_student_msg 的若干字段
    """
    if not class_id:
        return {}
    with connection.cursor() as cursor:
        cursor.execute("SELECT * FROM `xg_class_msg` WHERE id = %s LIMIT 1", (class_id,))
        return cursor.fetchone() or {}


def sync_from_his_xssj(truncate_before_sync: bool = False):
    table_his = "his_xssj"
    table_xg_student = TABLE_MAPPING.get("xg_student_msg", "xg_student_msg")
    table_sys_user = "sys_user"
    table_sys_user_role = "sys_user_role"

    connection = get_db_connection()
    try:
        with connection.cursor() as cursor:
            # 1) 是否清空目标表
            if truncate_before_sync:
                logger.info("正在清空表 %s ...", table_xg_student)
                cursor.execute(f"TRUNCATE TABLE `{table_xg_student}`")
                connection.commit()
                logger.info("已清空 %s", table_xg_student)

            # 2) 读取需同步的数据（跳过 班级编号 = '#YDQX#' 或 ''）
            select_sql = f"SELECT * FROM `{table_his}` WHERE `班级编号` IS NOT NULL AND `班级编号` NOT IN (%s, %s)"
            cursor.execute(select_sql, ("#YDQX#", ""))
            rows = cursor.fetchall()
            logger.info("从 %s 读取到 %d 条候选记录（已过滤 '#YDQX#' 与空班级编号）", table_his, len(rows))

            if not rows:
                logger.info("没有符合条件的历史学生数据，退出。")
                return

            # 预加载已存在的学号集合与 sys_user user_id 以便去重
            cursor.execute(f"SELECT student_no FROM `{table_xg_student}`")
            existing_student_nos = {r["student_no"] for r in cursor.fetchall()}
            cursor.execute("SELECT user_id FROM `sys_user`")
            existing_user_ids = {int(r["user_id"]) for r in cursor.fetchall()}

            base_datetime = datetime.now()
            timestamp_str = base_datetime.strftime("%Y%m%d%H%M%S")
            sequence = 1

            inserted_student_count = 0
            inserted_user_count = 0
            inserted_role_count = 0
            batch_counter = 0

            # Prepare SQL statements (只插入必要字段)
            insert_xg_student_sql = f"""
            INSERT INTO `{table_xg_student}` (
                id, tenant_id, create_dept, create_by, create_time,
                update_by, update_time, deleted, student_no, id_card,
                name, gender, birth_date, political_status, phone,
                is_boarder, home_address, guardian_phone, admission_category,
                student_status, enrollment_year, enrollment_date,
                school_system, academic_level, training_type,
                major_id, major_name, class_id, class_name, grade,
                college_id, college_name, counselor_id, counselor_name,
                class_teacher_id, class_teacher_name
            ) VALUES (
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s,
                %s, %s, %s,
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s,
                %s, %s, %s, %s
            )
            """

            insert_sys_user_sql = f"""
            INSERT INTO `{table_sys_user}` (
                user_id, dept_id, user_name, nick_name, user_type,
                email, phonenumber, sex, avatar, password, status,
                del_flag, login_ip, login_date, create_by, create_time,
                update_by, update_time, remark, tenant_id, account_type
            ) VALUES (
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s
            )
            """

            insert_user_role_sql = f"""
            INSERT INTO `{table_sys_user_role}` (`user_id`, `role_id`) VALUES (%s, %s)
            """

            for row in rows:
                try:
                    student_no = row.get("学号") or row.get("学生学号") or row.get("xh") or row.get("xs_xh") or row.get("xs_xh_x") or row.get("学号".strip())
                    # 如果无法从通用列名获取学号，尝试直接使用字段名 '学号'（上面多重尝试尽量兼容）
                    if not student_no:
                        logger.warning("跳过无学号记录: %s", row)
                        continue

                    student_no = str(student_no).strip()
                    if student_no in existing_student_nos:
                        logger.info("跳过已存在学号: %s", student_no)
                        continue

                    name = row.get("姓名") or row.get("xm") or row.get("学生姓名")
                    if not name:
                        logger.warning("跳过无姓名记录: %s", row)
                        continue

                    # 跳过指定班级编号的记录（再次防护）
                    class_code = row.get("班级编号")
                    if not class_code or str(class_code).strip() in ("", "#YDQX#"):
                        continue

                    class_name = row.get("班级名称") or row.get("班级")
                    class_id, class_name_val, major_id, major_name, college_id, college_name = find_class_dept(connection, str(class_code).strip(), class_name)

                    # 从 xg_class_msg 读取额外字段
                    class_msg = load_class_msg(connection, class_id)

                    # 处理字段映射
                    id_card = row.get("身份证号") or row.get("sfzjh")
                    gender = map_gender(row.get("性别") or row.get("xs_xb"))
                    birth_date = None
                    try:
                        raw_birth = row.get("出生日期") or row.get("csrq")
                        if raw_birth:
                            birth_date = datetime.strptime(str(raw_birth).split()[0], "%Y-%m-%d").date()
                    except Exception:
                        birth_date = None

                    political = row.get("政治面貌")
                    political_status = None
                    if political:
                        # 用户提供的映射示例: 群众 - 13 、 预备党员 - 02 、 团员 - 03
                        mapping_pol = {"群众": "13", "预备党员": "02", "团员": "03"}
                        political_status = mapping_pol.get(str(political).strip())

                    phone = normalize_phone(row.get("联系方式") or row.get("联系电话") or row.get("lxdh"))
                    guardian_phone = normalize_phone(row.get("家长电话") or row.get("guardian_phone"), max_len=20)
                    is_boarder = 1 if str(row.get("是否住校") or "").strip() == "是" else 0
                    home_address = row.get("家庭住址") or row.get("jtdz")
                    admission_category = "统招"
                    student_status_mapped = map_student_status(row.get("学生状态") or row.get("xsdqzt") or row.get("状态"))

                    enrollment_year = None
                    try:
                        enr = row.get("入学年份") or row.get("rxnj")
                        if enr:
                            enrollment_year = int(str(enr).strip())
                    except Exception:
                        enrollment_year = None

                    enrollment_date_dt = None
                    try:
                        enr_date_raw = row.get("入学日期") or row.get("rxrq")
                        if enr_date_raw:
                            parsed = parse_enrollment_date(str(enr_date_raw))
                            if parsed:
                                enrollment_date_dt = parsed.date()
                    except Exception:
                        enrollment_date_dt = None

                    # 生成唯一 id (与 sys_user.user_id 一致)
                    if sequence > 99999:
                        logger.error("单次同步ID序列超过99999，停止以避免ID溢出")
                        break
                    unique_id = int(f"{timestamp_str}{sequence:05d}")
                    sequence += 1

                    # 插入 xg_student_msg
                    params_xg_student = (
                        unique_id,
                        DEFAULT_TENANT_ID,
                        class_id or 0,
                        DEFAULT_CREATE_BY,
                        base_datetime,
                        DEFAULT_UPDATE_BY,
                        base_datetime,
                        0,
                        student_no,
                        id_card,
                        name,
                        gender,
                        birth_date,
                        political_status,
                        phone,
                        is_boarder,
                        home_address,
                        guardian_phone,
                        admission_category,
                        student_status_mapped,
                        enrollment_year,
                        enrollment_date_dt,
                        class_msg.get("school_system"),
                        class_msg.get("class_type"),
                        class_msg.get("training_type"),
                        major_id,
                        major_name,
                        class_id,
                        class_name_val,
                        class_msg.get("grade"),
                        college_id,
                        college_name,
                        class_msg.get("counselor_id"),
                        class_msg.get("counselor_name"),
                        class_msg.get("class_teacher_id"),
                        class_msg.get("class_teacher_name"),
                    )

                    # 插入 sys_user (如果不存在)
                    params_sys_user = (
                        unique_id,
                        class_id,
                        student_no,
                        name,
                        "sys_user",
                        None,
                        phone,
                        gender,
                        None,
                        DEFAULT_PASSWORD_HASH,
                        map_sys_user_status(row.get("学生状态") or row.get("xsdqzt")),
                        "0",
                        None,
                        None,
                        DEFAULT_CREATE_BY,
                        base_datetime,
                        DEFAULT_UPDATE_BY,
                        base_datetime,
                        "历史数据导入",
                        DEFAULT_TENANT_ID,
                        "2",
                    )

                    # 执行插入（带重试）
                    attempts = 0
                    while attempts < DB_RETRY_MAX:
                        try:
                            with connection.cursor() as c:
                                c.execute(insert_xg_student_sql, params_xg_student)
                                # sys_user 可能已经存在（通过 user_id 判断）
                                if unique_id not in existing_user_ids:
                                    c.execute(insert_sys_user_sql, params_sys_user)
                                    existing_user_ids.add(unique_id)
                                    inserted_user_count += 1
                                # sys_user_role
                                c.execute("SELECT 1 FROM `sys_user_role` WHERE user_id = %s AND role_id = %s LIMIT 1", (unique_id, STUDENT_ROLE_ID))
                                if not c.fetchone():
                                    c.execute(insert_user_role_sql, (unique_id, STUDENT_ROLE_ID))
                                    inserted_role_count += 1

                                inserted_student_count += 1
                                existing_student_nos.add(student_no)
                                batch_counter += 1

                                if batch_counter >= BATCH_SIZE:
                                    connection.commit()
                                    logger.info("已批量提交 %d 条记录", batch_counter)
                                    batch_counter = 0
                            break
                        except pymysql.err.OperationalError as e:
                            attempts += 1
                            logger.warning("数据库操作失败（尝试 %d/%d）: %s", attempts, DB_RETRY_MAX, e)
                            time.sleep(attempts * 2)
                            # 重新获取连接
                            try:
                                connection.ping(reconnect=True)
                            except Exception:
                                connection = get_db_connection()
                        except Exception as e:
                            logger.exception("插入记录失败: %s; 数据: %s", e, params_xg_student)
                            # 不重试一般数据错误，跳过此记录
                            break
                except Exception as outer_exc:
                    logger.exception("处理单条 his_xssj 记录出错: %s", outer_exc)

            # 提交剩余
            if batch_counter > 0:
                connection.commit()
                logger.info("已提交剩余 %d 条记录", batch_counter)

            logger.info("同步完成：插入学生 %d 条，新增 sys_user %d 条，新增角色绑定 %d 条", inserted_student_count, inserted_user_count, inserted_role_count)
    finally:
        try:
            connection.close()
        except Exception:
            pass


def prompt_confirm_truncate() -> bool:
    answer = input("是否先清空 `xg_student_msg` 表？输入 y 确认，其他任意键取消: ").strip().lower()
    return answer == "y"


def prompt_confirm_delete_students() -> bool:
    answer = input("是否删除 `sys_user` 中 account_type=2 的学生用户及其 sys_user_role 关联？输入 y 确认，其他任意键取消: ").strip().lower()
    return answer == "y"


def split_batches(values: List[int], size: int = 1000) -> List[List[int]]:
    if not values:
        return []
    return [values[i : i + size] for i in range(0, len(values), size)]


def delete_student_accounts() -> Tuple[int, int]:
    """
    删除 sys_user 中 account_type='2' 的用户以及其在 sys_user_role 中的关联记录。
    返回 (deleted_users_count, deleted_roles_count)
    """
    deleted_users = 0
    deleted_roles = 0
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT user_id FROM `sys_user` WHERE account_type = %s", ("2",))
            rows = cursor.fetchall()
            user_ids = [int(r["user_id"]) for r in rows]
            if not user_ids:
                logger.info("sys_user 中无 account_type=2 的用户，无需删除。")
                return 0, 0

            logger.info("找到 %d 个 student sys_user 将被删除（包括其 sys_user_role 关联）", len(user_ids))

            for batch in split_batches(user_ids, size=500):
                placeholders = ",".join(["%s"] * len(batch))
                # 删除角色关联
                sql_roles = f"DELETE FROM `sys_user_role` WHERE user_id IN ({placeholders})"
                cursor.execute(sql_roles, batch)
                deleted_roles += cursor.rowcount or 0
                # 删除用户
                sql_users = f"DELETE FROM `sys_user` WHERE user_id IN ({placeholders})"
                cursor.execute(sql_users, batch)
                deleted_users += cursor.rowcount or 0
            conn.commit()
            logger.info("已删除 %d users, %d role rows", deleted_users, deleted_roles)
            return deleted_users, deleted_roles
    finally:
        try:
            conn.close()
        except Exception:
            pass


def main():
    parser = argparse.ArgumentParser(description="从 his_xssj 同步历史学生到 xg 系统")
    parser.add_argument("--yes-delete-students", action="store_true", help="跳过提示，直接删除 sys_user 中 account_type=2 的学生及其 sys_user_role 关联")
    parser.add_argument("--yes-truncate", action="store_true", help="跳过提示，直接清空 xg_student_msg 表后同步")
    args = parser.parse_args()

    # 询问是否删除 sys_user 中 account_type=2 的学生用户（以及其 sys_user_role）
    if getattr(args, "yes_delete_students", False):
        do_delete_students = True
    else:
        do_delete_students = prompt_confirm_delete_students()

    if do_delete_students:
        deleted_users, deleted_roles = delete_student_accounts()
        logger.info("删除操作完成：删除用户 %d，删除角色关联 %d", deleted_users, deleted_roles)

    # 询问是否清空表
    if args.yes_truncate:
        do_truncate = True
    else:
        do_truncate = prompt_confirm_truncate()

    sync_from_his_xssj(truncate_before_sync=do_truncate)


if __name__ == "__main__":
    main()