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

#!/usr/bin/env python3
"""WP-SD0 baseline inventory for Smart Diagnosis formalization.

Reusable: rerun this script to regenerate the same evidence set.
All bigint IDs are serialized as strings to avoid JSON precision loss.
The script never marks empty modules as passed.
"""

from __future__ import annotations

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

import pymysql
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter


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},
}
INVALID_TENANTS = (0, 1, 1300000000001)
FACT_TABLES = (
    ("dwd_requirement_examine_detail", "tenant_id"),
    ("dwd_ship_trans", "tenant_id"),
    ("dwd_order_schedule_trans", "tenant_id"),
    ("dwd_supplier_delivery", "tenant_id"),
    ("dwd_s4_purchase_execution", "tenant_id"),
    ("dwd_material_readiness", "tenant_id"),
    ("dwd_material_shortage", "tenant_id"),
    ("dwd_supplier_risk", "tenant_id"),
    ("dwd_qc_trans", "tenant_id"),
    ("mdp_std_so", "tenant_id"),
    ("mdp_std_s6_report", "tenant_id"),
    ("mdp_std_fqc_result", "tenant_id"),
    ("mdp_std_ipqc_inspection", "tenant_id"),
    ("mdp_std_s4_iqc", "tenant_id"),
    ("qms_qcp_inspbill", "tenant_id"),
    ("qms_gcjyd", "tenant_id"),
    ("qms_qcpp_inspbill", "tenant_id"),
    ("ado_s8_exception", "tenant_id"),
    ("ado_smart_ops_kpi_atomic_day", "tenant_id"),
    ("ado_smart_ops_improvement_plan", "TenantId"),
)
HEADER_FILL = PatternFill("solid", fgColor="1F4E79")
HEADER_FONT = Font(color="FFFFFF", bold=True)
WARN_FILL = PatternFill("solid", fgColor="FFF2CC")
EMPTY_FILL = PatternFill("solid", fgColor="F4CCCC")


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, bytes):
        return value.decode("utf-8", errors="replace")
    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, Status status
        FROM SysTenant
        WHERE Id=%s
        """,
        (tenant_id,),
    )
    row = cur.fetchone()
    return json_safe(row) if row else {"id": str(tenant_id), "missing": True}


def kpi_master(cur: pymysql.cursors.DictCursor, tenant_id: int) -> dict:
    by_module = fetch_all(
        cur,
        """
        SELECT ModuleCode module_code, MetricLevel metric_level,
               COUNT(*) metric_count,
               SUM(CASE WHEN IFNULL(IsEnabled,1)=1 THEN 1 ELSE 0 END) enabled_count
        FROM ado_smart_ops_kpi_master
        WHERE TenantId=%s
        GROUP BY ModuleCode, MetricLevel
        ORDER BY ModuleCode, MetricLevel
        """,
        (tenant_id,),
    )
    trees = fetch_all(
        cur,
        """
        SELECT p.ModuleCode module_code,
               p.MetricCode l1_code,
               p.MetricName l1_name,
               CAST(p.Id AS CHAR) l1_id,
               SUM(CASE WHEN c.MetricLevel=2 THEN 1 ELSE 0 END) l2_count,
               SUM(CASE WHEN c.MetricLevel=3 THEN 1 ELSE 0 END) l3_count,
               SUM(CASE WHEN c.MetricLevel=4 THEN 1 ELSE 0 END) l4_count
        FROM ado_smart_ops_kpi_master p
        LEFT JOIN ado_smart_ops_kpi_master c
          ON c.TenantId=p.TenantId AND c.ParentId=p.Id
        WHERE p.TenantId=%s AND p.MetricLevel=1
        GROUP BY p.ModuleCode, p.MetricCode, p.MetricName, p.Id
        ORDER BY p.ModuleCode, p.MetricCode
        """,
        (tenant_id,),
    )
    missing_modules = [
        module
        for module in MODULES
        if not any(row["module_code"] == module for row in by_module)
    ]
    incomplete_l1 = [
        row
        for row in trees
        if int(row["l2_count"] or 0) == 0
    ]
    return {
        "by_module_level": by_module,
        "l1_trees": trees,
        "missing_modules": missing_modules,
        "l1_without_l2": incomplete_l1,
        "total": sum(int(row["metric_count"]) for row in by_module),
    }


def daily_values(cur: pymysql.cursors.DictCursor, tenant_id: int) -> dict:
    result: dict[str, object] = {}
    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 module_code,
                   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 IFNULL(is_deleted,0)=0
            GROUP BY module_code
            ORDER BY module_code
            """,
            (tenant_id,),
        )
    result["atomic"] = fetch_all(
        cur,
        """
        SELECT domain_code, metric_code,
               COUNT(*) row_count,
               MIN(stat_date) min_date,
               MAX(stat_date) max_date,
               SUM(CASE WHEN IFNULL(customer_code,'')<>''
                          OR IFNULL(product_code,'')<>''
                          OR IFNULL(order_no,'')<>''
                          OR IFNULL(supplier_code,'')<>''
                          OR IFNULL(material_code,'')<>''
                          OR IFNULL(work_order_no,'')<>''
                        THEN 1 ELSE 0 END) dimensioned_rows,
               SUM(CASE WHEN IFNULL(grain,'') IN ('module','') THEN 1 ELSE 0 END) summary_rows
        FROM ado_smart_ops_kpi_atomic_day
        WHERE tenant_id=%s AND IFNULL(is_deleted,0)=0
        GROUP BY domain_code, metric_code
        ORDER BY domain_code, metric_code
        """,
        (tenant_id,),
    )
    return result


