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

"""S5/S6/S7 契约：同一 Mock 载荷分别以 DB/API 源标签落 stg→std，比对 source_biz_key 集合。

说明：mdp_std_* 业务唯一键在 DB/API 两路上会冲突，后写会覆盖 sync_batch_id，
故不能依赖「两批同时留在 std 再 LIKE 比对」。本脚本改为：
  1) 跑 DB 路线 → 快照键集 A → 删除该批 std
  2) 跑 API 路线 → 快照键集 B
  3) 断言 A≡B（行数与键集合一致）

用法：
    set MOCK_TOKEN=uat-mock-token
    python run_contract_s5_s6_s7.py
"""
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")


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} data.list invalid")
    return rows


def biz_key(row: dict, fields: list[str]) -> str:
    vals = []
    for f in fields:
        if f not in row or row[f] is None or row[f] == "":
            return str(row.get("bizKey") or row.get("RecID") or row.get("id") or row.get("djbh"))
        vals.append(str(row[f]))
    return "#".join(vals)


def upsert(cur, table: str, source_system: str, source_table: str, row: dict, fields: list[str], batch: str, tenant_id=0):
    biz = biz_key(row, fields)
    rid = str(row.get("RecID") or row.get("id") or row.get("bizKey") or biz)
    raw = json.dumps(row, ensure_ascii=False)
    cur.execute(
        f"""
        INSERT INTO {table}
          (tenant_id, source_system, source_table, source_row_id, source_biz_key,
           raw_data, sync_batch_id, sync_time, process_status, create_time)
        VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), 'PENDING', NOW())
        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', update_time=NOW()
        """,
        (tenant_id, source_system, source_table, rid, biz, raw, batch),
    )


def transform_s5(cur, source_system: str, batch: str):
    cur.execute(
        """
        INSERT INTO mdp_std_purchase_receipt
        (tenant_id, factory_id, source_system, domain, receiver, line, rct_date, supp, sort_name,
         item_num, item_name, item_spec, um, qty_ordered, qty_received, lot_serial, location,
         ord_nbr, ord_line, blanket_line, pur_ord, pur_line, sales_job, address1,
         req, req_line, dop_req, source_biz_key, sync_batch_id, sync_time)
        SELECT
            IFNULL(p.tenant_id, 0), 1, %s,
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')),
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Line')) AS SIGNED),
            STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RctDate')),'null'),''), '%%Y-%%m-%%d %%H:%%i:%%s'),
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Supp')),
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Supp')),
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.ItemNum')), NULL, NULL,
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.UM')),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.QtyOrded')) AS DECIMAL(18,6)),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.QtyReceived')) AS DECIMAL(18,6)),
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.LotSerial')),
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Location')),
            JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.OrdNbr')),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.OrdLine')) AS SIGNED),
            NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL,
            IFNULL(NULLIF(p.source_biz_key,''), CONCAT(
                JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')), '#',
                JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')), '#',
                JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Line')))),
            %s, NOW()
        FROM mdp_stg_purchase_receipt p
        INNER JOIN mdp_stg_purchase_receipt d
          ON d.source_table='PurOrdRctMaster' AND d.source_system=%s AND d.sync_batch_id=%s
         AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
         AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Receiver'))
        WHERE p.source_table='PurOrdRctDetail' AND p.source_system=%s AND p.sync_batch_id=%s
          AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctType'))='rc'
        ON DUPLICATE KEY UPDATE
            qty_received=VALUES(qty_received), sync_batch_id=VALUES(sync_batch_id),
            sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
        """,
        (source_system, batch, source_system, batch, source_system, batch),
    )


