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

#!/usr/bin/env python3
"""WP-SD7 module readiness inventory.

Reusable. Never auto-passes a module.
Checks L1→L2 (and L3/L4 when configured), operational-factory facts,
KPI day coverage, S8 exception presence, and improvement-loop status.
Live create→approve→execute→verify→close is recorded only when actually run.
"""

from __future__ import annotations

import json
import re
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path

import pymysql


ROOT = Path(__file__).resolve().parents[4]
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"
EVIDENCE = ROOT / "doc" / "plan" / "UAT留证" / "2026-08-17-智慧诊断正式落地"
MODULES = ("S1", "S2", "S3", "S4", "S5", "S6", "S7", "S9")
TENANTS = {
    "A": {"code": "A", "name": "UATTEST_CHL", "tenant_id": 838257186181189},
    "B": {"code": "B", "name": "UATTEST_CHLB", "tenant_id": 838257212780613},
    "DEMO": {"code": "DEMO", "name": "UATDEMO", "tenant_id": 838257237606469},
}
MODULE_SOURCES = {
    "S1": ("order_schedule",),
    "S2": ("order_schedule",),
    "S3": ("supplier_delivery", "material_readiness"),
    "S4": ("supplier_delivery",),
    "S5": ("iqc_bill",),
    "S6": ("order_schedule", "s6_report"),
    "S7": ("fqc_result",),
    "S9": (
        "order_schedule",
        "supplier_delivery",
        "material_readiness",
        "s6_report",
        "iqc_bill",
        "fqc_result",
    ),
}
SOURCE_SQL = {
    "order_schedule": """
        SELECT COUNT(*) c FROM dwd_order_schedule_trans
        WHERE tenant_id=%s AND IFNULL(work_order,'')<>''
          AND (factory_id IS NULL OR factory_id=0 OR factory_id=1 OR factory_id=%s)
          AND (factory_id IS NULL OR factory_id<>%s)
    """,
    "supplier_delivery": """
        SELECT COUNT(*) c FROM dwd_supplier_delivery
        WHERE tenant_id=%s AND IFNULL(po_no,'')<>''
          AND (factory_id IS NULL OR factory_id=0 OR factory_id=1 OR factory_id=%s)
          AND (factory_id IS NULL OR factory_id<>%s)
    """,
    "material_readiness": """
        SELECT COUNT(*) c FROM dwd_material_readiness
        WHERE tenant_id=%s AND IFNULL(shortage_qty,0)>0
          AND (factory_id IS NULL OR factory_id=0 OR factory_id=1 OR factory_id=%s)
          AND (factory_id IS NULL OR factory_id<>%s)
    """,
    "s6_report": """
        SELECT COUNT(*) c FROM mdp_std_s6_report
        WHERE tenant_id=%s AND IFNULL(work_order_no,'')<>''
          AND (factory_id IS NULL OR factory_id=0 OR factory_id=1 OR factory_id=%s)
          AND (factory_id IS NULL OR factory_id<>%s)
    """,
    "iqc_bill": """
        SELECT COUNT(*) c FROM qms_qcp_inspbill
        WHERE tenant_id=%s AND IFNULL(FBILLNO,'')<>''
    """,
    "fqc_result": """
        SELECT COUNT(*) c FROM mdp_std_fqc_result
        WHERE tenant_id=%s AND IFNULL(bill_no,'')<>''
    """,
}


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",
        cursorclass=pymysql.cursors.DictCursor,
    )


