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

#!/usr/bin/env python3
"""WP-DATA0/1 baseline: enabled KPI coverage, trends, targets, source facts."""

from __future__ import annotations

import argparse
import json
from collections import defaultdict
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any

from apply_sql_file import connect


TENANT_ID = 838257186181189
FACTORY_ID = 1
EVIDENCE = Path(__file__).resolve().parents[2] / "UAT留证" / "2026-08-21-全模块实际界面缺失数据闭环"

PIPELINE_CODES = {
    "S1_L1_001", "S1_L1_002", "S1_L1_003", "S1_L1_004",
    "S1_L2_001", "S1_L2_002", "S1_L2_003", "S1_L2_004", "S1_L2_005", "S1_L2_006",
    "S1_L2_010", "S1_L2_011", "S1_L2_012", "S1_L2_013", "S1_L2_014", "S1_L2_015",
    "S1_L3_001", "S1_L3_002", "S1_L3_003", "S1_L3_004", "S1_L3_005",
    "S1_L3_101", "S1_L3_102", "S1_L3_103", "S1_L3_104", "S1_L3_105",
    "S1_L4_001", "S1_L4_002", "S1_L4_003", "S1_L4_004",
    "S1_L4_101", "S1_L4_102", "S1_L4_103", "S1_L4_104",
    "S2_L1_001", "S2_L1_002", "S2_L1_003", "S2_L1_004",
    "S2_L2_001", "S2_L2_002", "S2_L2_003",
    "S2_L3_001", "S2_L3_002", "S2_L3_003", "S2_L3_004", "S2_L3_005", "S2_L3_006",
    "S3_L1_001", "S3_L1_002", "S3_L1_003", "S3_L1_004",
    "S3_L2_004", "S3_L2_005",
    "S3_L3_001", "S3_L3_002", "S3_L3_003", "S3_L3_004", "S3_L3_005", "S3_L3_006",
    "S4_L1_001", "S4_L1_002", "S4_L1_003", "S4_L1_004",
    "S4_L2_001", "S4_L2_002", "S4_L2_003", "S4_L2_004",
    "S4_L3_001", "S4_L3_002", "S4_L3_003", "S4_L3_004",
    "S5_L1_001", "S5_L1_002", "S5_L1_003", "S5_L1_004",
    "S5_L2_001", "S5_L2_002", "S5_L2_003", "S5_L2_004",
    "S6_L1_001", "S6_L1_002", "S6_L1_003",
    "S6_L2_001", "S6_L2_002", "S6_L2_003",
    "S7_L1_001", "S7_L1_002", "S7_L1_003",
    "S7_L2_001", "S7_L2_002",
    "S8_L1_001", "S8_L1_002",
    "S9_L1_001",
}

SEED_ONLY = {
    "S1_L2_004", "S1_L3_201", "S1_L3_202", "S1_L3_203", "S1_L3_204", "S1_L3_205",
    "S1_L4_201", "S1_L4_202", "S1_L4_203", "S1_L4_204",
}

SEMANTIC = {
    "S9_L1_002", "S9_L1_003", "S9_L1_004", "S9_L1_005",
}

NO_STAFF_DENOM = {
    "S3_L2_003", "S3_L2_006",
    "S5_L1_003", "S5_L2_006", "S5_L2_010", "S5_L2_014", "S5_L3_001", "S5_L3_003",
    "S6_L3_003", "S7_L2_003", "S7_L2_007", "S7_L2_011", "S7_L2_015",
}

NO_MRP_SOURCE = {"S3_L2_001", "S3_L2_002", "S3_L2_003"}
NO_PROCESS_REVIEW = {
    "S1_L2_007", "S1_L2_008", "S1_L2_009",
    "S1_L3_301", "S1_L3_302", "S1_L3_303", "S1_L3_304", "S1_L3_305",
    "S1_L4_301", "S1_L4_302", "S1_L4_303", "S1_L4_304",
}
NO_BOM_EVENT = {
    "S1_L3_201", "S1_L3_202", "S1_L3_203", "S1_L3_204", "S1_L3_205",
    "S1_L4_201", "S1_L4_202", "S1_L4_203", "S1_L4_204",
}
NO_DELIVERY_REVIEW_EVENT = {
    "S1_L3_401", "S1_L3_402", "S1_L3_403", "S1_L3_404", "S1_L3_405",
    "S1_L4_401", "S1_L4_402", "S1_L4_403", "S1_L4_404",
}
OEE = {"S6_L3_007"}


def dumps(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=str), encoding="utf-8")


