AiDOPWarehouse/doc/_verify_s1_s4_business_chain_e2e.py

#!/usr/bin/env python3
"""S1-S4 business chain E2E.

S1 example path:
  新建订单 -> 订单评审(含齐套/资源检查) -> 生成物料需求 -> 生产排程 -> 同步物料需求
Then smoke S2/S3/S4 list + key actions.
"""
from __future__ import annotations

import json
import sys
import urllib.error
import urllib.request
from datetime import date, timedelta

import pymysql

BASE = "http://127.0.0.1:5005"
TENANT = 797403760988229
CONN = dict(
    host="123.60.180.165",
    port=3306,
    user="aidopremote",
    password="1234567890aiDOP#",
    database="aidopdev",
    charset="utf8mb4",
    connect_timeout=20,
    cursorclass=pymysql.cursors.DictCursor,
)


def http_json(method, path, body=None, token=None, timeout=180):
    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, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(f"{BASE}{path}", data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8")
            try:
                return resp.status, json.loads(raw) if raw else {}
            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]}
    except Exception as e:
        return 0, {"error": str(e)}


def sm2_encrypt_password(plain: str) -> str:
    """Match frontend sm-crypto-v2 mode=1 (C1C3C2) + Admin.NET GMUtil.SM2Decrypt."""
    from gmssl import sm2

    pk = "0484C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312"
    pub = pk[2:] if pk.startswith("04") else pk
    crypt = sm2.CryptSM2(public_key=pub, private_key=None, mode=1)
    cipher = crypt.encrypt(plain.encode("utf-8"))
    if isinstance(cipher, (bytes, bytearray)):
        hex_s = cipher.hex()
    else:
        hex_s = str(cipher)
    # gmssl usually returns without leading 04; backend accepts both
    return hex_s


def login():
    payload = {
        "account": "AIDOPDemo",
        "password": sm2_encrypt_password("1234567890dop"),
        "tenantId": str(TENANT),
    }
    st, body = http_json("POST", "/api/sysAuth/login", payload)
    token = (body.get("result") or {}).get("accessToken") or body.get("accessToken")
    if not token:
        raise RuntimeError(f"login failed http={st}: {body}")
    return token


def unwrap(body):
    if isinstance(body, dict) and body.get("code") == 200 and "result" in body:
        return body["result"]
    return body


def api_ok(status, body):
    if status != 200:
        return False
    if isinstance(body, dict) and body.get("error"):
        return False
    if body.get("code") in (200, None) and body.get("type") in ("success", None):
        # Furion unify: code=200 type=success; NonUnify may omit
        if body.get("code") == 200 and body.get("type") == "success":
            return True
        if "result" in body or "message" in body or "Message" in body:
            return True
        # NonUnify plain object
        if body.get("code") is None and body.get("type") is None and not body.get("error"):
            return True
    # NonUnify services often return plain DTO
    if status == 200 and isinstance(body, dict) and "code" not in body:
        return True
    return False


def step(report, name, status, body, extra=None):
    item = {
        "step": name,
        "http": status,
        "ok": api_ok(status, body) if status else False,
        "body": unwrap(body) if isinstance(body, dict) else body,
    }
    if extra:
        item.update(extra)
    if isinstance(body, dict) and body.get("message"):
        item["message"] = body.get("message")
    report["steps"].append(item)
    return item


def q(conn, sql, params=None):
    with conn.cursor() as cur:
        cur.execute(sql, params or ())
        return cur.fetchall()


def q1(conn, sql, params=None):
    rows = q(conn, sql, params)
    return rows[0] if rows else {}


