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

"""S1–S4 贴源契约：同一 Mock 载荷以 DB/API 源标签写入对应 mdp_stg_*，比对 source_biz_key。

不跑完整 std（S1–S4 transform 依赖多表 JOIN）；验证执行器 Writer 键规格一致即可。
"""
from __future__ import annotations

import json
import sys
from datetime import datetime

import requests

from _db import get_conn

MOCK = "http://127.0.0.1:8018"
TOKEN = "uat-mock-token"
TS = datetime.now().strftime("%Y%m%d%H%M%S")

# (mod, path, stg_table, source_table, biz_fields)
CASES = [
    ("S1", "/api/sales-order", "mdp_stg_so", "crm_seorder", ["bill_no"]),
    ("S1", "/api/shipment", "mdp_stg_ship_trans", "ASNBOLShipperDetail", ["Id", "Line"]),
    ("S2", "/api/work-order", "mdp_stg_schedule", "WorkOrdMaster", ["Domain", "WorkOrd"]),
    ("S2", "/api/schedule", "mdp_stg_schedule", "ScheduleResultOpMaster", ["Domain", "WorkOrd", "Op", "WorkDate"]),
    ("S3", "/api/purchase-order", "mdp_stg_purchase_order", "PurOrdDetail", ["Domain", "PurOrd", "Line"]),
    ("S3", "/api/delivery", "mdp_stg_delivery", "srm_polist_ds", ["dsnum"]),
    ("S3", "/api/supplier", "mdp_stg_supplier", "SuppMaster", ["Domain", "Supp"]),
    ("S4", "/api/s4-shipment", "mdp_stg_s4_shipment", "scm_shdzb", ["glid", "id"]),
    ("S4", "/api/s4-iqc", "mdp_stg_s4_iqc", "PurOrdRctDetail", ["Domain", "Receiver", "Line"]),
    ("S4", "/api/s4-return", "mdp_stg_s4_return", "srm_polist_ds", ["dsnum"]),
]


def fetch(path: str) -> list[dict]:
    r = requests.get(MOCK.rstrip("/") + path, headers={"Authorization": f"Bearer {TOKEN}"}, timeout=30)
    r.raise_for_status()
    rows = r.json().get("data", {}).get("list")
    if not isinstance(rows, list):
        raise RuntimeError(f"{path} invalid")
    return rows


def biz_key(row: dict, fields: list[str]) -> str:
    vals = []
    for f in fields:
        # mock 样本字段大小写可能不一致
        if f in row and row[f] not in (None, ""):
            vals.append(str(row[f]))
            continue
        # 尝试常见别名
        alt = next((k for k in row if k.lower() == f.lower()), None)
        if alt is None or row[alt] in (None, ""):
            return str(row.get("bizKey") or row.get("id") or row.get("Id") or row.get("RecID"))
        vals.append(str(row[alt]))
    return "#".join(vals)


def upsert(cur, stg: str, source_system: str, source_table: str, row: dict, fields: list[str], batch: str):
    biz = biz_key(row, fields)
    rid = str(row.get("RecID") or row.get("id") or row.get("Id") or row.get("bizKey") or biz)
    raw = json.dumps(row, ensure_ascii=False)
    cur.execute(
        f"""
        INSERT INTO {stg}
          (tenant_id, source_system, source_table, source_row_id, source_biz_key,
           raw_data, sync_batch_id, sync_time, process_status)
        VALUES (0, %s, %s, %s, %s, %s, %s, NOW(), 'PENDING')
        ON DUPLICATE KEY UPDATE
          source_row_id=VALUES(source_row_id), raw_data=VALUES(raw_data),
          sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
          process_status='PENDING'
        """,
        (source_system, source_table, rid, biz, raw, batch),
    )


def keys_of(cur, stg: str, batch: str) -> set[str]:
    cur.execute(f"SELECT source_biz_key AS k FROM {stg} WHERE sync_batch_id=%s", (batch,))
    return {str(r["k"]) for r in cur.fetchall() if r["k"] is not None}


def run_case(conn, mod: str, path: str, stg: str, source_table: str, fields: list[str]) -> bool:
    rows = fetch(path)
    db_batch = f"CT_DB_{mod}_{source_table}_{TS}"
    api_batch = f"CT_API_{mod}_{source_table}_{TS}"
    with conn.cursor() as cur:
        for r in rows:
            upsert(cur, stg, "AIDOPDEV_MYSQL", source_table, r, fields, db_batch)
        keys_db = keys_of(cur, stg, db_batch)
        cur.execute(f"DELETE FROM {stg} WHERE sync_batch_id=%s", (db_batch,))

        for r in rows:
            upsert(cur, stg, "WMS_API", source_table, r, fields, api_batch)
        keys_api = keys_of(cur, stg, api_batch)

    label = f"{mod}:{source_table}"
    print(f"[{label}] db={len(keys_db)} api={len(keys_api)} path={path}")
    if not keys_db or not keys_api:
        print(f"[FAIL] {label} empty", file=sys.stderr)
        return False
    if keys_db != keys_api:
        print(f"[FAIL] {label} only_db={sorted(keys_db-keys_api)[:5]} only_api={sorted(keys_api-keys_db)[:5]}", file=sys.stderr)
        return False
    print(f"[PASS] {label}")
    return True


def main() -> int:
    conn = get_conn()
    try:
        results = [run_case(conn, *c) for c in CASES]
    finally:
        conn.close()
    if not all(results):
        print(f"[FAIL] {sum(1 for x in results if not x)}/{len(results)} cases", file=sys.stderr)
        return 1
    print(f"[PASS] S1-S4 stg contract {len(results)} cases (ts={TS})")
    return 0


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