def transform_s6(cur, source_system: str, batch: str):
    cur.execute(
        """
        INSERT INTO mdp_std_ipqc_inspection
        (tenant_id, factory_id, source_system, bill_no, product_model, production_batch_no, production_work_order,
         result_judgement, attachment, remark, inspector, process_code, process_name, production_person,
         sample_qty, form_no, version_no, effective_date, material_code, material_name,
         inspec_standard_version, inspec_standard_code, inspection_status,
         source_row_id, source_biz_key, sync_batch_id, sync_time)
        SELECT
            IFNULL(m.tenant_id, 1300000000001), 1, %s,
            IFNULL(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.djbh')), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.id'))),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.cplx')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.scph')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.lydjbh')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.jgpd')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.fj')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.bz')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.jyr')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.gxbm')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.gxmc')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.sczyry')),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.ybl')) AS DECIMAL(18,6)),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.bdbh')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.bbh')),
            STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.sxrq')),'null'),''), '%%Y-%%m-%%d %%H:%%i:%%s'),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.wlbm')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.wlmc')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.jgbb')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.jgbh')),
            JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.status')),
            IFNULL(m.source_row_id, JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.id'))),
            IFNULL(NULLIF(m.source_biz_key,''), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.djbh'))),
            %s, NOW()
        FROM mdp_stg_ipqc_pull m
        WHERE m.source_table='qms_gcjyd' AND m.source_system=%s AND m.sync_batch_id=%s
        ON DUPLICATE KEY UPDATE
            bill_no=VALUES(bill_no), sync_batch_id=VALUES(sync_batch_id),
            sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
        """,
        (source_system, batch, source_system, batch),
    )


def transform_s7(cur, source_system: str, batch: str):
    cur.execute(
        """
        INSERT INTO mdp_std_production_receipt
        (tenant_id, factory_id, source_system, domain, master_rec_id, detail_rec_id, nbr, line,
         receipt_date, status, status_desc, remark, prod_line, work_ord, erp_work_ord,
         department, department_desc, applicant_name, item_num, item_name, item_spec, um,
         location_to, location_to_desc, lot_serial, qty_rec, qty_to, location_from, location_from_desc,
         ord_nbr, source_biz_key, sync_batch_id, sync_time)
        SELECT
            IFNULL(n.tenant_id, 0), 1, %s,
            JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain')),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.RecID')) AS SIGNED),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) AS SIGNED),
            JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Nbr')),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Line')) AS SIGNED),
            STR_TO_DATE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Date')),'null'),''), '%%Y-%%m-%%d %%H:%%i:%%s'),
            UPPER(IFNULL(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Status')), '')),
            UPPER(IFNULL(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Status')), '')),
            JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Remark')),
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ProdLine')),
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.WorkOrd')),
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Address')),
            JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Department')),
            JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Department')),
            JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Name')),
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum')), NULL, NULL, NULL,
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LocationTo')), NULL,
            CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LotSerial')) AS CHAR),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyRec')) AS DECIMAL(18,5)),
            CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyTo')) AS DECIMAL(18,5)),
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LocationFrom')), NULL,
            JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdNbr')),
            CONCAT(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain')), '#',
                   JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID'))),
            %s, NOW()
        FROM mdp_stg_production_receipt n
        INNER JOIN mdp_stg_production_receipt d
          ON d.source_table='NbrDetail' AND d.source_system=%s AND d.sync_batch_id=%s
         AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
         AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Nbr')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Nbr'))
        WHERE n.source_table='NbrMaster' AND n.source_system=%s AND n.sync_batch_id=%s
          AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Type'))='WOI'
          AND CAST(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.IsActive')) AS SIGNED)=1
        ON DUPLICATE KEY UPDATE
            qty_rec=VALUES(qty_rec), sync_batch_id=VALUES(sync_batch_id),
            sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
        """,
        (source_system, batch, source_system, batch, source_system, batch),
    )


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


def compare(mod: str, keys_db: set[str], keys_api: set[str]) -> bool:
    only_db = sorted(keys_db - keys_api)
    only_api = sorted(keys_api - keys_db)
    print(f"[{mod}] db_keys={len(keys_db)} api_keys={len(keys_api)}")
    if not keys_db or not keys_api:
        print(f"[FAIL] {mod} 某一侧键集为空", file=sys.stderr)
        return False
    if keys_db != keys_api:
        print(f"[FAIL] {mod} 键集不等 only_db={only_db[:10]} only_api={only_api[:10]}", file=sys.stderr)
        return False
    print(f"[PASS] {mod} DB≡API source_biz_key")
    return True