def classify(code: str, has_value: bool, trend_points: int, monthly: bool) -> str:
    if code in SEMANTIC:
        return "BLOCKED_SEMANTIC"
    if code in OEE:
        return "BLOCKED_SOURCE_DATA"
    if code in NO_MRP_SOURCE or code in NO_PROCESS_REVIEW or code in NO_DELIVERY_REVIEW_EVENT:
        return "DISABLE_CANDIDATE" if code in NO_PROCESS_REVIEW or code in NO_DELIVERY_REVIEW_EVENT else "BLOCKED_SOURCE_DATA"
    if code in NO_BOM_EVENT:
        return "BLOCKED_SOURCE_DATA"
    if code in NO_STAFF_DENOM and not has_value:
        return "BLOCKED_SOURCE_DATA"
    if code not in PIPELINE_CODES:
        return "BLOCKED_PIPELINE"
    if not has_value:
        return "BLOCKED_SOURCE_DATA"
    if monthly:
        return "COMPLETE_SUMMARY_ONLY" if trend_points < 7 else "COMPLETE_EXACT"
    if trend_points >= 7:
        return "COMPLETE_EXACT"
    return "COMPLETE_SUMMARY_ONLY" if code.startswith("S8_") else "BLOCKED_SOURCE_DATA"


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--tenant", type=int, default=TENANT_ID)
    parser.add_argument("--factory", type=int, default=FACTORY_ID)
    args = parser.parse_args()
    tenant_id = args.tenant
    factory_id = args.factory
    since = date.today() - timedelta(days=14)

    conn = connect()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT m.Id, m.MetricCode, m.ModuleCode, m.MetricLevel, m.ParentId, m.MetricName,
                       m.Formula, m.Unit, m.Direction, m.IsHomePage, m.StatFrequency, m.IsEnabled
                FROM ado_smart_ops_kpi_master m
                WHERE m.IsEnabled=1 AND m.TenantId IN (0, %s)
                ORDER BY m.ModuleCode, m.MetricLevel, m.MetricCode
                """,
                (tenant_id,),
            )
            masters = cur.fetchall()

            cur.execute(
                """
                SELECT metric_code, biz_date, metric_value, target_value, target_source, calc_time, 1 AS lvl
                FROM ado_s9_kpi_value_l1_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0
                UNION ALL
                SELECT metric_code, biz_date, metric_value, target_value, target_source, calc_time, 2
                FROM ado_s9_kpi_value_l2_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0
                UNION ALL
                SELECT metric_code, biz_date, metric_value, target_value, target_source, calc_time, 3
                FROM ado_s9_kpi_value_l3_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0
                UNION ALL
                SELECT metric_code, biz_date, metric_value, target_value, target_source, calc_time, 4
                FROM ado_s9_kpi_value_l4_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0
                """,
                (tenant_id, factory_id) * 4,
            )
            values = cur.fetchall()

            cur.execute(
                """
                SELECT id, module_code, status, heartbeat_at, error_message, started_at, finished_at
                FROM ado_module_dashboard_rebuild_job
                WHERE tenant_id=%s
                ORDER BY id DESC
                LIMIT 20
                """,
                (tenant_id,),
            )
            jobs = cur.fetchall()

            facts: dict[str, Any] = {}
            for sql, key in (
                ("SELECT COUNT(*) c FROM ado_contract_review WHERE tenant_id=%s", "contract_review"),
                ("SELECT COUNT(*) c FROM ado_contract_review_flow WHERE tenant_id=%s", "contract_review_flow"),
                ("SELECT COUNT(*) c FROM ado_product_design WHERE tenant_id=%s", "product_design"),
                ("SELECT COUNT(*) c FROM mdp_stg_so WHERE tenant_id=%s AND source_table='ado_contract_review'", "stg_contract_review"),
                ("SELECT COUNT(*) c FROM mdp_stg_so WHERE tenant_id=%s AND source_table='ado_contract_review_flow'", "stg_review_flow"),
                ("SELECT COUNT(*) c FROM mdp_stg_so WHERE tenant_id=%s AND source_table='ado_product_design'", "stg_product_design"),
                ("SELECT COUNT(*) c FROM mdp_std_purchase_request WHERE tenant_id=%s", "purchase_request"),
                ("SELECT COUNT(*) c FROM mdp_std_purchase_request WHERE tenant_id=%s AND request_date IS NOT NULL AND send_date IS NOT NULL AND send_date>=request_date", "purchase_request_cycle"),
                ("SELECT COUNT(*) c FROM mdp_std_purchase_receipt WHERE tenant_id=%s", "purchase_receipt"),
                ("SELECT COUNT(*) c FROM qms_qcp_inspbill WHERE tenant_id=%s", "iqc_bill"),
                ("SELECT COUNT(*) c FROM qms_qcp_inspbill WHERE tenant_id=%s AND FINSPESTARTDATE IS NOT NULL AND FINSPEENDDATE IS NOT NULL", "iqc_cycle"),
                ("SELECT COUNT(*) c FROM ado_s8_exception WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0", "s8_exception"),
            ):
                if "%s" in sql and sql.count("%s") == 2:
                    cur.execute(sql, (tenant_id, factory_id))
                else:
                    cur.execute(sql, (tenant_id,))
                facts[key] = int(cur.fetchone()["c"])

            cur.execute(
                """
                SELECT id, exception_code, status, severity, module_code, stage_code, factory_id, created_at, closed_at
                FROM ado_s8_exception
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0
                """,
                (tenant_id, factory_id),
            )
            s8_rows = cur.fetchall()
    finally:
        conn.close()

    by_code: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in values:
        by_code[row["metric_code"]].append(row)
    for rows in by_code.values():
        rows.sort(key=lambda x: x["biz_date"] or date.min, reverse=True)

    enabled = []
    coverage = []
    trends = []
    targets = []
    status_rows = []
    counts = defaultdict(int)
    monthly_codes = {m["MetricCode"] for m in masters if (m.get("StatFrequency") or "") in ("月", "月度") or "人效" in (m["MetricName"] or "") or "周转" in (m["MetricName"] or "")}

    for m in masters:
        code = m["MetricCode"]
        rows = by_code.get(code, [])
        latest = next((r for r in rows if r["metric_value"] is not None), None)
        recent = [r for r in rows if r["biz_date"] and r["biz_date"] >= since and r["metric_value"] is not None]
        points = len({r["biz_date"] for r in recent})
        has_value = latest is not None
        monthly = code in monthly_codes or code.endswith("_003") and "人效" in (m["MetricName"] or "")
        status = classify(code, has_value, points, monthly)
        counts[status] += 1
        enabled.append({
            "metricCode": code,
            "moduleCode": m["ModuleCode"],
            "level": m["MetricLevel"],
            "name": m["MetricName"],
            "home": bool(m["IsHomePage"]),
            "formula": m["Formula"],
        })
        coverage.append({
            "metricCode": code,
            "hasValue": has_value,
            "latestDate": latest["biz_date"] if latest else None,
            "latestValue": latest["metric_value"] if latest else None,
            "calcTime": latest["calc_time"] if latest else None,
        })
        trends.append({
            "metricCode": code,
            "recentPoints": points,
            "allPoints": len({r["biz_date"] for r in rows if r["metric_value"] is not None}),
        })
        targets.append({
            "metricCode": code,
            "targetValue": latest["target_value"] if latest else None,
            "targetSource": latest["target_source"] if latest else None,
        })
        status_rows.append({
            "metricCode": code,
            "moduleCode": m["ModuleCode"],
            "level": m["MetricLevel"],
            "name": m["MetricName"],
            "hasValue": has_value,
            "latestDate": str(latest["biz_date"]) if latest and latest["biz_date"] else None,
            "latestValue": latest["metric_value"] if latest else None,
            "recentPoints": points,
            "status": status,
            "pipeline": code in PIPELINE_CODES,
            "seedOnly": code in SEED_ONLY,
        })

    screenshot = [
        {"page": "/dashboard/home", "module": m["ModuleCode"], "metricCode": m["MetricCode"], "title": m["MetricName"], "missingType": "home-card"}
        for m in masters if m["IsHomePage"]
    ]
    screenshot.extend(
        {"page": f"/aidop/{m['ModuleCode'].lower()}/kanban", "module": m["ModuleCode"], "metricCode": m["MetricCode"], "title": m["MetricName"], "missingType": "module-drill"}
        for m in masters if m["ModuleCode"] in {"S1", "S3", "S5", "S6", "S7", "S8", "S9"}
    )

    payload_status = {
        "generatedAt": datetime.now().isoformat(timespec="seconds"),
        "tenantId": tenant_id,
        "factoryId": factory_id,
        "enabledCount": len(masters),
        "statusCounts": dict(counts),
        "facts": facts,
        "s8Exceptions": s8_rows,
        "rebuildJobs": jobs,
        "items": status_rows,
    }

    dumps(EVIDENCE / "00-baseline-enabled-kpi.json", {"tenantId": tenant_id, "factoryId": factory_id, "count": len(enabled), "items": enabled})
    dumps(EVIDENCE / "01-baseline-value-coverage.json", {"tenantId": tenant_id, "hasValue": sum(1 for x in coverage if x["hasValue"]), "missing": sum(1 for x in coverage if not x["hasValue"]), "items": coverage})
    dumps(EVIDENCE / "02-baseline-trend-coverage.json", {"since": str(since), "items": trends})
    dumps(EVIDENCE / "03-baseline-target-coverage.json", {"items": targets})
    dumps(EVIDENCE / "04-screenshot-metric-map.json", screenshot)
    dumps(EVIDENCE / "05-kpi-status-classification.json", payload_status)
    print(json.dumps({
        "enabled": len(masters),
        "hasValue": sum(1 for x in coverage if x["hasValue"]),
        "missing": sum(1 for x in coverage if not x["hasValue"]),
        "statusCounts": dict(counts),
        "facts": facts,
        "s8": s8_rows,
        "jobs": [{"id": j["id"], "module": j["module_code"], "status": j["status"]} for j in jobs[:8]],
        "evidence": str(EVIDENCE),
    }, ensure_ascii=False, default=str, indent=2))


if __name__ == "__main__":
    main()