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