AiDOPWarehouse/doc/_verify_old_api_equivalence_e2e.py

#!/usr/bin/env python3
"""E2E-01: 旧 Business API 等价迁移统一验收编排。

1. 扫描 Web/ 无旧 api/business 直连
2. 串联运行 S1 / S2-S4 / S1-S4 full 子脚本
3. 校验 aidop_action_run_log 关键 action_code 存在
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DOC = ROOT / "doc"
WEB = ROOT / "Web"
BASE = os.environ.get("AIDOP_API_BASE", "http://127.0.0.1:5005")
TENANT = 797403760988229

REQUIRED_ACTION_CODES = [
    "S1_ORDER_REVIEW",
    "S1_DELIVERY_CONFIRM",
    "S1_ORDER_REFRESH_PLAN",
    "S2_SCHEDULE_GENERATE",
    "S2_PRIORITY_RECHECK",
    "S2_KITTING_CHECK_BATCH",
    "S2_KITTING_CHECK_SINGLE",
    "S2_WORK_ORDER_RELEASE",
    "S4_MRP_GENERATE",
    "S3_DELIVERY_GENERATE",
    "S4_PROCUREMENT_PIPELINE",
]

CHILD_SCRIPTS = [
    ("order_review", DOC / "_verify_order_review_e2e.py"),
    ("s2_s4_chain", DOC / "_verify_s2_s4_chain_e2e.py"),
    ("s1_s4_full", DOC / "_verify_s1_s4_full_e2e.py"),
    ("s3_delivery_18step", DOC / "_verify_s3_delivery_18step_e2e.py"),
]
PREP_SCRIPT = DOC / "_prep_close_evidence.py"

OLD_URL_PATTERNS = [
    "123.60.180.165:8087",
    "/api/business/",
]


def scan_old_frontend_urls() -> list[str]:
    hits: list[str] = []
    if not WEB.exists():
        return hits
    skip_dirs = {"node_modules", "dist", ".git", ".pnpm"}
    for path in WEB.rglob("*"):
        if not path.is_file():
            continue
        if any(part in skip_dirs for part in path.parts):
            continue
        if path.suffix not in {".ts", ".vue", ".js", ".tsx", ".jsx"}:
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        for pat in OLD_URL_PATTERNS:
            if pat in text:
                hits.append(f"{path.relative_to(ROOT)}: {pat}")
    return hits


def http_json(method: str, path: str, body=None, token: str | None = None):
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(f"{BASE}{path}", data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            raw = resp.read().decode("utf-8")
            try:
                return resp.status, json.loads(raw)
            except json.JSONDecodeError:
                return resp.status, {"raw": raw[:2000]}
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        try:
            return e.code, json.loads(raw)
        except json.JSONDecodeError:
            return e.code, {"raw": raw[:2000]}


def login() -> str:
    node_script = r"""
