AiDOPWarehouse/doc/plan/sql/uat-data/run_wp_sd_p039_seed.py

#!/usr/bin/env python3
"""P-039: seed tenant-isolated test facts so diagnosis loops can run.

Does not claim production truth. Rows are tagged P039 / P039_TEST_20260818.
"""

from __future__ import annotations

import re
from pathlib import Path

import pymysql
from pymysql.constants import CLIENT

ROOT = Path(__file__).resolve().parents[4]
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"

A = {"tenant": 838257186181189, "factory": 838257186320453}
B = {"tenant": 838257212780613, "factory": 838257212858437}
DEMO = {"tenant": 838257237606469, "factory": 838257237676101}


def connect() -> pymysql.Connection:
    raw = CONFIG.read_text(encoding="utf-8-sig")
    value = next(
        item
        for item in re.findall(r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw)
        if "Database=aidopdev" in item
    )
    parts = {
        item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip()
        for item in value.split(";")
        if "=" in item
    }
    return pymysql.connect(
        host=parts["server"],
        port=int(parts["port"]),
        user=parts["uid"],
        password=parts["pwd"],
        database=parts["database"],
        charset="utf8mb4",
        autocommit=True,
        client_flag=CLIENT.MULTI_STATEMENTS,
        cursorclass=pymysql.cursors.DictCursor,
    )


SQL = f"""
INSERT INTO ado_s8_exception (
  tenant_id, factory_id, exception_code, title, description, scene_code, source_type,
  status, severity, priority_score, priority_level, occurrence_dept_id, responsible_dept_id,
  timeout_flag, created_at, is_deleted, module_code, consecutive_hit_count, consecutive_miss_count,
  related_object_code, created_by)
SELECT {A['tenant']}, {A['factory']}, 'EX-20260818-P039-A', 'P039 测试异常 产销协同超期',
  'P039 编制测试数据，用于 A 租户 S8 诊断闭环测通', 'S1', 'MANUAL', 'NEW', 'SERIOUS', 80.00, 'P1',
  {A['factory']}, {A['factory']}, 0, NOW(), 0, 'S1', 1, 0, 'UATA-SO-P039', 'P039'
FROM DUAL
WHERE NOT EXISTS (
  SELECT 1 FROM ado_s8_exception
  WHERE tenant_id={A['tenant']} AND factory_id={A['factory']} AND exception_code='EX-20260818-P039-A');

INSERT INTO mdp_std_fqc_result (
  tenant_id, source_system, bill_no, inspect_time, material_code, material_name, judgment,
  inspect_qty, qualified_qty, unqualified_qty, sap_work_order_no, source_row_id, source_biz_key,
  sync_batch_id, sync_time)
SELECT {B['tenant']}, 'P039', 'UATB-P039-FQC-001', NOW(), '3121C0035', 'P039测试物料', 'NG',
  10, 0, 10, 'UATB-WO-001', 'P039-B-FQC-001', 'P039-B-FQC-001', 'P039_TEST_20260818', NOW()
FROM DUAL
WHERE NOT EXISTS (
  SELECT 1 FROM mdp_std_fqc_result
  WHERE tenant_id={B['tenant']} AND source_row_id='P039-B-FQC-001');

INSERT INTO mdp_std_s6_report (
  tenant_id, factory_id, source_system, work_order_no, report_date, report_qty,
  source_row_id, source_biz_key, sync_batch_id, sync_time)
SELECT {B['tenant']}, {B['factory']}, 'P039', 'UATB-WO-001', NOW(), 8,
  'P039-B-S6-001', 'P039-B-S6-001', 'P039_TEST_20260818', NOW()
FROM DUAL
WHERE NOT EXISTS (
  SELECT 1 FROM mdp_std_s6_report
  WHERE tenant_id={B['tenant']} AND source_system='P039' AND source_biz_key='P039-B-S6-001');

INSERT INTO qms_qcp_inspbill (
  id, tenant_id, FBILLNO, FBILLSTATUS, FCREATETIME, SampleQty, TestQty, FCOMMENT, UnqualifiedDescription)
SELECT 101991258985080001, {DEMO['tenant']}, 'UATDEMO-P039-IQC-001', 'NG', NOW(), 10, 10,
  'P039 编制测试 IQC', 'P039 test NG'
FROM DUAL
WHERE NOT EXISTS (
  SELECT 1 FROM qms_qcp_inspbill
  WHERE tenant_id={DEMO['tenant']} AND FBILLNO='UATDEMO-P039-IQC-001');
"""


def main() -> None:
    conn = connect()
    try:
        with conn.cursor() as cur:
            cur.execute(SQL)
            while True:
                if not cur.nextset():
                    break
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT 'A_s8' k, COUNT(*) c FROM ado_s8_exception
                 WHERE tenant_id=%s AND exception_code='EX-20260818-P039-A' AND IFNULL(is_deleted,0)=0
                UNION ALL
                SELECT 'B_fqc', COUNT(*) FROM mdp_std_fqc_result
                 WHERE tenant_id=%s AND bill_no='UATB-P039-FQC-001'
                UNION ALL
                SELECT 'B_s6', COUNT(*) FROM mdp_std_s6_report
                 WHERE tenant_id=%s AND source_biz_key='P039-B-S6-001'
                UNION ALL
                SELECT 'D_iqc', COUNT(*) FROM qms_qcp_inspbill
                 WHERE tenant_id=%s AND FBILLNO='UATDEMO-P039-IQC-001'
                """,
                (A["tenant"], B["tenant"], B["tenant"], DEMO["tenant"]),
            )
            print({row["k"]: row["c"] for row in cur.fetchall()})
    finally:
        conn.close()


if __name__ == "__main__":
    main()