def snapshot(conn, order_id=None, bill_no=None):
    snap = {}
    if order_id:
        snap["order"] = q1(
            conn,
            "SELECT Id, bill_no, status FROM crm_seorder WHERE Id=%s AND tenant_id=%s AND IsDeleted=0",
            (order_id, TENANT),
        )
        snap["entries"] = q(
            conn,
            """
            SELECT Id, entry_seq, item_number, qty, progress, plan_date, sys_capacity_date
            FROM crm_seorderentry
            WHERE seorder_id=%s AND tenant_id=%s AND IsDeleted=0 ORDER BY entry_seq
            """,
            (order_id, TENANT),
        )
        entry_ids = [r["Id"] for r in snap["entries"]]
        if entry_ids:
            fmt = ",".join(["%s"] * len(entry_ids))
            snap["work_orders"] = q(
                conn,
                f"""
                SELECT RecID, WorkOrd, ItemNum, Status, BusinessID
                FROM WorkOrdMaster
                WHERE tenant_id=%s AND BusinessID IN ({fmt})
                ORDER BY WorkOrd
                """,
                [TENANT, *entry_ids],
            )
            snap["examine"] = q(
                conn,
                f"""
                SELECT Id, sentry_id, morder_no, need_qty, kitting_times, create_time
                FROM b_examine_result
                WHERE tenant_id=%s AND sentry_id IN ({fmt}) AND IFNULL(IsDeleted,0)=0
                ORDER BY create_time DESC LIMIT 10
                """,
                [TENANT, *entry_ids],
            )
            snap["examine_lines"] = q1(
                conn,
                f"""
                SELECT COUNT(*) AS c FROM b_bom_child_examine b
                INNER JOIN b_examine_result e ON e.Id=b.examine_id
                WHERE e.tenant_id=%s AND e.sentry_id IN ({fmt})
                  AND IFNULL(e.IsDeleted,0)=0 AND IFNULL(b.is_use,1)=1
                """,
                [TENANT, *entry_ids],
            ).get("c", 0)
        else:
            snap["work_orders"] = []
            snap["examine"] = []
            snap["examine_lines"] = 0
    snap["demand"] = q1(
        conn, "SELECT COUNT(*) AS c FROM ic_demandschedule WHERE tenant_id=%s", (TENANT,)
    ).get("c", 0)
    snap["pr"] = q1(
        conn,
        "SELECT COUNT(*) AS c FROM srm_pr_main WHERE tenant_id=%s AND IFNULL(IsDeleted,0)=0",
        (TENANT,),
    ).get("c", 0)
    snap["po"] = q1(
        conn,
        "SELECT COUNT(*) AS c FROM srm_po_main WHERE tenant_id=%s AND IFNULL(IsDeleted,0)=0",
        (TENANT,),
    ).get("c", 0)
    snap["schedule_active"] = q1(
        conn,
        """
        SELECT COUNT(*) AS c FROM WorkOrdMaster
        WHERE tenant_id=%s AND Status IN ('p','r','w','s')
        """,
        (TENANT,),
    ).get("c", 0)
    if bill_no:
        snap["action_logs"] = q(
            conn,
            """
            SELECT id, action_code, biz_no, status, message, start_time
            FROM aidop_action_run_log
            WHERE tenant_id=%s AND (biz_no=%s OR biz_no LIKE %s OR action_code LIKE 'S1_%%' OR action_code LIKE 'S4_%%')
            ORDER BY start_time DESC LIMIT 8
            """,
            (TENANT, bill_no, f"%{bill_no}%"),
        )
    return snap


