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

#!/usr/bin/env python3
"""WP-SD7 live API loop: diagnose, create, approve, execute, verify, close.

Passwords come from the environment and are never written to evidence.
Does not hand-edit plan status in the database.
"""

from __future__ import annotations

import argparse
import json
import os
import urllib.error
import urllib.request
from datetime import datetime, timedelta
from pathlib import Path

from gmssl import sm2


ROOT = Path(__file__).resolve().parents[4]
EVIDENCE = ROOT / "doc" / "plan" / "UAT留证" / "2026-08-17-智慧诊断正式落地"
BASE = os.environ.get("AIDOP_API_BASE", "http://127.0.0.1:5005")
SM2_PK = "84C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312"
BATCH = datetime.now().strftime("%Y%m%d")
TENANTS = {
    "A": {"account": "UATAdminA", "tenant_id": "838257186181189", "user_id": "838257187360837"},
    "B": {"account": "UATAdminB", "tenant_id": "838257212780613", "user_id": "838257213620293"},
    "DEMO": {"account": "UATDemoAdmin", "tenant_id": "838257237606469", "user_id": "838257238302789"},
}


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


def encrypt(value: str) -> str:
    return sm2.CryptSM2(public_key=SM2_PK, private_key=None, mode=1).encrypt(value.encode()).hex()


def unwrap(body):
    if isinstance(body, dict) and "result" in body:
        return body.get("result")
    return body


def as_dict(value, fallback=None):
    return value if isinstance(value, dict) else (fallback or {})


def login(account: str, tenant_id: str, password: str, already_encrypted: bool = False) -> str:
    cipher = password if already_encrypted else encrypt(password)
    status, body = request(
        "POST",
        "/api/sysAuth/login",
        {"account": account, "password": cipher, "tenantId": tenant_id},
    )
    token = (unwrap(body) or {}).get("accessToken") if isinstance(unwrap(body), dict) else None
    if status != 200 or not token:
        raise SystemExit(f"login failed for {account}: HTTP {status} {body}")
    return token


def diagnose(token: str, module: str, metric_code: str | None = None) -> dict:
    path = f"/api/AidopKanban/smart-diagnosis/{module}"
    if metric_code:
        path += f"?metricCode={metric_code}"
    status, body = request("GET", path, token=token)
    return {"http": status, "body": unwrap(body) or {}}


def dashboard(token: str, module: str) -> dict:
    status, body = request("GET", f"/api/AidopKanban/dashboard-page/{module}", token=token)
    return {"http": status, "body": unwrap(body) or {}}


