all-project/用户信息同步统一身份认证/oracle_gen_merge.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""从 MySQL 源库导出数据，生成 Oracle MERGE SQL，供 sqlplus 执行。"""

from __future__ import annotations

import argparse
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable, List, Optional, Sequence, Tuple

try:
    import pymysql
except ImportError:
    print("ERROR: 需要安装 pymysql: pip install pymysql", file=sys.stderr)
    sys.exit(1)


def connect_mysql(host: str, port: int, user: str, password: str, database: str):
    return pymysql.connect(
        host=host,
        port=port,
        user=user,
        password=password,
        database=database,
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
    )


def q(val: Any, max_len: Optional[int] = None) -> str:
    """Python 值转 Oracle SQL 字面量。"""
    if val is None:
        return "NULL"
    if isinstance(val, datetime):
        s = val.strftime("%Y%m%d%H%M%S")
    else:
        s = str(val).strip()
    if s == "":
        return "NULL"
    if max_len is not None:
        s = s[:max_len]
    s = s.replace("'", "''")
    return f"'{s}'"


def right10(val: Any) -> Optional[str]:
    if val is None:
        return None
    s = str(val).strip()
    if not s or s == "0":
        return None
    return s[-10:]


def fmt_dt(val: Any, fmt: str = "%Y%m%d%H%M%S") -> Optional[str]:
    if val is None:
        return None
    if isinstance(val, datetime):
        return val.strftime(fmt)
    s = str(val).strip()
    return s or None


def sex_code(val: Any) -> str:
    s = "" if val is None else str(val)
    if s == "0":
        return "1"
    if s == "1":
        return "2"
    return "0"


def write_header(out) -> None:
    out.write("-- generated by oracle_gen_merge.py\n")
    out.write("WHENEVER SQLERROR EXIT SQL.SQLCODE\n")
    out.write("SET ECHO OFF\n")
    out.write("SET FEEDBACK OFF\n")
    out.write("SET DEFINE OFF\n")
    out.write("SET SERVEROUTPUT ON SIZE UNLIMITED\n")
    out.write("\n")


def commit_batch(out, count: int, label: str) -> None:
    out.write("COMMIT;\n")
    out.write(f"PROMPT {label} committed batch, total so far: {count}\n")
    out.write("\n")


def gen_dept(cur, out, incr_sql: str, batch_size: int) -> int:
    sql = f"""
    SELECT
      d.dept_id,
      d.dept_name,
      d.parent_id,
      d.status,
      d.del_flag,
      IFNULL(d.update_time, d.create_time) AS ts
    FROM sys_dept d
    WHERE 1=1
    {incr_sql}
    """
    cur.execute(sql)
    n = 0
    for row in cur:
        dwh = right10(row["dept_id"])
        if not dwh:
            continue
        lsdwh = right10(row["parent_id"])
        dwyxbs = "1" if row["status"] == "0" and row["del_flag"] == "0" else "0"
        scbz = "y" if row["del_flag"] == "1" else "n"
        zhgxsj = fmt_dt(row["ts"])
        out.write(
            f"""MERGE INTO GXXX_YXSDWJBSJZL t
USING (SELECT {q(dwh, 10)} AS dwh FROM dual) s
ON (t.dwh = s.dwh)
WHEN MATCHED THEN UPDATE SET
  t.dwmc = {q(row['dept_name'], 60)},
  t.lsdwh = {q(lsdwh, 10)},
  t.dwyxbs = {q(dwyxbs, 1)},
  t.zhgxsj = {q(zhgxsj, 20)},
  t.scbz = {q(scbz, 1)}
WHEN NOT MATCHED THEN INSERT (dwh, dwmc, lsdwh, dwyxbs, zhgxsj, scbz)
VALUES ({q(dwh, 10)}, {q(row['dept_name'], 60)}, {q(lsdwh, 10)}, {q(dwyxbs, 1)}, {q(zhgxsj, 20)}, {q(scbz, 1)});
"""
        )
        n += 1
        if n % batch_size == 0:
            commit_batch(out, n, "DEPT")
    if n % batch_size != 0:
        commit_batch(out, n, "DEPT")
    out.write(f"PROMPT DEPT done: {n}\n\n")
    return n


def gen_teacher(cur, out, incr_sql: str, batch_size: int) -> Tuple[int, int]:
    sql = f"""
    SELECT
      u.user_name,
      u.nick_name,
      u.dept_id,
      u.sex,
      u.status,
      u.del_flag,
      u.phonenumber,
      u.email,
      IFNULL(u.update_time, u.create_time) AS ts
    FROM sys_user u
    WHERE u.account_type = '1'
    {incr_sql}
    """
    cur.execute(sql)
    n_base = 0
    n_tx = 0
    for row in cur:
        gh = (row["user_name"] or "").strip()
        if not gh:
            continue
        gh = gh[:20]
        dwh = right10(row["dept_id"]) or "0"
        dwh = dwh[:10]
        xm = (row["nick_name"] or row["user_name"] or "")[:36]
        xbm = sex_code(row["sex"])
        dqztm = "11" if row["status"] == "0" else "99"
        scbz = "y" if row["del_flag"] == "1" else "n"
        zhgxsj = fmt_dt(row["ts"])
        yddh = row["phonenumber"] or None
        dzxx = row["email"] or None

        out.write(
            f"""MERGE INTO GXJG_JZGJCSJZLB t