def scope_proxy(atomic_rows: list[dict]) -> dict:
    """dataSource.scope is computed at query time; persist a data-side proxy."""
    total = sum(int(row["row_count"]) for row in atomic_rows)
    dimensioned = sum(int(row["dimensioned_rows"] or 0) for row in atomic_rows)
    summary = sum(int(row["summary_rows"] or 0) for row in atomic_rows)
    return {
        "note": "scope is API-computed; this is a data-side proxy from atomic grain/dimensions",
        "atomic_total": total,
        "module_summary_proxy": summary,
        "filtered_atomic_proxy": dimensioned,
        "partial_filtered_proxy": max(total - summary - dimensioned, 0),
        "empty": total == 0,
    }


def improvement_plans(cur: pymysql.cursors.DictCursor, tenant_id: int) -> dict:
    plans = fetch_all(
        cur,
        """
        SELECT CAST(Id AS CHAR) id,
               CAST(TenantId AS CHAR) tenant_id,
               CAST(FactoryId AS CHAR) factory_id,
               PlanNo plan_no,
               ModuleCode module_code,
               MetricCode metric_code,
               ProblemLevel problem_level,
               ProblemName problem_name,
               Status status,
               CAST(OwnerUserId AS CHAR) owner_user_id,
               CAST(FlowInstanceId AS CHAR) flow_instance_id,
               VerifyResult verify_result,
               CASE
                 WHEN ActionItemsJson IS NULL OR ActionItemsJson='' THEN 0
                 ELSE 1
               END has_action_json,
               CHAR_LENGTH(IFNULL(ActionItemsJson,'')) action_json_chars,
               CreateTime create_time,
               UpdateTime update_time
        FROM ado_smart_ops_improvement_plan
        WHERE TenantId=%s
        ORDER BY CreateTime
        """,
        (tenant_id,),
    )
    by_status: dict[str, int] = {}
    for plan in plans:
        status = str(plan.get("status") or "UNKNOWN")
        by_status[status] = by_status.get(status, 0) + 1
    flow_ids = [int(plan["flow_instance_id"]) for plan in plans if plan.get("flow_instance_id")]
    approval = []
    if flow_ids:
        placeholders = ",".join(["%s"] * len(flow_ids))
        approval = fetch_all(
            cur,
            f"""
            SELECT CAST(Id AS CHAR) id,
                   CAST(BizId AS CHAR) biz_id,
                   Status status,
                   CurrentNodeId current_node_id
            FROM ApprovalFlowInstance
            WHERE Id IN ({placeholders})
            """,
            tuple(flow_ids),
        )
    return {
        "count": len(plans),
        "by_status": by_status,
        "with_action_json": sum(int(plan["has_action_json"]) for plan in plans),
        "with_flow": len(flow_ids),
        "plans": plans,
        "approval_instances": approval,
    }