def main():
    report = {
        "title": "S1-S4 business chain E2E",
        "base": BASE,
        "tenant_id": TENANT,
        "steps": [],
        "checks": [],
    }

    st, body = http_json("GET", "/api/sysAuth/captcha")
    step(report, "backend_health", st, body, {"ok": st == 200})
    if st != 200:
        report["passed"] = False
        print(json.dumps(report, ensure_ascii=False, indent=2, default=str))
        sys.exit(1)

    token = login()
    report["login"] = "ok"
    domain = str(TENANT)
    plan_date = (date.today() + timedelta(days=30)).isoformat()
    bill_suffix = date.today().strftime("%Y%m%d")

    # ── S1-1 新建订单 ──
    save_body = {
        "billNo": None,
        "orderType": 2,
        "customNo": "10000000",
        "customName": "瑞贝德",
        "date": date.today().isoformat(),
        "urgent": 0,
        "country": "中国",
        "entries": [
            {
                "entrySeq": 1,
                "itemNumber": "91H0DE3",
                "itemName": "通用内镜直线切割吻合器弯转型钉匣-ENDORMC6035R-中国中文-0",
                "qty": 10,
                "planDate": plan_date,
                "unit": "PCS",
            },
            {
                "entrySeq": 2,
                "itemNumber": "1A0049H",
                "itemName": "一次性使用RCS系列端端吻合器-RCS25D-英文包装前",
                "qty": 5,
                "planDate": plan_date,
                "unit": "PCS",
            },
        ],
    }
    st, body = http_json("POST", "/api/Order/seorder/save", save_body, token)
    save_res = unwrap(body) if isinstance(body, dict) else {}
    order_id = (save_res or {}).get("id") or (save_res or {}).get("Id")
    step(report, "s1_create_order", st, body, {"order_id": order_id})
    if not order_id:
        report["passed"] = False
        print(json.dumps(report, ensure_ascii=False, indent=2, default=str))
        sys.exit(2)

    with pymysql.connect(**CONN) as conn:
        order_row = q1(
            conn,
            "SELECT Id, bill_no FROM crm_seorder WHERE Id=%s AND tenant_id=%s",
            (order_id, TENANT),
        )
        bill_no = order_row.get("bill_no")
        report["bill_no"] = bill_no
        report["order_id"] = order_id
        report["before_review"] = snapshot(conn, order_id, bill_no)

    # ── S1-2 订单评审（资源检查/齐套 + 预备工单） ──
    st, body = http_json("POST", "/api/Order/seorder/review", {"ids": [order_id]}, token, timeout=300)
    review_res = unwrap(body) if isinstance(body, dict) else {}
    step(report, "s1_order_review", st, body)

    with pymysql.connect(**CONN) as conn:
        report["after_review"] = snapshot(conn, order_id, bill_no)

    # ── S1-3 物料齐套看板刷新 + 生成物料需求 ──
    st, body = http_json(
        "POST",
        "/api/Supply/work-order-material-readiness/refresh",
        None,
        token,
        timeout=300,
    )
    step(report, "s3_kitting_refresh", st, body)

    st, body = http_json(
        "POST",
        f"/api/WorkOrder/material-requirement/generate?domain={domain}",
        None,
        token,
        timeout=300,
    )
    step(report, "s1_generate_material_requirement", st, body)

    # ── S1-4 生产排程 + 同步物料需求 ──
    st, body = http_json(
        "POST",
        f"/api/Production/scheduling/generate?domain={domain}",
        None,
        token,
        timeout=300,
    )
    step(report, "s2_production_schedule", st, body)

    st, body = http_json(
        "POST",
        f"/api/WorkOrder/material-requirement/generate?domain={domain}",
        None,
        token,
        timeout=300,
    )
    step(report, "s2_sync_material_requirement", st, body)

    # ── S1 交期确认（若评审后 progress 已到 2/3 可继续） ──
    st, body = http_json("POST", "/api/Order/seorder/confirm-delivery", {"ids": [order_id]}, token)
    step(report, "s1_confirm_delivery", st, body)

    # ── S2 列表冒烟 ──
    st, body = http_json(
        "GET",
        f"/api/Production/scheduling/list?page=1&pageSize=10&domain={domain}",
        token=token,
    )
    step(report, "s2_scheduling_list", st, body)

    st, body = http_json(
        "GET",
        "/api/Production/work-order-progress/list?page=1&pageSize=10",
        token=token,
    )
    step(report, "s2_work_order_progress_list", st, body)

    # ── S3 列表冒烟 ──
    st, body = http_json("GET", "/api/Supply/purchase-request/list?page=1&pageSize=10", token=token)
    step(report, "s3_purchase_request_list", st, body)

    st, body = http_json("GET", "/api/Supply/purchase-order/list?page=1&pageSize=10", token=token)
    step(report, "s3_purchase_order_list", st, body)

    st, body = http_json(
        "GET",
        "/api/Supply/work-order-material-readiness/list?page=1&pageSize=10",
        token=token,
    )
    step(report, "s3_material_readiness_list", st, body)

    st, body = http_json("POST", "/api/Supply/delivery-schedule/generate", None, token, timeout=180)
    step(report, "s3_delivery_schedule_generate", st, body)

    # ── S4 列表冒烟 ──
    st, body = http_json(
        "GET",
        "/api/ProcurementExecution/supplier-shipment/list?page=1&pageSize=10",
        token=token,
    )
    step(report, "s4_supplier_shipment_list", st, body)

    # 看板 KPI
    for mod in ("S1", "S2", "S3", "S4"):
        st, body = http_json("GET", f"/api/AidopKanban/module-detail?moduleCode={mod}", token=token)
        step(report, f"kanban_{mod}", st, body)

    with pymysql.connect(**CONN) as conn:
        report["after"] = snapshot(conn, order_id, bill_no)

    after = report["after"]
    after_review = report["after_review"]
    checks = [
        {"name": "create_order", "ok": bool(order_id), "detail": bill_no},
        {
            "name": "review_api",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s1_order_review"),
            "detail": review_res.get("message") if isinstance(review_res, dict) else review_res,
        },
        {
            "name": "work_orders_after_review",
            "ok": len(after_review.get("work_orders") or []) >= 1,
            "detail": [w.get("WorkOrd") for w in (after_review.get("work_orders") or [])],
        },
        {
            "name": "resource_check_examine",
            "ok": len(after_review.get("examine") or []) >= 1
            or (after_review.get("examine_lines") or 0) >= 1
            or (isinstance(review_res, dict) and (review_res.get("resourceCheckCount") or 0) >= 1),
            "detail": {
                "examine": len(after_review.get("examine") or []),
                "lines": after_review.get("examine_lines"),
                "api": (review_res or {}).get("resourceCheckCount") if isinstance(review_res, dict) else None,
            },
        },
        {
            "name": "generate_material_requirement",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s1_generate_material_requirement"),
        },
        {
            "name": "production_schedule",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s2_production_schedule"),
        },
        {
            "name": "sync_material_requirement",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s2_sync_material_requirement"),
        },
        {
            "name": "s2_scheduling_list",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s2_scheduling_list"),
        },
        {
            "name": "s3_purchase_request_list",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s3_purchase_request_list"),
        },
        {
            "name": "s3_material_readiness_list",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s3_material_readiness_list"),
        },
        {
            "name": "s4_supplier_shipment_list",
            "ok": next(s["ok"] for s in report["steps"] if s["step"] == "s4_supplier_shipment_list"),
        },
        {
            "name": "demand_or_pr_after_mrp",
            "ok": (after.get("demand") or 0) > 0 or (after.get("pr") or 0) > 0,
            "detail": {"demand": after.get("demand"), "pr": after.get("pr"), "po": after.get("po")},
        },
    ]
    report["checks"] = checks
    report["passed"] = all(c["ok"] for c in checks)
    report["failed"] = [c["name"] for c in checks if not c["ok"]]
    report["step_failures"] = [s["step"] for s in report["steps"] if not s.get("ok")]

    out = r"d:\Projects\Ai-DOP\SourceCode\ZZYDOP\doc\_verify_s1_s4_business_chain_e2e_result.json"
    with open(out, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2, default=str)

    # compact console summary
    summary = {
        "passed": report["passed"],
        "bill_no": bill_no,
        "order_id": order_id,
        "failed": report["failed"],
        "step_failures": report["step_failures"],
        "checks": checks,
        "after": {
            "work_orders": [w.get("WorkOrd") for w in (after.get("work_orders") or [])],
            "entry_progress": [e.get("progress") for e in (after.get("entries") or [])],
            "demand": after.get("demand"),
            "pr": after.get("pr"),
            "po": after.get("po"),
            "examine_lines": after.get("examine_lines"),
        },
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2, default=str))
    sys.exit(0 if report["passed"] else 5)


if __name__ == "__main__":
    main()