AiDOPWarehouse/doc/db/mdp/contract_tests/verify_contract.py

"""契约一致校验（DB vs API 等价）。

对同一 mdp_std_* 表，比较 DB 路线批次（sync_batch_id LIKE 前缀 CT_DB_%）与
API 路线批次（CT_API_%）在 source_biz_key 上的集合与行数是否一致。

用法：
    python verify_contract.py --table mdp_std_purchase_receipt
    # 自定义批次前缀：
    python verify_contract.py --table mdp_std_so --db-prefix CT_DB_ --api-prefix CT_API_

退出码：0=一致；1=不一致；2=两侧任一为空（未跑入站）。

前置：DB 路线与 API 路线两次入站都已完成，且分别以 CT_DB_/CT_API_ 前缀写入
mdp_std_*.sync_batch_id（见任务书 §附录 E.2 / §附录 I 回归步骤）。
"""
import argparse
import sys

from _db import query


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--table", required=True, help="目标 mdp_std_* 表名")
    ap.add_argument("--db-prefix", default="CT_DB_")
    ap.add_argument("--api-prefix", default="CT_API_")
    ap.add_argument("--key", default="source_biz_key", help="业务键列，默认 source_biz_key")
    args = ap.parse_args()

    t, k = args.table, args.key
    counts = query(
        f"""
        SELECT 'db' AS src, COUNT(*) AS row_cnt, COUNT(DISTINCT {k}) AS key_cnt
          FROM {t} WHERE sync_batch_id LIKE %s
        UNION ALL
        SELECT 'api', COUNT(*), COUNT(DISTINCT {k})
          FROM {t} WHERE sync_batch_id LIKE %s
        """,
        (args.db_prefix + "%", args.api_prefix + "%"),
    )
    by = {r["src"]: r for r in counts}
    db, api = by.get("db"), by.get("api")
    print(f"[counts] db={db}  api={api}")
    if not db or not api or db["row_cnt"] == 0 or api["row_cnt"] == 0:
        print("[FAIL] 某一侧为空：请先分别跑 DB 与 API 入站。", file=sys.stderr)
        return 2

    only_db = query(
        f"""
        SELECT {k} AS biz FROM {t} WHERE sync_batch_id LIKE %s
          AND {k} NOT IN (SELECT {k} FROM {t} WHERE sync_batch_id LIKE %s)
        """,
        (args.db_prefix + "%", args.api_prefix + "%"),
    )
    only_api = query(
        f"""
        SELECT {k} AS biz FROM {t} WHERE sync_batch_id LIKE %s
          AND {k} NOT IN (SELECT {k} FROM {t} WHERE sync_batch_id LIKE %s)
        """,
        (args.api_prefix + "%", args.db_prefix + "%"),
    )

    ok = True
    if db["key_cnt"] != api["key_cnt"]:
        print(f"[FAIL] 键数不等：db={db['key_cnt']} api={api['key_cnt']}", file=sys.stderr)
        ok = False
    if only_db:
        print(f"[FAIL] DB 有 API 无 ({len(only_db)})：{[r['biz'] for r in only_db[:20]]}", file=sys.stderr)
        ok = False
    if only_api:
        print(f"[FAIL] API 有 DB 无 ({len(only_api)})：{[r['biz'] for r in only_api[:20]]}", file=sys.stderr)
        ok = False

    if ok:
        print("[PASS] DB 与 API 路线 source_biz_key 集合与行数一致。")
        return 0
    return 1


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