def dual_s5(conn) -> bool:
    masters = fetch("/api/receipt-master")
    details = fetch("/api/receipt")
    db_batch = f"CT_DB_S5_{TS}"
    api_batch = f"CT_API_S5_{TS}"
    with conn.cursor() as cur:
        for r in masters:
            upsert(cur, "mdp_stg_purchase_receipt", "AIDOPDEV_MYSQL", "PurOrdRctMaster", r, ["Domain", "Receiver"], db_batch)
        for r in details:
            upsert(cur, "mdp_stg_purchase_receipt", "AIDOPDEV_MYSQL", "PurOrdRctDetail", r, ["Domain", "Receiver", "Line"], db_batch)
        transform_s5(cur, "AIDOPDEV_MYSQL", db_batch)
        keys_db = snapshot_keys(cur, "mdp_std_purchase_receipt", db_batch)
        cur.execute("DELETE FROM mdp_std_purchase_receipt WHERE sync_batch_id=%s", (db_batch,))

        for r in masters:
            upsert(cur, "mdp_stg_purchase_receipt", "WMS_API", "PurOrdRctMaster", r, ["Domain", "Receiver"], api_batch)
        for r in details:
            upsert(cur, "mdp_stg_purchase_receipt", "WMS_API", "PurOrdRctDetail", r, ["Domain", "Receiver", "Line"], api_batch)
        transform_s5(cur, "WMS_API", api_batch)
        keys_api = snapshot_keys(cur, "mdp_std_purchase_receipt", api_batch)
    return compare("S5", keys_db, keys_api)


def dual_s6(conn) -> bool:
    rows = fetch("/api/ipqc")
    db_batch = f"CT_DB_S6_{TS}"
    api_batch = f"CT_API_S6_{TS}"
    with conn.cursor() as cur:
        for r in rows:
            upsert(cur, "mdp_stg_ipqc_pull", "AIDOPDEV_MYSQL", "qms_gcjyd", r, ["djbh"], db_batch, tenant_id=1300000000001)
        transform_s6(cur, "AIDOPDEV_MYSQL", db_batch)
        keys_db = snapshot_keys(cur, "mdp_std_ipqc_inspection", db_batch)
        cur.execute("DELETE FROM mdp_std_ipqc_inspection WHERE sync_batch_id=%s", (db_batch,))

        for r in rows:
            upsert(cur, "mdp_stg_ipqc_pull", "WMS_API", "qms_gcjyd", r, ["djbh"], api_batch, tenant_id=1300000000001)
        transform_s6(cur, "WMS_API", api_batch)
        keys_api = snapshot_keys(cur, "mdp_std_ipqc_inspection", api_batch)
    return compare("S6", keys_db, keys_api)


def dual_s7(conn) -> bool:
    masters = fetch("/api/production-receipt")
    details = fetch("/api/production-receipt-detail")
    db_batch = f"CT_DB_S7_{TS}"
    api_batch = f"CT_API_S7_{TS}"
    with conn.cursor() as cur:
        for r in masters:
            upsert(cur, "mdp_stg_production_receipt", "AIDOPDEV_MYSQL", "NbrMaster", r, ["Domain", "Nbr"], db_batch)
        for r in details:
            upsert(cur, "mdp_stg_production_receipt", "AIDOPDEV_MYSQL", "NbrDetail", r, ["Domain", "Nbr", "Line"], db_batch)
        transform_s7(cur, "AIDOPDEV_MYSQL", db_batch)
        keys_db = snapshot_keys(cur, "mdp_std_production_receipt", db_batch)
        cur.execute("DELETE FROM mdp_std_production_receipt WHERE sync_batch_id=%s", (db_batch,))

        for r in masters:
            upsert(cur, "mdp_stg_production_receipt", "WMS_API", "NbrMaster", r, ["Domain", "Nbr"], api_batch)
        for r in details:
            upsert(cur, "mdp_stg_production_receipt", "WMS_API", "NbrDetail", r, ["Domain", "Nbr", "Line"], api_batch)
        transform_s7(cur, "WMS_API", api_batch)
        keys_api = snapshot_keys(cur, "mdp_std_production_receipt", api_batch)
    return compare("S7", keys_db, keys_api)


def main() -> int:
    conn = get_conn()
    try:
        ok = all([dual_s5(conn), dual_s6(conn), dual_s7(conn)])
    finally:
        conn.close()
    if not ok:
        return 1
    print(f"[PASS] S5/S6/S7 contract all green (ts={TS})")
    return 0


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