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

#!/usr/bin/env python3
"""Generate truthful WP5/WP6 reconciliation and FUNC coverage workbooks."""

from __future__ import annotations

import json
import re
from collections import Counter
from datetime import datetime
from pathlib import Path

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


ROOT = Path(__file__).resolve().parents[4]
ACCEPTANCE = ROOT / "doc" / "plan" / "项目验收"
L1_FILE = ACCEPTANCE / "02-客户项目最低验收对照表.xlsx"
L2_FILE = ACCEPTANCE / "03-产品全量功能验收对照表.xlsx"
OUTPUT = ACCEPTANCE / "06-逐FUNC-UAT覆盖明细.xlsx"
RECON_OUTPUT = ACCEPTANCE / "reconciliation.xlsx"
CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json"

TENANTS = {
    "A": 838257186181189,
    "B": 838257212780613,
    "DEMO": 838257237606469,
}


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 rows_from_sheet(path: Path, sheet: str | int) -> list[dict[str, object]]:
    workbook = load_workbook(path, read_only=True, data_only=True)
    worksheet = workbook.worksheets[sheet] if isinstance(sheet, int) else workbook[sheet]
    rows = list(worksheet.iter_rows(values_only=True))
    headers = [str(value or "").strip() for value in rows[0]]
    result = [
        {headers[index]: value for index, value in enumerate(row)}
        for row in rows[1:]
        if any(value is not None for value in row)
    ]
    workbook.close()
    return result


def find_value(row: dict[str, object], *needles: str) -> object:
    for key, value in row.items():
        normalized = key.replace(" ", "").lower()
        if any(needle.lower() in normalized for needle in needles):
            return value
    return None


def module_of(func_code: str) -> str:
    match = re.match(r"FUNC-(S\d)-", func_code or "")
    return match.group(1) if match else ""


def coverage_for(module: str, name: str, path: str) -> tuple[str, str, str]:
    text = f"{name} {path}".lower()
    if module == "S0":
        return "B", "DATA_READY", "B 套已具备主数据、BOM、工艺、产线与日历前置"
    if module in {"S1", "S2", "S3", "S4"}:
        return "A+B", "DATA_READY", "A/B 已具备订单—工单—采购—交货主链及 DWD 构建"
    if module == "S5":
        if any(token in text for token in ("iqc", "检验", "库存")):
            return "A+B", "API_VERIFIED", "A04 IQC 异常处置已走正式 API；库存状态已准备"
        return "A+B", "DATA_READY", "S5 标准事实已准备，操作型步骤仍需逐 TC 执行"
    if module == "S6":
        if any(token in text for token in ("ipqc", "过程检验")):
            return "A+B", "API_VERIFIED", "A05 IPQC 异常与返工处置已走正式 API"
        return "A+B+DEMO", "DATA_READY", "工单报工、工艺路线和 Demo 趋势事实已准备"
    if module == "S7":
        if any(token in text for token in ("fqc", "成品检验")):
            return "A+B", "API_VERIFIED", "A04 FQC 异常与返工处置已走正式 API"
        return "A+B+DEMO", "DATA_READY", "成品入库及 Demo 质量趋势已准备"
    if module == "S8":
        return "A+DEMO", "DATA_READY", "Demo 已有 39 异常、65 时间线、158 检测记录"
    if module == "S9":
        return "DEMO", "DATA_READY", "169 项 KPI 基线及日值可供九宫格/诊断/ChatBI 查询"
    return "NON-DATA", "PENDING_EXECUTION", "需按接口、权限、性能或交付专项执行"