const { sm2 } = require('sm-crypto-v2');
const PK = '0484C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312';
console.log(JSON.stringify({ account: 'AIDOPDemo', password: sm2.doEncrypt('1234567890dop', PK, 1), tenantId: '797403760988229' }));
"""
    proc = subprocess.run(
        ["node", "-e", node_script],
        cwd=str(WEB),
        capture_output=True,
        text=True,
        timeout=30,
    )
    if proc.returncode != 0:
        raise RuntimeError(proc.stderr or proc.stdout)
    _, body = http_json("POST", "/api/sysAuth/login", json.loads(proc.stdout.strip()))
    token = (body.get("result") or {}).get("accessToken") or body.get("accessToken")
    if not token:
        raise RuntimeError(f"login failed: {body}")
    return token


def parse_child_report(stdout: str) -> dict | None:
    text = (stdout or "").strip()
    if not text:
        return None
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        start = text.find("{")
        end = text.rfind("}")
        if start >= 0 and end > start:
            try:
                return json.loads(text[start : end + 1])
            except json.JSONDecodeError:
                return None
    return None


def scan_frontend_wiring() -> list[dict]:
    """CLOSE-E2E：检查关键能力是否已挂到前端页面（彻底打通之一）。"""
    checks: list[dict] = []

    def file_contains(rel: str, *needles: str) -> bool:
        path = ROOT / rel
        if not path.exists():
            return False
        text = path.read_text(encoding="utf-8", errors="ignore")
        return all(n in text for n in needles)

    def files_contain(rels: list[str], *needles: str) -> bool:
        text = ""
        for rel in rels:
            path = ROOT / rel
            if path.exists():
                text += path.read_text(encoding="utf-8", errors="ignore")
        return all(n in text for n in needles)

    wiring = [
        ("S1_order_review_page", "Web/src/views/aidop/business/orderList.vue", ("reviewSeOrder", "confirmSeOrderDelivery"), False),
        ("S2_schedule_page", "Web/src/views/aidop/production/workOrderSchedulingList.vue", ("productionSchedule",), False),
        (
            "S2_dispatch_page",
            [
                "Web/src/views/aidop/business/workOrderDispatchList.vue",
                "Web/src/views/aidop/business/workOrderDispatchForm.vue",
            ],
            ("onKitCheck", "releaseWorkOrder"),
            True,
        ),
        ("S2_single_kitting_page", "Web/src/views/aidop/business/workOrderDispatchList.vue", ("kittingCheckSingle",), False),
        ("S3_delivery_generate_page", "Web/src/views/aidop/s3/supply/deliveryScheduleList.vue", ("generateDeliverySchedule",), False),
        ("S3_procurement_pipeline_page", "Web/src/views/aidop/s3/supply/deliveryScheduleList.vue", ("executeProcurementPipeline",), False),
        ("S4_read_path_page", "Web/src/views/aidop/data-platform/actionRunLogs.vue", ("fetchS4ReadPathConsistency",), False),
    ]
    for item in wiring:
        name, rel, needles, multi = item
        if multi:
            ok = files_contain(rel, *needles)
            file_label = " + ".join(rel)
        else:
            ok = file_contains(rel, *needles)
            file_label = rel
        checks.append({"name": name, "file": file_label, "ok": ok, "needles": list(needles)})
    return checks


def build_module_status(child_reports: dict[str, dict | None], wiring: list[dict]) -> dict:
    """汇总 S1-S4 打通状态：passed / 待对账 / 待补齐 / 需决策。"""
    modules: dict[str, dict] = {
        "S1": {"status": "待对账", "evidence": [], "gaps": []},
        "S2": {"status": "待对账", "evidence": [], "gaps": []},
        "S3": {"status": "待对账", "evidence": [], "gaps": []},
        "S4": {"status": "待对账", "evidence": [], "gaps": []},
    }

    def child_passed(name: str) -> bool:
        r = child_reports.get(name)
        return isinstance(r, dict) and bool(r.get("passed"))

    if child_passed("order_review"):
        modules["S1"]["evidence"].append("order_review_e2e")
    if child_passed("s1_s4_full"):
        modules["S1"]["evidence"].append("s1_s4_full_e2e")
        modules["S4"]["evidence"].append("s1_s4_full_e2e")
    if child_passed("s2_s4_chain"):
        modules["S2"]["evidence"].append("s2_s4_chain_e2e")
        modules["S3"]["evidence"].append("s2_s4_chain_e2e")
        modules["S4"]["evidence"].append("s2_s4_chain_e2e")
    if child_passed("s3_delivery_18step"):
        modules["S3"]["evidence"].append("s3_delivery_18step_e2e")

    wiring_map = {w["name"]: w["ok"] for w in wiring}
    if not wiring_map.get("S1_order_review_page"):
        modules["S1"]["gaps"].append("订单评审页未接线")
    if not wiring_map.get("S2_schedule_page") or not wiring_map.get("S2_dispatch_page"):
        modules["S2"]["gaps"].append("排程/下达页未接线")
    if not wiring_map.get("S2_single_kitting_page"):
        modules["S2"]["gaps"].append("单工单齐套无页面入口")
    if not wiring_map.get("S3_delivery_generate_page"):
        modules["S3"]["gaps"].append("交货计划生成无页面入口")
    if not wiring_map.get("S3_procurement_pipeline_page"):
        modules["S3"]["gaps"].append("采购闭环无页面入口")
    if not wiring_map.get("S4_read_path_page"):
        modules["S4"]["gaps"].append("S4读路径一致性无页面入口")

    modules["S2"]["gaps"].append("库存占用：需决策")
    modules["S3"]["gaps"].append("真实SAP/QAD：需决策")
    modules["S1"]["gaps"].append("替代料口径：需决策")

    decision_only = lambda gaps: not gaps or all("需决策" in g for g in gaps)
    required_evidence = {
        "S1": {"order_review_e2e", "s1_s4_full_e2e"},
        "S2": {"s2_s4_chain_e2e"},
        "S3": {"s3_delivery_18step_e2e"},
        "S4": {"s1_s4_full_e2e", "s2_s4_chain_e2e"},
    }

    for key, mod in modules.items():
        if mod["gaps"] and any("未接线" in g or "无页面" in g for g in mod["gaps"]):
            mod["status"] = "待补齐"
        elif not mod["evidence"]:
            mod["status"] = "待补齐"
        elif set(mod["evidence"]) >= required_evidence[key] and decision_only(mod["gaps"]):
            mod["status"] = "已打通"
        elif mod["evidence"]:
            mod["status"] = "待对账"
    return modules


def run_child(name: str, script: Path) -> dict:
    if not script.exists():
        return {"name": name, "ok": False, "error": f"missing script: {script}"}
    proc = subprocess.run(
        [sys.executable, str(script)],
        cwd=str(ROOT),
        capture_output=True,
        text=True,
        timeout=300,
    )
    parsed = parse_child_report(proc.stdout or "")
    result = {
        "name": name,
        "ok": proc.returncode == 0,
        "exit_code": proc.returncode,
        "parsed": parsed,
        "stdout_tail": (proc.stdout or "")[-4000:],
        "stderr_tail": (proc.stderr or "")[-2000:],
    }
    if isinstance(parsed, dict):
        result["child_passed"] = parsed.get("passed")
        result["child_checks"] = parsed.get("checks")
    return result


def check_action_logs(token: str) -> dict:
    checks = []
    for code in REQUIRED_ACTION_CODES:
        st, body = http_json(
            "GET",
            f"/api/Aidop/action-run-log/list?actionCode={code}&page=1&pageSize=1",
            token=token,
        )
        total = 0
        if isinstance(body, dict):
            total = body.get("total") or 0
        checks.append({"action_code": code, "http": st, "total": total, "ok": st == 200 and total > 0})
    return {"checks": checks, "ok": all(c["ok"] for c in checks)}


def main():
    report = {
        "matrix_doc": "doc/plan/旧Business API等价验收矩阵.md",
        "steps": [],
        "checks": [],
    }

    url_hits = scan_old_frontend_urls()
    report["steps"].append({"step": "old_url_scan", "hits": url_hits, "ok": len(url_hits) == 0})
    report["checks"].append({"name": "no_old_frontend_url", "ok": len(url_hits) == 0, "detail": url_hits})

    prep_result = run_child("prep_close_evidence", PREP_SCRIPT)
    report["steps"].append({"step": "prep_close_evidence", **prep_result})
    report["checks"].append({"name": "prep_close_evidence", "ok": prep_result.get("ok", False)})

    child_results = []
    child_reports: dict[str, dict | None] = {}
    for name, script in CHILD_SCRIPTS:
        result = run_child(name, script)
        child_results.append(result)
        child_reports[name] = result.get("parsed") if isinstance(result.get("parsed"), dict) else None
        report["steps"].append({"step": f"child_{name}", **result})
        report["checks"].append({"name": f"child_{name}", "ok": result.get("ok", False)})

    wiring = scan_frontend_wiring()
    report["steps"].append({"step": "frontend_wiring", "checks": wiring, "ok": all(c["ok"] for c in wiring)})
    report["checks"].append({
        "name": "frontend_wiring",
        "ok": all(c["ok"] for c in wiring),
        "detail": [c for c in wiring if not c["ok"]],
    })
    report["module_status"] = build_module_status(child_reports, wiring)
    report["gaps"] = {
        k: v.get("gaps", [])
        for k, v in report["module_status"].items()
    }

    try:
        token = login()
        log_check = check_action_logs(token)
        report["steps"].append({"step": "action_log_codes", **log_check})
        report["checks"].append({
            "name": "action_log_codes",
            "ok": log_check["ok"],
            "detail": [c for c in log_check["checks"] if not c["ok"]],
        })
    except Exception as ex:
        report["steps"].append({"step": "action_log_codes", "ok": False, "error": str(ex)})
        report["checks"].append({"name": "action_log_codes", "ok": False, "detail": str(ex)})

    report["passed"] = all(c["ok"] for c in report["checks"])

    out = DOC / "_verify_old_api_equivalence_e2e_result.json"
    out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(report, ensure_ascii=False, indent=2))
    sys.exit(0 if report["passed"] else 5)


if __name__ == "__main__":
    main()