def json_safe(value: object) -> object:
    if isinstance(value, datetime):
        return value.isoformat(timespec="seconds")
    if isinstance(value, date):
        return value.isoformat()
    if isinstance(value, Decimal):
        return str(value)
    if isinstance(value, dict):
        return {str(key): json_safe(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [json_safe(item) for item in value]
    if isinstance(value, int) and abs(value) > 2**53:
        return str(value)
    return value


def scalar(cur: pymysql.cursors.DictCursor, sql: str, params: tuple = ()) -> int:
    cur.execute(sql, params)
    row = cur.fetchone() or {}
    return int(next(iter(row.values())) or 0)


def fetch_all(cur: pymysql.cursors.DictCursor, sql: str, params: tuple = ()) -> list[dict]:
    cur.execute(sql, params)
    return [json_safe(row) for row in cur.fetchall()]


def tenant_meta(cur: pymysql.cursors.DictCursor, tenant_id: int) -> dict:
    cur.execute(
        """
        SELECT CAST(Id AS CHAR) id, Title title, CAST(OrgId AS CHAR) org_id
        FROM SysTenant WHERE Id=%s
        """,
        (tenant_id,),
    )
    row = cur.fetchone()
    return json_safe(row) if row else {"id": str(tenant_id), "missing": True}


def trees(cur: pymysql.cursors.DictCursor, tenant_id: int, module: str) -> list[dict]:
    return fetch_all(
        cur,
        """
        SELECT p.MetricCode l1_code, p.MetricName l1_name,
               COUNT(DISTINCT CASE WHEN c2.MetricLevel=2 THEN c2.Id END) l2_count,
               COUNT(DISTINCT CASE WHEN c3.MetricLevel=3 THEN c3.Id END) l3_count,
               COUNT(DISTINCT CASE WHEN c4.MetricLevel=4 THEN c4.Id END) l4_count
        FROM ado_smart_ops_kpi_master p
        LEFT JOIN ado_smart_ops_kpi_master c2
          ON c2.TenantId=p.TenantId AND c2.ParentId=p.Id AND c2.MetricLevel=2
        LEFT JOIN ado_smart_ops_kpi_master c3
          ON c3.TenantId=p.TenantId AND c3.ParentId=c2.Id AND c3.MetricLevel=3
        LEFT JOIN ado_smart_ops_kpi_master c4
          ON c4.TenantId=p.TenantId AND c4.ParentId=c3.Id AND c4.MetricLevel=4
        WHERE p.TenantId=%s AND p.ModuleCode=%s AND p.MetricLevel=1 AND IFNULL(p.IsEnabled,1)=1
        GROUP BY p.MetricCode, p.MetricName
        ORDER BY p.MetricCode
        """,
        (tenant_id, module),
    )


def kpi_days(cur: pymysql.cursors.DictCursor, tenant_id: int, factory_id: int, module: str) -> dict:
    result = {}
    for level, table in (
        ("L1", "ado_s9_kpi_value_l1_day"),
        ("L2", "ado_s9_kpi_value_l2_day"),
        ("L3", "ado_s9_kpi_value_l3_day"),
        ("L4", "ado_s9_kpi_value_l4_day"),
    ):
        result[level] = fetch_all(
            cur,
            f"""
            SELECT COUNT(*) row_count, COUNT(DISTINCT metric_code) metric_count,
                   MIN(biz_date) min_date, MAX(biz_date) max_date
            FROM {table}
            WHERE tenant_id=%s AND module_code=%s AND IFNULL(is_deleted,0)=0
              AND (factory_id IS NULL OR factory_id=0 OR factory_id=1 OR factory_id=%s)
              AND (factory_id IS NULL OR factory_id<>%s)
            """,
            (tenant_id, module, factory_id, tenant_id),
        )[0]
    return result


def facts(cur: pymysql.cursors.DictCursor, tenant_id: int, factory_id: int, module: str) -> dict:
    out = {}
    for key in MODULE_SOURCES[module]:
        sql = SOURCE_SQL[key]
        if key in ("iqc_bill", "fqc_result"):
            out[key] = scalar(cur, sql, (tenant_id,))
        else:
            out[key] = scalar(cur, sql, (tenant_id, factory_id, tenant_id))
    return out


def module_report(cur: pymysql.cursors.DictCursor, tenant: dict, factory_id: int, module: str) -> dict:
    tree = trees(cur, tenant["tenant_id"], module)
    complete = [row for row in tree if int(row["l2_count"] or 0) > 0]
    natural_stop = [row["l1_code"] for row in tree if int(row["l2_count"] or 0) == 0]
    with_l3 = [row["l1_code"] for row in tree if int(row["l3_count"] or 0) > 0]
    with_l4 = [row["l1_code"] for row in tree if int(row["l4_count"] or 0) > 0]
    day = kpi_days(cur, tenant["tenant_id"], factory_id, module)
    fact = facts(cur, tenant["tenant_id"], factory_id, module)
    fact_total = sum(fact.values())
    l1_rows = int(day["L1"]["row_count"] or 0)
    blockers = []
    if not complete:
        blockers.append("no_l1_l2_chain")
    if fact_total <= 0:
        blockers.append("no_registered_facts")
    if l1_rows <= 0 and int(day["L2"]["row_count"] or 0) <= 0:
        blockers.append("no_kpi_day_values")
    plans = fetch_all(
        cur,
        """
        SELECT CAST(Id AS CHAR) id, PlanNo plan_no, Status status, RootCause root_cause
        FROM ado_smart_ops_improvement_plan
        WHERE TenantId=%s AND ModuleCode=%s
        ORDER BY CreateTime DESC
        """,
        (tenant["tenant_id"], module),
    )
    closed_loop = any(
        row["status"] == "closed" and "[WP-SD7/" in str(row.get("root_cause") or "")
        for row in plans
    )
    if not closed_loop:
        blockers.append("improvement_loop_not_closed")
    return {
        "module": module,
        "l1_count": len(tree),
        "complete_l1_l2": [row["l1_code"] for row in complete],
        "natural_stop_no_l2": natural_stop,
        "l3_configured": with_l3,
        "l4_configured": with_l4,
        "trees": tree,
        "kpi_days": day,
        "facts": fact,
        "fact_total": fact_total,
        "plans": plans,
        "blockers": blockers,
        "verdict": "blocked" if blockers else "ready_for_signoff",
        "auto_passed": False,
    }


def s8_report(cur: pymysql.cursors.DictCursor, tenant_id: int) -> dict:
    count = scalar(
        cur,
        "SELECT COUNT(*) c FROM ado_s8_exception WHERE tenant_id=%s AND IFNULL(is_deleted,0)=0",
        (tenant_id,),
    )
    linked = scalar(
        cur,
        """
        SELECT COUNT(*) c
        FROM ado_smart_ops_improvement_plan p
        WHERE p.TenantId=%s AND p.ModuleCode='S8'
        """,
        (tenant_id,),
    )
    blockers = []
    if count <= 0:
        blockers.append("no_s8_exception")
    if linked <= 0:
        blockers.append("no_s8_linked_plan")
    return {
        "exception_count": count,
        "linked_plan_count": linked,
        "blockers": blockers,
        "verdict": "blocked" if blockers else "ready_for_signoff",
        "note": "S8 does not copy KPI diagnosis tree",
        "auto_passed": False,
    }


def isolation(cur: pymysql.cursors.DictCursor, tenant_ids: list[int]) -> dict:
    leaks = []
    for table, column in (
        ("dwd_order_schedule_trans", "tenant_id"),
        ("dwd_supplier_delivery", "tenant_id"),
        ("dwd_material_readiness", "tenant_id"),
        ("mdp_std_s6_report", "tenant_id"),
        ("qms_qcp_inspbill", "tenant_id"),
        ("mdp_std_fqc_result", "tenant_id"),
    ):
        for tenant_id in tenant_ids:
            other = [x for x in tenant_ids if x != tenant_id]
            placeholders = ",".join(["%s"] * len(other))
            count = scalar(
                cur,
                f"SELECT COUNT(*) c FROM {table} WHERE {column}=%s AND {column} IN ({placeholders})",
                (tenant_id, *other),
            )
            if count:
                leaks.append({"table": table, "tenant_id": str(tenant_id), "count": count})
    factory_eq_tenant = {}
    for table in (
        "dwd_order_schedule_trans",
        "dwd_supplier_delivery",
        "dwd_material_readiness",
        "mdp_std_s6_report",
        "ado_s9_kpi_value_l1_day",
    ):
        factory_eq_tenant[table] = fetch_all(
            cur,
            f"""
            SELECT CAST(tenant_id AS CHAR) tenant_id, COUNT(*) row_count
            FROM {table}
            WHERE tenant_id IN ({",".join(["%s"] * len(tenant_ids))})
              AND factory_id=tenant_id
            GROUP BY tenant_id
            """,
            tuple(tenant_ids),
        )
    return {"cross_tenant_impossible_leaks": leaks, "factory_eq_tenant": factory_eq_tenant}


def main() -> None:
    EVIDENCE.mkdir(parents=True, exist_ok=True)
    conn = connect()
    try:
        cur = conn.cursor()
        tenants = []
        any_blocked = False
        for item in TENANTS.values():
            meta = tenant_meta(cur, item["tenant_id"])
            factory_id = int(meta.get("org_id") or 0)
            modules = [
                module_report(cur, item, factory_id, module) for module in MODULES
            ]
            s8 = s8_report(cur, item["tenant_id"])
            if any(row["verdict"] != "ready_for_signoff" for row in modules) or s8["verdict"] != "ready_for_signoff":
                any_blocked = True
            tenants.append(
                {
                    "tenant": {**item, **meta, "factory_id": str(factory_id)},
                    "modules": modules,
                    "s8": s8,
                }
            )
        payload = {
            "work_package": "WP-SD7",
            "generated_at": datetime.now().isoformat(timespec="seconds"),
            "auto_passed": False,
            "overall_verdict": "blocked" if any_blocked else "ready_for_signoff",
            "must_not_pass": [
                "page_open_only",
                "kpi_without_facts",
                "other_tenant_facts",
                "hand_edited_status",
                "canned_root_cause",
                "s1_only_then_declare_all_done",
            ],
            "isolation": isolation(cur, [x["tenant_id"] for x in TENANTS.values()]),
            "tenants": tenants,
        }
        out = EVIDENCE / "07-module-readiness.json"
        out.write_text(json.dumps(json_safe(payload), ensure_ascii=False, indent=2), encoding="utf-8")
        print(out)
        print("overall_verdict=", payload["overall_verdict"])
        for tenant in tenants:
            print(tenant["tenant"]["name"], "S8", tenant["s8"]["verdict"])
            for module in tenant["modules"]:
                print(
                    " ",
                    module["module"],
                    module["verdict"],
                    "facts=",
                    module["fact_total"],
                    "blockers=",
                    ",".join(module["blockers"]) or "-",
                )
    finally:
        conn.close()


if __name__ == "__main__":
    main()