def pollution(cur: pymysql.cursors.DictCursor) -> dict:
    rows = []
    for table, tenant_col in FACT_TABLES + (
        ("ado_s9_kpi_value_l1_day", "tenant_id"),
        ("ado_s9_kpi_value_l2_day", "tenant_id"),
        ("ado_s9_kpi_value_l3_day", "tenant_id"),
        ("ado_s9_kpi_value_l4_day", "tenant_id"),
        ("ado_smart_ops_kpi_master", "TenantId"),
    ):
        factory_col = None
        cur.execute(
            """
            SELECT COLUMN_NAME
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s
              AND COLUMN_NAME IN ('factory_id','FactoryId')
            """,
            (table,),
        )
        found = cur.fetchone()
        if found:
            factory_col = found["COLUMN_NAME"]
        invalid_sql = f"""
            SELECT COUNT(*) n
            FROM `{table}`
            WHERE `{tenant_col}` IS NULL OR `{tenant_col}` IN %s
        """
        invalid = scalar(cur, invalid_sql, (INVALID_TENANTS,))
        same_scope = 0
        if factory_col:
            same_scope = scalar(
                cur,
                f"""
                SELECT COUNT(*) n FROM `{table}`
                WHERE `{factory_col}` IS NOT NULL
                  AND `{factory_col}`=`{tenant_col}`
                """,
            )
        rows.append(
            {
                "table": table,
                "invalid_tenant_rows": invalid,
                "factory_eq_tenant_rows": same_scope,
            }
        )
    return {
        "invalid_tenant_total": sum(int(row["invalid_tenant_rows"]) for row in rows),
        "factory_eq_tenant_total": sum(int(row["factory_eq_tenant_rows"]) for row in rows),
        "tables": rows,
    }


def fact_counts(cur: pymysql.cursors.DictCursor, tenant_id: int) -> list[dict]:
    result = []
    for table, tenant_col in FACT_TABLES:
        count = scalar(
            cur,
            f"SELECT COUNT(*) n FROM `{table}` WHERE `{tenant_col}`=%s",
            (tenant_id,),
        )
        result.append({"table": table, "row_count": count, "empty": count == 0})
    return result


def collect_tenant(cur: pymysql.cursors.DictCursor, spec: dict) -> dict:
    tenant_id = spec["tenant_id"]
    master = kpi_master(cur, tenant_id)
    values = daily_values(cur, tenant_id)
    facts = fact_counts(cur, tenant_id)
    empty_facts = [row["table"] for row in facts if row["empty"]]
    empty_value_modules = {
        level: [module for module in MODULES if not any(row["module_code"] == module for row in values[level])]
        for level in ("L1", "L2", "L3", "L4")
    }
    return {
        "code": spec["code"],
        "name": spec["name"],
        "tenant_id": str(tenant_id),
        "sys_tenant": tenant_meta(cur, tenant_id),
        "kpi_master": master,
        "daily_values": values,
        "scope_proxy": scope_proxy(values["atomic"]),
        "improvement_plans": improvement_plans(cur, tenant_id),
        "fact_tables": facts,
        "gaps": {
            "missing_kpi_modules": master["missing_modules"],
            "l1_without_l2": [row["l1_code"] for row in master["l1_without_l2"]],
            "empty_value_modules": empty_value_modules,
            "empty_fact_tables": empty_facts,
        },
    }


def autofit(ws) -> None:
    for column in ws.columns:
        letter = get_column_letter(column[0].column)
        width = 12
        for cell in column:
            width = max(width, min(len(str(cell.value or "")), 48))
        ws.column_dimensions[letter].width = width + 2