def database_snapshot(cur: pymysql.cursors.DictCursor) -> dict[str, object]:
    result: dict[str, object] = {"tenants": {}, "jobs": [], "invalid_rows": 0}
    for code, tenant_id in TENANTS.items():
        metrics: dict[str, int] = {}
        queries = {
            "sales_orders": "SELECT COUNT(*) n FROM crm_seorder WHERE tenant_id=%s",
            "sales_order_lines": "SELECT COUNT(*) n FROM crm_seorderentry WHERE tenant_id=%s",
            "work_orders": "SELECT COUNT(*) n FROM WorkOrdMaster WHERE tenant_id=%s",
            "purchase_requests": "SELECT COUNT(*) n FROM srm_pr_main WHERE tenant_id=%s",
            "purchase_orders": "SELECT COUNT(*) n FROM PurOrdMaster WHERE tenant_id=%s",
            "s6_reports": "SELECT COUNT(*) n FROM mdp_std_s6_report WHERE tenant_id=%s",
            "fqc_results": "SELECT COUNT(*) n FROM mdp_std_fqc_result WHERE tenant_id=%s",
            "kpi_l1": "SELECT COUNT(*) n FROM ado_s9_kpi_value_l1_day WHERE tenant_id=%s AND is_deleted=0",
            "kpi_l2": "SELECT COUNT(*) n FROM ado_s9_kpi_value_l2_day WHERE tenant_id=%s AND is_deleted=0",
            "kpi_l3": "SELECT COUNT(*) n FROM ado_s9_kpi_value_l3_day WHERE tenant_id=%s AND is_deleted=0",
            "kpi_l4": "SELECT COUNT(*) n FROM ado_s9_kpi_value_l4_day WHERE tenant_id=%s AND is_deleted=0",
            "kpi_master": "SELECT COUNT(*) n FROM ado_smart_ops_kpi_master WHERE TenantId=%s",
        }
        for name, query in queries.items():
            cur.execute(query, (tenant_id,))
            metrics[name] = int(cur.fetchone()["n"])
        cur.execute(
            """
            SELECT COUNT(*) n FROM ado_s8_exception
            WHERE tenant_id=%s
            """,
            (tenant_id,),
        )
        metrics["s8_exceptions"] = int(cur.fetchone()["n"])
        result["tenants"][code] = metrics

    cur.execute(
        """
        WITH latest AS (
          SELECT j.*,
            ROW_NUMBER() OVER(
              PARTITION BY tenant_id,module_code
              ORDER BY submitted_at DESC,id DESC
            ) rn
          FROM ado_module_dashboard_rebuild_job j
          WHERE tenant_id IN (%s,%s,%s)
        )
        SELECT CAST(tenant_id AS CHAR) tenant_id,module_code,status,current_stage,
               progress_percent,batch_id,submitted_at,finished_at,error_message
        FROM latest WHERE rn=1 ORDER BY tenant_id,module_code
        """,
        tuple(TENANTS.values()),
    )
    result["jobs"] = list(cur.fetchall())

    invalid_tables = (
        "dwd_material_readiness",
        "dwd_material_shortage",
        "dwd_order_schedule_trans",
        "dwd_supplier_delivery",
        "dwd_supplier_risk",
        "dwd_supply_demand",
        "ado_s9_kpi_value_l1_day",
        "ado_s9_kpi_value_l2_day",
        "ado_s9_kpi_value_l3_day",
        "ado_s9_kpi_value_l4_day",
    )
    invalid_counts = {}
    for table in invalid_tables:
        cur.execute(
            f"""
            SELECT COUNT(*) n FROM `{table}`
            WHERE tenant_id IS NULL OR tenant_id IN (0,1,1300000000001)
            """
        )
        invalid_counts[table] = int(cur.fetchone()["n"])
    result["invalid_by_table"] = invalid_counts
    result["invalid_rows"] = sum(invalid_counts.values())
    return result


def style_sheet(worksheet) -> None:
    worksheet.freeze_panes = "A2"
    worksheet.auto_filter.ref = worksheet.dimensions
    header_fill = PatternFill("solid", fgColor="1F4E78")
    for cell in worksheet[1]:
        cell.fill = header_fill
        cell.font = Font(color="FFFFFF", bold=True)
        cell.alignment = Alignment(horizontal="center", vertical="center")
    for column in worksheet.columns:
        letter = get_column_letter(column[0].column)
        width = min(max(len(str(cell.value or "")) for cell in column) + 2, 55)
        worksheet.column_dimensions[letter].width = max(width, 12)
    for row in worksheet.iter_rows(min_row=2):
        for cell in row:
            cell.alignment = Alignment(vertical="top", wrap_text=True)


def append_rows(worksheet, headers: list[str], rows: list[list[object]]) -> None:
    worksheet.append(headers)
    for row in rows:
        worksheet.append(row)
    style_sheet(worksheet)


def func_rows(source_rows: list[dict[str, object]], level: str) -> list[list[object]]:
    output = []
    for source in source_rows:
        func = str(find_value(source, "功能编号", "系统功能编号") or "")
        name = str(find_value(source, "功能名称", "系统功能名称") or "")
        ac = str(find_value(source, "验收编号") or "")
        req = str(find_value(source, "关联req") or "")
        tc = str(find_value(source, "关联tc") or "")
        path = str(find_value(source, "系统路径", "入口") or "")
        module = module_of(func)
        data_set, readiness, evidence = coverage_for(module, name, path)
        execution = (
            "PENDING"
            if readiness in {"DATA_READY", "PENDING_EXECUTION"}
            else "PASSED"
        )
        output.append(
            [
                level,
                module,
                func,
                name,
                ac,
                req,
                tc,
                path,
                data_set,
                readiness,
                execution,
                evidence,
                "",
            ]
        )
    return output