def create_plan(token: str, diagnosis: dict, module: str) -> dict:
    root = diagnosis.get("root") or {}
    evidence = diagnosis.get("evidence") or {}
    items = evidence.get("items") or []
    if not items:
        return {"skipped": True, "reason": "no_evidence_items"}
    first = items[0]
    payload = {
        "moduleCode": module,
        "metricCode": diagnosis.get("metricCode") or root.get("metricCode"),
        "problemLevel": 1,
        "problemMetricCode": root.get("metricCode"),
        "problemName": root.get("metricName") or f"{module} 诊断问题",
        "problemDept": root.get("department"),
        "problemSeverity": "yellow",
        "isCrossDept": module == "S9",
        "targetValue": str(root.get("targetValue") or ""),
        "actualValue": str(root.get("currentValue") or ""),
        "gapLabel": root.get("gapLabel"),
        "rootCause": (
            f"[WP-SD7/{BATCH}] 依据事实 {first.get('objectType')} {first.get('objectCode')}："
            f"{first.get('title')}。人工填写，非自动建议。"
        ),
        "actionItems": [
            {
                "content": f"核对事实 {first.get('objectCode')} 并关闭偏差",
                "dueDate": (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d"),
                "status": "pending",
            },
            {
                "content": "复盘看板筛选与无筛选口径是否一致",
                "dueDate": (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d"),
                "status": "pending",
            },
        ],
        "dueDate": (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%dT00:00:00"),
    }
    status, body = request("POST", "/api/AdoSmartOpsImprovementPlan/createFromDiagnosis", payload, token)
    return {"http": status, "body": unwrap(body) or {}}


def approve_until_done(token: str, plan_id) -> list[dict]:
    steps = []
    for _ in range(6):
        status, body = request(
            "POST",
            "/api/flowTask/myPendingPage",
            {"page": 1, "pageSize": 50, "bizType": "SMART_OPS_IMPROVEMENT"},
            token,
        )
        page = unwrap(body) or {}
        items = page.get("items") or page.get("Items") or []
        match = None
        for item in items:
            biz_id = item.get("bizId") or item.get("BizId")
            if str(biz_id) == str(plan_id):
                match = item
                break
        if not match:
            steps.append({"done": True, "pending": len(items)})
            break
        task_id = match.get("id") or match.get("Id") or match.get("taskId")
        st, app = request("POST", "/api/flowTask/approve", {"taskId": task_id, "comment": "WP-SD7 正式审批通过"}, token)
        steps.append({"http": st, "taskId": str(task_id) if task_id else None, "body": unwrap(app)})
    return steps


def complete_actions(token: str, plan: dict) -> list[dict]:
    results = []
    actions = plan.get("actionItems") or plan.get("ActionItems") or []
    plan_id = plan.get("id") or plan.get("Id")
    today = datetime.now().strftime("%Y-%m-%d")
    for action in actions:
        action_id = action.get("id") or action.get("Id")
        steps = []
        current = (action.get("status") or action.get("Status") or "pending").lower()
        wanted = []
        if current == "pending":
            wanted.append("in_progress")
        if current in ("pending", "in_progress", "doing"):
            wanted.append("completed")
        for status in wanted:
            st, body = request(
                "POST",
                "/api/AdoSmartOpsImprovementPlan/updateActionItem",
                {
                    "id": plan_id,
                    "actionId": action_id,
                    "status": status,
                    "completedAt": today if status == "completed" else None,
                    "proofRemark": "WP-SD7 按事实核对后关闭，非自动建议",
                },
                token,
            )
            msg = body.get("message") if isinstance(body, dict) else None
            if not msg and isinstance(unwrap(body), str):
                msg = unwrap(body)
            steps.append({"http": st, "status": status, "message": msg})
            nxt = unwrap(body)
            if isinstance(nxt, dict):
                plan = nxt
        results.append({"actionId": str(action_id) if action_id else None, "steps": steps})
    return results, plan


def run_module(token: str, module: str) -> dict:
    board = dashboard(token, module)
    global_diag = diagnose(token, module)
    body = global_diag.get("body") or {}
    metric = body.get("metricCode")
    specified = diagnose(token, module, metric) if metric else None
    evidence = body.get("evidence") or {}
    created = create_plan(token, body, module) if evidence.get("items") else {
        "skipped": True,
        "reason": "no_evidence_or_diagnosis",
    }
    plan = as_dict(created.get("body"))
    plan_id = plan.get("id") or plan.get("Id")
    board_metrics = {
        (m.get("metricCode") or m.get("MetricCode")): (m.get("currentValue") or m.get("CurrentValue") or m.get("metricValue"))
        for m in ((board.get("body") or {}).get("metrics") or [])
        if (m.get("metricLevel") or m.get("MetricLevel") or m.get("level")) in (1, "1", None)
    }
    submit = approve = start = actions = verify = close = None
    if plan_id:
        st, sub = request("POST", "/api/AdoSmartOpsImprovementPlan/submitApproval", {"id": plan_id}, token)
        submit_msg = sub.get("message") if isinstance(sub, dict) else None
        submit = {"http": st, "message": submit_msg, "body": unwrap(sub)}
        submit_failed = st >= 400 or (isinstance(submit_msg, str) and ("失败" in submit_msg or "拒绝" in submit_msg))
        if submit_failed:
            return {
                "module": module,
                "dashboardHttp": board.get("http"),
                "global": {
                    "http": global_diag["http"],
                    "selectionMode": body.get("selectionMode"),
                    "metricCode": metric,
                    "hasChildren": body.get("hasChildren"),
                    "rootStatus": (body.get("root") or {}).get("status"),
                    "evidenceScope": evidence.get("scope"),
                    "evidenceTotal": evidence.get("total"),
                    "firstFact": ((evidence.get("items") or [{}])[0] or {}).get("objectCode"),
                },
                "specified_same_as_current": None
                if not specified
                else (specified.get("body") or {}).get("metricCode") == metric,
                "kanban_root_present": True if module == "S8" else (metric in board_metrics if metric else False),
                "create": {
                    "http": created.get("http"),
                    "skipped": created.get("skipped"),
                    "reason": created.get("reason"),
                    "planNo": plan.get("planNo") or plan.get("PlanNo"),
                    "planId": str(plan_id) if plan_id else None,
                },
                "submitApproval": submit,
                "loopClosed": False,
            }
        approve = approve_until_done(token, plan_id)
        st, started = request("POST", "/api/AdoSmartOpsImprovementPlan/startExecution", {"id": plan_id}, token)
        started_body = as_dict(unwrap(started), plan)
        start = {"http": st, "status": started_body.get("status") or started_body.get("Status")}
        actions, plan = complete_actions(token, started_body)
        st, ver = request(
            "POST",
            "/api/AdoSmartOpsImprovementPlan/submitVerification",
            {"id": plan_id, "verifyResult": "INSUFFICIENT_DATA", "verifyRemark": "样本不足时与自动评价一致"},
            token,
        )
        verify = {
            "http": st,
            "auto": (unwrap(ver) or {}).get("autoVerifyResult") or (unwrap(ver) or {}).get("AutoVerifyResult"),
            "status": (unwrap(ver) or {}).get("status") or (unwrap(ver) or {}).get("Status"),
        }
        st, closed = request("POST", "/api/AdoSmartOpsImprovementPlan/close", {"id": plan_id}, token)
        close = {"http": st, "status": (unwrap(closed) or {}).get("status") or (unwrap(closed) or {}).get("Status")}
    return {
        "module": module,
        "dashboardHttp": board.get("http"),
        "global": {
            "http": global_diag["http"],
            "selectionMode": body.get("selectionMode"),
            "metricCode": metric,
            "hasChildren": body.get("hasChildren"),
            "rootStatus": (body.get("root") or {}).get("status"),
            "evidenceScope": evidence.get("scope"),
            "evidenceTotal": evidence.get("total"),
            "firstFact": ((evidence.get("items") or [{}])[0] or {}).get("objectCode"),
        },
        "specified_same_as_current": None
        if not specified
        else (specified.get("body") or {}).get("metricCode") == metric,
        "kanban_root_present": True if module == "S8" else (metric in board_metrics if metric else False),
        "create": {
            "http": created.get("http"),
            "skipped": created.get("skipped"),
            "reason": created.get("reason"),
            "planNo": plan.get("planNo") or plan.get("PlanNo"),
            "planId": str(plan_id) if plan_id else None,
        },
        "submitApproval": {"http": (submit or {}).get("http"), "message": (submit or {}).get("message")} if submit else None,
        "approveSteps": [{"http": x.get("http"), "done": x.get("done")} for x in (approve or [])],
        "startExecution": start,
        "actionsCompleted": actions,
        "verify": verify,
        "close": close,
        "loopClosed": (close or {}).get("status") == "closed",
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--tenant", choices=TENANTS.keys(), default="A")
    parser.add_argument("--modules", default="S1,S2,S3,S4,S5,S6,S7,S9")
    parser.add_argument("--password", default="")
    parser.add_argument("--out", default="", help="Evidence filename under the UAT folder")
    args = parser.parse_args()
    password = args.password or os.environ.get("AIDOP_UAT_PASSWORD", "").strip()
    if not password:
        raise SystemExit("AIDOP_UAT_PASSWORD or --password is required")

    tenant = TENANTS[args.tenant]
    token = login(tenant["account"], tenant["tenant_id"], password)
    report = {
        "work_package": "WP-SD7-API",
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "tenant": args.tenant,
        "account": tenant["account"],
        "auto_passed": False,
        "modules": [],
    }
    for module in [x.strip().upper() for x in args.modules.split(",") if x.strip()]:
        report["modules"].append(run_module(token, module))
    report["closed_count"] = sum(1 for x in report["modules"] if x.get("loopClosed"))
    report["blocked_count"] = sum(1 for x in report["modules"] if not x.get("loopClosed"))
    out = EVIDENCE / (args.out or f"07-api-loop-{args.tenant}.json")
    out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(out)
    print(json.dumps(
        {
            "tenant": args.tenant,
            "closed": report["closed_count"],
            "blocked": report["blocked_count"],
            "modules": [
                f"{x['module']}:{x['global']['evidenceScope']}:closed={x.get('loopClosed')}"
                for x in report["modules"]
            ],
        },
        ensure_ascii=False,
    ))


if __name__ == "__main__":
    main()