def style_header(ws) -> None:
    for cell in ws[1]:
        cell.fill = HEADER_FILL
        cell.font = HEADER_FONT
        cell.alignment = Alignment(horizontal="center")


def write_excel(path: Path, sheets: dict[str, list[list[object]]]) -> None:
    wb = Workbook()
    first = True
    for name, rows in sheets.items():
        ws = wb.active if first else wb.create_sheet(name)
        if first:
            ws.title = name
            first = False
        for row in rows:
            ws.append(row)
        if rows:
            style_header(ws)
            for excel_row in ws.iter_rows(min_row=2):
                if any(str(cell.value) in {"0", "[]", "MISSING", "EMPTY"} for cell in excel_row):
                    for cell in excel_row:
                        if str(cell.value) in {"0", "[]", "MISSING", "EMPTY"}:
                            cell.fill = EMPTY_FILL
        autofit(ws)
    path.parent.mkdir(parents=True, exist_ok=True)
    wb.save(path)


def main() -> None:
    EVIDENCE.mkdir(parents=True, exist_ok=True)
    conn = connect()
    payload: dict[str, object] = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "script": "doc/plan/sql/uat-data/run_wp_sd0_baseline.py",
        "gate": "WP-SD0 baseline only; empty modules are listed, not passed",
        "tenants": {},
        "pollution": {},
        "empty_tables": [],
        "notes": [
            "dataSource.scope is computed at query time; baseline stores an atomic-grain proxy.",
            "All bigint IDs are strings.",
            "Do not treat this file as acceptance pass.",
        ],
    }
    try:
        with conn.cursor() as cur:
            for spec in TENANTS.values():
                tenant = collect_tenant(cur, spec)
                tenant["gaps"]["no_improvement_plans"] = tenant["improvement_plans"]["count"] == 0
                payload["tenants"][spec["code"]] = tenant
            payload["pollution"] = pollution(cur)
    finally:
        conn.close()

    empty_tables = []
    for tenant in payload["tenants"].values():
        empty_tables.extend(
            {
                "tenant": tenant["code"],
                "table": table,
            }
            for table in tenant["gaps"]["empty_fact_tables"]
        )
        for module in tenant["kpi_master"]["missing_modules"]:
            empty_tables.append({"tenant": tenant["code"], "table": f"kpi_master:{module}"})
    payload["empty_tables"] = empty_tables

    baseline_path = EVIDENCE / "00-baseline.json"
    baseline_path.write_text(
        json.dumps(json_safe(payload), ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    kpi_rows = [["tenant", "tenant_id", "module", "level", "metric_count", "enabled_count"]]
    tree_rows = [["tenant", "module", "l1_code", "l1_name", "l2_count", "l3_count", "l4_count", "tree_status"]]
    value_rows = [["tenant", "level", "module", "row_count", "metric_count", "min_date", "max_date"]]
    atomic_rows = [["tenant", "domain", "metric", "row_count", "dimensioned", "summary", "min_date", "max_date"]]
    gap_rows = [["tenant", "gap_type", "detail"]]
    for tenant in payload["tenants"].values():
        for row in tenant["kpi_master"]["by_module_level"]:
            kpi_rows.append(
                [
                    tenant["code"],
                    tenant["tenant_id"],
                    row["module_code"],
                    row["metric_level"],
                    row["metric_count"],
                    row["enabled_count"],
                ]
            )
        for row in tenant["kpi_master"]["l1_trees"]:
            status = "OK" if int(row["l2_count"] or 0) > 0 else "NO_L2"
            tree_rows.append(
                [
                    tenant["code"],
                    row["module_code"],
                    row["l1_code"],
                    row["l1_name"],
                    row["l2_count"],
                    row["l3_count"],
                    row["l4_count"],
                    status,
                ]
            )
        for level in ("L1", "L2", "L3", "L4"):
            for row in tenant["daily_values"][level]:
                value_rows.append(
                    [
                        tenant["code"],
                        level,
                        row["module_code"],
                        row["row_count"],
                        row["metric_count"],
                        row["min_date"],
                        row["max_date"],
                    ]
                )
            for module in tenant["gaps"]["empty_value_modules"][level]:
                value_rows.append([tenant["code"], level, module, 0, 0, "", "",])
                gap_rows.append([tenant["code"], f"empty_{level}", module])
        for row in tenant["daily_values"]["atomic"]:
            atomic_rows.append(
                [
                    tenant["code"],
                    row["domain_code"],
                    row["metric_code"],
                    row["row_count"],
                    row["dimensioned_rows"],
                    row["summary_rows"],
                    row["min_date"],
                    row["max_date"],
                ]
            )
        for module in tenant["kpi_master"]["missing_modules"]:
            gap_rows.append([tenant["code"], "missing_kpi_module", module])
        for code in tenant["gaps"]["l1_without_l2"]:
            gap_rows.append([tenant["code"], "l1_without_l2", code])
        for table in tenant["gaps"]["empty_fact_tables"]:
            gap_rows.append([tenant["code"], "empty_fact_table", table])

    write_excel(
        EVIDENCE / "01-kpi-tree-inventory.xlsx",
        {
            "master": kpi_rows,
            "l1_trees": tree_rows,
            "daily_values": value_rows,
            "atomic": atomic_rows,
            "gaps": gap_rows,
        },
    )

    plan_rows = [[
        "tenant", "plan_id", "plan_no", "module", "metric", "status",
        "factory_id", "owner_user_id", "flow_instance_id", "has_action_json",
        "action_json_chars", "verify_result", "create_time",
    ]]
    approval_rows = [["tenant", "flow_id", "biz_id", "status", "current_node_id"]]
    fact_rows = [["tenant", "table", "row_count", "empty"]]
    pollution_rows = [["table", "invalid_tenant_rows", "factory_eq_tenant_rows"]]
    for tenant in payload["tenants"].values():
        plans = tenant["improvement_plans"]
        if not plans["plans"]:
            plan_rows.append([tenant["code"], "EMPTY", "", "", "", "", "", "", "", 0, 0, "", ""])
        for plan in plans["plans"]:
            plan_rows.append(
                [
                    tenant["code"],
                    plan["id"],
                    plan["plan_no"],
                    plan["module_code"],
                    plan["metric_code"],
                    plan["status"],
                    plan["factory_id"],
                    plan["owner_user_id"],
                    plan["flow_instance_id"],
                    plan["has_action_json"],
                    plan["action_json_chars"],
                    plan["verify_result"],
                    plan["create_time"],
                ]
            )
        for row in plans["approval_instances"]:
            approval_rows.append(
                [tenant["code"], row["id"], row["biz_id"], row["status"], row["current_node_id"]]
            )
        for row in tenant["fact_tables"]:
            fact_rows.append([tenant["code"], row["table"], row["row_count"], "EMPTY" if row["empty"] else "OK"])
    for row in payload["pollution"]["tables"]:
        pollution_rows.append(
            [row["table"], row["invalid_tenant_rows"], row["factory_eq_tenant_rows"]]
        )

    write_excel(
        EVIDENCE / "02-improvement-plan-inventory.xlsx",
        {
            "plans": plan_rows,
            "approvals": approval_rows,
            "fact_tables": fact_rows,
            "pollution": pollution_rows,
        },
    )

    log_path = EVIDENCE / "00-baseline-run.log"
    log_path.write_text(
        "\n".join(
            [
                f"generated_at={payload['generated_at']}",
                f"script={payload['script']}",
                f"invalid_tenant_total={payload['pollution']['invalid_tenant_total']}",
                f"factory_eq_tenant_total={payload['pollution']['factory_eq_tenant_total']}",
                *[
                    f"{code}: kpi={tenant['kpi_master']['total']} plans={tenant['improvement_plans']['count']} empty_facts={len(tenant['gaps']['empty_fact_tables'])}"
                    for code, tenant in payload["tenants"].items()
                ],
                f"empty_entries={len(empty_tables)}",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    print(log_path.read_text(encoding="utf-8"))


if __name__ == "__main__":
    main()