USING (SELECT {q(gh, 20)} AS gh FROM dual) s
ON (t.gh = s.gh)
WHEN MATCHED THEN UPDATE SET
  t.dwh = {q(dwh, 10)},
  t.xm = {q(xm, 36)},
  t.xbm = {q(xbm, 1)},
  t.dqztm = {q(dqztm, 2)},
  t.zhgxsj = {q(zhgxsj, 20)},
  t.scbz = {q(scbz, 1)}
WHEN NOT MATCHED THEN INSERT (gh, dwh, xm, xbm, dqztm, zhgxsj, scbz)
VALUES ({q(gh, 20)}, {q(dwh, 10)}, {q(xm, 36)}, {q(xbm, 1)}, {q(dqztm, 2)}, {q(zhgxsj, 20)}, {q(scbz, 1)});
"""
        )
        n_base += 1

        tid = gh[:48]
        out.write(
            f"""MERGE INTO GXJG_GRTXFSZL t
USING (SELECT {q(tid, 48)} AS id FROM dual) s
ON (t.id = s.id)
WHEN MATCHED THEN UPDATE SET
  t.gh = {q(gh, 20)},
  t.yddh = {q(yddh, 30)},
  t.dzxx = {q(dzxx, 40)},
  t.zhgxsj = {q(zhgxsj, 20)},
  t.scbz = {q(scbz, 1)}
WHEN NOT MATCHED THEN INSERT (id, gh, yddh, dzxx, zhgxsj, scbz)
VALUES ({q(tid, 48)}, {q(gh, 20)}, {q(yddh, 30)}, {q(dzxx, 40)}, {q(zhgxsj, 20)}, {q(scbz, 1)});
"""
        )
        n_tx += 1

        if n_base % batch_size == 0:
            commit_batch(out, n_base, "TEACHER")

    if n_base % batch_size != 0:
        commit_batch(out, n_base, "TEACHER")
    out.write(f"PROMPT TEACHER done: base={n_base} tx={n_tx}\n\n")
    return n_base, n_tx


def gen_student(cur, out, incr_sql: str, batch_size: int) -> Tuple[int, int]:
    sql = f"""
    SELECT
      s.student_no,
      s.name,
      s.name_english,
      s.former_name,
      s.gender,
      s.birth_date,
      s.native_place,
      s.id_card,
      s.political_status,
      s.blood_type,
      s.is_single_child,
      s.college_id,
      s.class_id,
      s.student_status,
      s.deleted,
      s.postal_code,
      s.current_address,
      s.home_address,
      s.phone,
      s.email,
      s.qq,
      s.wechat,
      c.class_code,
      IFNULL(s.update_time, s.create_time) AS ts
    FROM xg_student_msg s
    LEFT JOIN xg_class_msg c ON c.id = s.class_id AND c.deleted = 0
    WHERE 1=1
    {incr_sql}
    """
    cur.execute(sql)
    n_base = 0
    n_tx = 0
    for row in cur:
        xh = (row["student_no"] or "").strip()
        if not xh:
            continue
        xh = xh[:20]
        dwh = str(row["college_id"] or "0")[:20]
        if not dwh:
            dwh = "0"
        szbh = (row["class_code"] or row["class_id"] or "")
        szbh = str(szbh)[:20] if szbh else None
        status = "" if row["student_status"] is None else str(row["student_status"])
        sfzx = "1" if status in ("1", "2", "3") else "0"
        sfyby = "1" if status == "5" else "0"
        scbz = "y" if row["deleted"] == 1 else "n"
        zhgxsj = fmt_dt(row["ts"])
        csrq = fmt_dt(row["birth_date"], "%Y%m%d")
        sfdszn = "1" if row["is_single_child"] == 1 else "0"
        zzmmm = (row["political_status"] or None)
        if zzmmm is not None:
            zzmmm = str(zzmmm)[:2]
        txdz = row["current_address"] or row["home_address"]
        jstxh = row["qq"] or row["wechat"]

        out.write(
            f"""MERGE INTO GXXS_XSJBSJZL t
