safety-eval-service/docs/db-diff/compare_schema.py

# -*- coding: utf-8 -*-
"""
只读对比：本地库 vs 线上库 结构差异，并生成升级 SQL。
本脚本仅对 information_schema 执行 SELECT 查询，绝不修改线上数据库。
"""
import json
import pymysql

SCHEMA = "safety-eval-service"

LOCAL = dict(host="192.168.20.100", port=33080,
             user="root", password="Mysql@zcloud33080")
ONLINE = dict(host="nlb-kd2xz70qhllfet2koj.cn-beijing.nlb.aliyuncsslb.com",
              port=33068, user="root", password="5tS3owZ7w8Uk1egv")

COL_SQL = """
SELECT TABLE_NAME, COLUMN_NAME, ORDINAL_POSITION, COLUMN_TYPE, IS_NULLABLE,
       COLUMN_DEFAULT, EXTRA, COLUMN_COMMENT, CHARACTER_SET_NAME, COLLATION_NAME
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA=%s
ORDER BY TABLE_NAME, ORDINAL_POSITION
"""

TBL_SQL = """
SELECT TABLE_NAME, ENGINE, TABLE_COLLATION, TABLE_COMMENT
FROM information_schema.TABLES
WHERE TABLE_SCHEMA=%s AND TABLE_TYPE='BASE TABLE'
ORDER BY TABLE_NAME
"""

IDX_SQL = """
SELECT TABLE_NAME, INDEX_NAME, NON_UNIQUE, SEQ_IN_INDEX, COLUMN_NAME, INDEX_TYPE
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA=%s
ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX
"""


def fetch(conn_args):
    conn = pymysql.connect(charset="utf8mb4", **conn_args)
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute(TBL_SQL, (SCHEMA,))
            tables = {r["TABLE_NAME"]: r for r in cur.fetchall()}
            cur.execute(COL_SQL, (SCHEMA,))
            cols = {}
            for r in cur.fetchall():
                cols.setdefault(r["TABLE_NAME"], {})[r["COLUMN_NAME"]] = r
            cur.execute(IDX_SQL, (SCHEMA,))
            idx = {}
            for r in cur.fetchall():
                idx.setdefault(r["TABLE_NAME"], {}).setdefault(
                    r["INDEX_NAME"], []).append(r)
    finally:
        conn.close()
    return tables, cols, idx


def q(name):
    return "`%s`" % name


def col_def(c):
    """根据本地列定义拼出 DDL 片段。"""
    parts = [q(c["COLUMN_NAME"]), c["COLUMN_TYPE"]]
    if c["CHARACTER_SET_NAME"] and c["COLLATION_NAME"]:
        # 保持与表默认字符集一致，通常无需显式声明，这里省略以免冲突
        pass
    parts.append("NOT NULL" if c["IS_NULLABLE"] == "NO" else "NULL")
    default = c["COLUMN_DEFAULT"]
    extra = (c["EXTRA"] or "")
    if default is not None:
        up = str(default).upper()
        if up in ("CURRENT_TIMESTAMP", "CURRENT_TIMESTAMP()") or \
           up.startswith("CURRENT_TIMESTAMP(") or extra.upper().startswith("DEFAULT_GENERATED"):
            parts.append("DEFAULT %s" % default)
        else:
            parts.append("DEFAULT '%s'" % str(default).replace("'", "''"))
    elif c["IS_NULLABLE"] == "YES":
        parts.append("DEFAULT NULL")
    extra_clean = extra.replace("DEFAULT_GENERATED", "").strip()
    if extra_clean:
        parts.append(extra_clean.upper())
    if c["COLUMN_COMMENT"]:
        parts.append("COMMENT '%s'" % c["COLUMN_COMMENT"].replace("'", "''"))
    return " ".join(parts)


def main():
    l_tbl, l_col, l_idx = fetch(LOCAL)
    o_tbl, o_col, o_idx = fetch(ONLINE)

    report = {
        "local_table_count": len(l_tbl),
        "online_table_count": len(o_tbl),
        "missing_tables": sorted(set(l_tbl) - set(o_tbl)),
        "extra_tables": sorted(set(o_tbl) - set(l_tbl)),
        "missing_columns": [],
        "extra_columns": [],
        "diff_columns": [],
        "missing_indexes": [],
    }

    for t in sorted(set(l_tbl) & set(o_tbl)):
        lc, oc = l_col.get(t, {}), o_col.get(t, {})
        for name in lc:
            if name not in oc:
                report["missing_columns"].append(
                    {"table": t, "column": name, "def": col_def(lc[name]),
                     "pos": lc[name]["ORDINAL_POSITION"]})
        for name in oc:
            if name not in lc:
                report["extra_columns"].append({"table": t, "column": name,
                                                "type": oc[name]["COLUMN_TYPE"]})
        for name in lc:
            if name not in oc:
                continue
            a, b = lc[name], oc[name]
            diffs = {}
            if a["COLUMN_TYPE"] != b["COLUMN_TYPE"]:
                diffs["type"] = [b["COLUMN_TYPE"], a["COLUMN_TYPE"]]
            if a["IS_NULLABLE"] != b["IS_NULLABLE"]:
                diffs["nullable"] = [b["IS_NULLABLE"], a["IS_NULLABLE"]]
            if (a["COLUMN_DEFAULT"] or "") != (b["COLUMN_DEFAULT"] or ""):
                diffs["default"] = [b["COLUMN_DEFAULT"], a["COLUMN_DEFAULT"]]
            if (a["EXTRA"] or "") != (b["EXTRA"] or ""):
                diffs["extra"] = [b["EXTRA"], a["EXTRA"]]
            if (a["COLUMN_COMMENT"] or "") != (b["COLUMN_COMMENT"] or ""):
                diffs["comment"] = [b["COLUMN_COMMENT"], a["COLUMN_COMMENT"]]
            if diffs:
                report["diff_columns"].append(
                    {"table": t, "column": name, "diffs": diffs,
                     "def": col_def(a)})

        li, oi = l_idx.get(t, {}), o_idx.get(t, {})
        for iname, rows in li.items():
            if iname == "PRIMARY":
                continue
            if iname not in oi:
                report["missing_indexes"].append({
                    "table": t, "index": iname,
                    "unique": rows[0]["NON_UNIQUE"] == 0,
                    "columns": [r["COLUMN_NAME"] for r in rows]})

    with open("schema_diff.json", "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2, default=str)

    print("本地表数:", report["local_table_count"],
          " 线上表数:", report["online_table_count"])
    print("线上缺表:", len(report["missing_tables"]), report["missing_tables"])
    print("线上多表:", len(report["extra_tables"]), report["extra_tables"])
    print("线上缺字段:", len(report["missing_columns"]))
    print("线上多字段:", len(report["extra_columns"]))
    print("字段定义不一致:", len(report["diff_columns"]))
    print("线上缺索引:", len(report["missing_indexes"]))


if __name__ == "__main__":
    main()