def main() -> None:
    l1_funcs = rows_from_sheet(L1_FILE, 2)
    l1_technical = rows_from_sheet(L1_FILE, 1)
    l2_funcs = rows_from_sheet(L2_FILE, 0)

    conn = connect()
    try:
        with conn.cursor() as cur:
            snapshot = database_snapshot(cur)
    finally:
        conn.close()

    workbook = Workbook()
    workbook.remove(workbook.active)
    summary = workbook.create_sheet("汇总")
    l1_rows = func_rows(l1_funcs, "L1")
    l2_rows = func_rows(l2_funcs, "L2")
    counts = Counter(row[9] for row in l1_rows + l2_rows)
    all_jobs_success = all(job["status"] == "SUCCESS" for job in snapshot["jobs"])
    summary_rows = [
        ["生成时间", datetime.now().isoformat(timespec="seconds")],
        ["L1 FUNC/AC", len(l1_rows)],
        ["L2 FUNC/AC", len(l2_rows)],
        ["技术规范项", len(l1_technical)],
        ["API_VERIFIED", counts["API_VERIFIED"]],
        ["DATA_READY", counts["DATA_READY"]],
        ["PENDING_EXECUTION", counts["PENDING_EXECUTION"]],
        ["非法租户 DWD/KPI 行", snapshot["invalid_rows"]],
        ["S1-S7 最新作业全部成功", "PASS" if all_jobs_success else "FAIL"],
        [
            "当前总判定",
            "待人工/外部系统验收"
            if any(row[10] == "PENDING" for row in l1_rows + l2_rows)
            else "PASSED",
        ],
    ]
    append_rows(summary, ["项目", "结果"], summary_rows)

    headers = [
        "层级",
        "模块",
        "FUNC",
        "功能名称",
        "AC",
        "REQ",
        "TC",
        "路径/入口",
        "覆盖数据套",
        "数据/API就绪状态",
        "执行结论",
        "现有证据",
        "人工证据链接/备注",
    ]
    append_rows(workbook.create_sheet("L1-FUNC覆盖"), headers, l1_rows)
    append_rows(workbook.create_sheet("L2-FUNC覆盖"), headers, l2_rows)

    technical_rows = []
    for source in l1_technical:
        technical_rows.append(
            [
                find_value(source, "验证编号"),
                find_value(source, "技术规范章节"),
                find_value(source, "规范要求", "指标"),
                find_value(source, "验收方法"),
                find_value(source, "验收通过标准"),
                "PENDING",
                "",
            ]
        )
    append_rows(
        workbook.create_sheet("L1-技术规范113项"),
        ["验证编号", "章节", "要求/指标", "验收方法", "通过标准", "执行结论", "证据"],
        technical_rows,
    )

    inventory_rows = []
    for code, metrics in snapshot["tenants"].items():
        for metric, value in metrics.items():
            inventory_rows.append([code, str(TENANTS[code]), metric, value])
    append_rows(
        workbook.create_sheet("三租户数据快照"),
        ["数据套", "tenant_id", "指标", "行数"],
        inventory_rows,
    )

    job_rows = [
        [
            job["tenant_id"],
            job["module_code"],
            job["status"],
            job["current_stage"],
            job["progress_percent"],
            job["batch_id"],
            job["submitted_at"],
            job["finished_at"],
            job["error_message"],
        ]
        for job in snapshot["jobs"]
    ]
    append_rows(
        workbook.create_sheet("S1-S7构建终态"),
        [
            "tenant_id",
            "模块",
            "状态",
            "阶段",
            "进度",
            "批次",
            "提交时间",
            "完成时间",
            "错误",
        ],
        job_rows,
    )
    workbook.save(OUTPUT)

    recon = Workbook()
    recon.remove(recon.active)
    append_rows(recon.create_sheet("三租户数据快照"), ["数据套", "tenant_id", "指标", "行数"], inventory_rows)
    append_rows(
        recon.create_sheet("非法租户门禁"),
        ["表", "非法行数", "结论"],
        [
            [table, count, "PASS" if count == 0 else "FAIL"]
            for table, count in snapshot["invalid_by_table"].items()
        ],
    )
    append_rows(
        recon.create_sheet("构建终态"),
        [
            "tenant_id",
            "模块",
            "状态",
            "阶段",
            "进度",
            "批次",
            "提交时间",
            "完成时间",
            "错误",
        ],
        job_rows,
    )
    recon.save(RECON_OUTPUT)

    evidence = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "l1_func_count": len(l1_rows),
        "l2_func_count": len(l2_rows),
        "technical_count": len(l1_technical),
        "readiness_counts": dict(counts),
        "snapshot": snapshot,
        "outputs": [str(OUTPUT), str(RECON_OUTPUT)],
    }
    Path(__file__).with_name("WP5-WP6-acceptance-evidence.json").write_text(
        json.dumps(evidence, ensure_ascii=False, default=str, indent=2),
        encoding="utf-8",
    )
    print(json.dumps({key: evidence[key] for key in evidence if key != "snapshot"}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()