USING (SELECT {q(xh, 20)} AS xh FROM dual) s
ON (t.xh = s.xh)
WHEN MATCHED THEN UPDATE SET
  t.xm = {q(row['name'], 60)},
  t.ywxm = {q(row['name_english'], 60)},
  t.cym = {q(row['former_name'], 60)},
  t.xbm = {q(sex_code(row['gender']), 1)},
  t.csrq = {q(csrq, 10)},
  t.jg = {q(row['native_place'], 60)},
  t.sfzjlxm = '1',
  t.sfzjh = {q(row['id_card'], 20)},
  t.zzmmm = {q(zzmmm, 2)},
  t.xxm = {q(row['blood_type'], 30)},
  t.sfdszn = {q(sfdszn, 1)},
  t.zhgxsj = {q(zhgxsj, 20)},
  t.scbz = {q(scbz, 1)},
  t.dwh = {q(dwh, 20)},
  t.szbh = {q(szbh, 20)},
  t.sfzx = {q(sfzx, 10)},
  t.sfyby = {q(sfyby, 6)}
WHEN NOT MATCHED THEN INSERT (
  xh, xm, ywxm, cym, xbm, csrq, jg, sfzjlxm, sfzjh, zzmmm, xxm,
  sfdszn, zhgxsj, scbz, dwh, szbh, sfzx, sfyby
) VALUES (
  {q(xh, 20)}, {q(row['name'], 60)}, {q(row['name_english'], 60)}, {q(row['former_name'], 60)},
  {q(sex_code(row['gender']), 1)}, {q(csrq, 10)}, {q(row['native_place'], 60)}, '1',
  {q(row['id_card'], 20)}, {q(zzmmm, 2)}, {q(row['blood_type'], 30)}, {q(sfdszn, 1)},
  {q(zhgxsj, 20)}, {q(scbz, 1)}, {q(dwh, 20)}, {q(szbh, 20)}, {q(sfzx, 10)}, {q(sfyby, 6)}
);
"""
        )
        n_base += 1

        out.write(
            f"""MERGE INTO GXXS_GRTXZL t
USING (SELECT {q(xh, 20)} AS xh FROM dual) s
ON (t.xh = s.xh)
WHEN MATCHED THEN UPDATE SET
  t.yzbm = {q(row['postal_code'], 6)},
  t.txdz = {q(txdz, 180)},
  t.yddh = {q(row['phone'], 30)},
  t.dzxx = {q(row['email'], 40)},
  t.jstxh = {q(jstxh, 40)},
  t.zhgxsj = {q(zhgxsj, 20)},
  t.scbz = {q(scbz, 1)}
WHEN NOT MATCHED THEN INSERT (xh, yzbm, txdz, yddh, dzxx, jstxh, zhgxsj, scbz)
VALUES (
  {q(xh, 20)}, {q(row['postal_code'], 6)}, {q(txdz, 180)}, {q(row['phone'], 30)},
  {q(row['email'], 40)}, {q(jstxh, 40)}, {q(zhgxsj, 20)}, {q(scbz, 1)}
);
"""
        )
        n_tx += 1

        if n_base % batch_size == 0:
            commit_batch(out, n_base, "STUDENT")

    if n_base % batch_size != 0:
        commit_batch(out, n_base, "STUDENT")
    out.write(f"PROMPT STUDENT done: base={n_base} tx={n_tx}\n\n")
    return n_base, n_tx


def build_incr(alias: str, last_sync: str, full: bool) -> str:
    if full or not last_sync or last_sync.startswith("1970"):
        return ""
    return (
        f"AND ({alias}.update_time > '{last_sync}' "
        f"OR {alias}.create_time > '{last_sync}')"
    )


def main() -> int:
    parser = argparse.ArgumentParser(description="生成 Oracle MERGE SQL")
    parser.add_argument("--host", required=True)
    parser.add_argument("--port", default="3306")
    parser.add_argument("--user", required=True)
    parser.add_argument("--password", required=True)
    parser.add_argument("--database", required=True)
    parser.add_argument("--output", required=True, help="输出 SQL 文件路径")
    parser.add_argument("--last-sync", default="1970-01-01 00:00:00")
    parser.add_argument("--full", action="store_true")
    parser.add_argument("--batch-size", type=int, default=200)
    args = parser.parse_args()

    out_path = Path(args.output)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    conn = connect_mysql(
        args.host, int(args.port), args.user, args.password, args.database
    )
    try:
        with conn.cursor() as cur, out_path.open("w", encoding="utf-8") as out:
            write_header(out)
            dept_n = gen_dept(
                cur, out, build_incr("d", args.last_sync, args.full), args.batch_size
            )
            tea_n, tea_tx = gen_teacher(
                cur, out, build_incr("u", args.last_sync, args.full), args.batch_size
            )
            stu_n, stu_tx = gen_student(
                cur, out, build_incr("s", args.last_sync, args.full), args.batch_size
            )
            out.write("COMMIT;\n")
            out.write(
                f"PROMPT SYNC SUMMARY dept={dept_n} teacher={tea_n}/{tea_tx} "
                f"student={stu_n}/{stu_tx}\n"
            )
            out.write("EXIT;\n")
        print(
            f"OK dept={dept_n} teacher={tea_n} teacher_tx={tea_tx} "
            f"student={stu_n} student_tx={stu_tx} file={out_path}"
        )
        return 0
    finally:
        conn.close()


if __name__ == "__main__":
    sys.exit(main())