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

#!/usr/bin/env python3
"""Record evidence-backed WP3 B00-B14 status without overstating completion."""

from __future__ import annotations

import json
import re
from datetime import datetime
from pathlib import Path

import pymysql
from pymysql.cursors import DictCursor


ROOT = Path(__file__).resolve().parents[4]
DATABASE_JSON = (
    ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"
)
EVIDENCE = Path(__file__).with_name("WP3-B00-B14-evidence.json")
TENANT_ID = 838257212780613


def connect() -> pymysql.Connection:
    raw = DATABASE_JSON.read_text(encoding="utf-8-sig")
    connection_string = next(
        value
        for value in re.findall(
            r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw
        )
        if "Database=aidopdev" in value
    )
    parts = {
        item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip()
        for item in connection_string.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,
        cursorclass=DictCursor,
    )


def scalar(cur: DictCursor, sql: str, args: tuple = ()) -> int:
    cur.execute(sql, args)
    return int(next(iter(cur.fetchone().values())))


def main() -> None:
    conn = connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS uat_step_execution (
                  id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
                  tenant_id BIGINT NOT NULL,
                  work_package VARCHAR(20) NOT NULL,
                  step_code VARCHAR(20) NOT NULL,
                  status VARCHAR(30) NOT NULL,
                  evidence_json JSON NULL,
                  verified_at DATETIME(3) NOT NULL,
                  UNIQUE KEY uk_uat_step (tenant_id,work_package,step_code)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                """
            )
            facts = {
                "users": scalar(
                    cur,
                    "SELECT COUNT(*) FROM SysUser WHERE TenantId=%s",
                    (TENANT_ID,),
                ),
                "items": scalar(
                    cur,
                    "SELECT COUNT(*) FROM ItemMaster WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "bom_headers": scalar(
                    cur,
                    "SELECT COUNT(*) FROM ProductStructureMaster WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "routing_ops": scalar(
                    cur,
                    "SELECT COUNT(*) FROM RoutingOpDetail WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "sales_orders": scalar(
                    cur,
                    "SELECT COUNT(*) FROM crm_seorder WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "work_orders": scalar(
                    cur,
                    "SELECT COUNT(*) FROM WorkOrdMaster WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "purchase_requests": scalar(
                    cur,
                    "SELECT COUNT(*) FROM srm_pr_main WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "purchase_orders": scalar(
                    cur,
                    "SELECT COUNT(*) FROM PurOrdMaster WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "supplier_shipments": scalar(
                    cur,
                    "SELECT COUNT(*) FROM scm_shd WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "iqc_closed": scalar(
                    cur,
                    """
                    SELECT COUNT(*) FROM ApprovalFlowInstance
                    WHERE BizId IN (
                      SELECT id FROM qms_qcp_inspbill
                      WHERE tenant_id=%s AND FBILLNO='UATB-B11-IQC-001'
                    ) AND Status=2
                    """,
                    (TENANT_ID,),
                ),
                "ipqc_closed": scalar(
                    cur,
                    """
                    SELECT COUNT(*) FROM ApprovalFlowInstance
                    WHERE BizId IN (
                      SELECT id FROM qms_gcjyd
                      WHERE tenant_id=%s AND djbh='UATB-B12-IPQC-001'
                    ) AND Status=2
                    """,
                    (TENANT_ID,),
                ),
                "fqc_closed": scalar(
                    cur,
                    """
                    SELECT COUNT(*) FROM ApprovalFlowInstance
                    WHERE BizId IN (
                      SELECT id FROM qms_qcpp_inspbill
                      WHERE tenant_id=%s AND FBILLNO='UATB-B12-FQC-001'
                    ) AND Status=2
                    """,
                    (TENANT_ID,),
                ),
                "s8_exceptions": scalar(
                    cur,
                    "SELECT COUNT(*) FROM ado_s8_exception WHERE tenant_id=%s",
                    (TENANT_ID,),
                ),
                "successful_modules": scalar(
                    cur,
                    """
                    SELECT COUNT(DISTINCT module_code)
                    FROM ado_module_dashboard_rebuild_job
                    WHERE tenant_id=%s AND status='SUCCESS'
                      AND module_code IN ('S1','S2','S3','S4','S5','S6','S7')
                    """,
                    (TENANT_ID,),
                ),
            }

            steps = {
                "B00": ("PASSED" if facts["users"] >= 4 else "BLOCKED", {"users": facts["users"]}),
                "B01": ("PASSED" if facts["items"] >= 20 else "BLOCKED", {"items": facts["items"]}),
                "B02": (
                    "PASSED"
                    if facts["bom_headers"] > 0 and facts["routing_ops"] > 0
                    else "BLOCKED",
                    {"bom_headers": facts["bom_headers"], "routing_ops": facts["routing_ops"]},
                ),
                "B03": ("CHECKPOINT_READY", {"sales_orders": facts["sales_orders"]}),
                "B04": ("CHECKPOINT_READY", {"sales_orders": facts["sales_orders"]}),
                "B05": ("CHECKPOINT_READY", {"work_orders": facts["work_orders"]}),
                "B06": ("CHECKPOINT_READY", {"work_orders": facts["work_orders"]}),
                "B07": ("CHECKPOINT_READY", {"purchase_requests": facts["purchase_requests"]}),
                "B08": ("CHECKPOINT_READY", {"purchase_orders": facts["purchase_orders"]}),
                "B09": ("CHECKPOINT_READY", {"supplier_shipments": facts["supplier_shipments"]}),
                "B10": ("CHECKPOINT_READY", {"supplier_shipments": facts["supplier_shipments"]}),
                "B11": (
                    "API_VERIFIED" if facts["iqc_closed"] else "PENDING_EXECUTION",
                    {"iqc_closed": facts["iqc_closed"]},
                ),
                "B12": (
                    "API_VERIFIED"
                    if facts["ipqc_closed"] and facts["fqc_closed"]
                    else "PENDING_EXECUTION",
                    {
                        "ipqc_closed": facts["ipqc_closed"],
                        "fqc_closed": facts["fqc_closed"],
                    },
                ),
                "B13": (
                    "API_VERIFIED" if facts["s8_exceptions"] else "PENDING_EXECUTION",
                    {"s8_exceptions": facts["s8_exceptions"]},
                ),
                "B14": (
                    "PASSED" if facts["successful_modules"] == 7 else "BLOCKED",
                    {"successful_modules": facts["successful_modules"]},
                ),
            }
            for step_code, (status, evidence) in steps.items():
                cur.execute(
                    """
                    INSERT INTO uat_step_execution
                      (tenant_id,work_package,step_code,status,evidence_json,verified_at)
                    VALUES(%s,'WP3',%s,%s,%s,NOW(3))
                    ON DUPLICATE KEY UPDATE
                      status=VALUES(status),
                      evidence_json=VALUES(evidence_json),
                      verified_at=VALUES(verified_at)
                    """,
                    (TENANT_ID, step_code, status, json.dumps(evidence)),
                )
    finally:
        conn.close()

    payload = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "tenant_id": str(TENANT_ID),
        "facts": facts,
        "steps": {
            code: {"status": status, "evidence": evidence}
            for code, (status, evidence) in steps.items()
        },
    }
    EVIDENCE.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(json.dumps(payload, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()