#!/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()