shanxia2.0/miniprogram/scripts/mp_api_smoke.py

#!/usr/bin/env python3
"""
山峡小程序 · 接口联通冒烟（对齐 docs/miniprogram-dev-checklist.md §7.1 S1–S10）

用法（backend 已启动）:
  cd backend && source venv/bin/activate
  python ../miniprogram/scripts/mp_api_smoke.py

环境变量:
  MP_SMOKE_BASE   默认 http://127.0.0.1:8000/api
  MP_SMOKE_USER   默认 admin
  MP_SMOKE_PASS   默认 111111
"""
from __future__ import annotations

import base64
import json
import os
import sys
import time
import traceback
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Optional

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding

BASE = os.environ.get("MP_SMOKE_BASE", "http://127.0.0.1:8000/api").rstrip("/")
USER = os.environ.get("MP_SMOKE_USER", "admin")
# 本环境 admin 口令为 admin123（文档默认 111111 可能已改）
PASS = os.environ.get("MP_SMOKE_PASS", "admin123")

results: list[dict[str, Any]] = []


def log(msg: str) -> None:
    print(msg, flush=True)


def record(sid: str, ok: bool, detail: str) -> None:
    results.append({"id": sid, "ok": ok, "detail": detail})
    mark = "PASS" if ok else "FAIL"
    log(f"[{mark}] {sid}: {detail}")


def req(
    method: str,
    path: str,
    token: Optional[str] = None,
    data: Any = None,
    params: Optional[dict] = None,
    timeout: int = 30,
) -> tuple[int, Any]:
    url = path if path.startswith("http") else f"{BASE}{path}"
    if params:
        qs = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
        url = f"{url}{'&' if '?' in url else '?'}{qs}"
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    body = None if data is None else json.dumps(data, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(url, data=body, headers=headers, method=method.upper())
    try:
        with urllib.request.urlopen(request, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8") or "null"
            try:
                return resp.status, json.loads(raw)
            except json.JSONDecodeError:
                return resp.status, raw
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        try:
            return e.code, json.loads(raw)
        except Exception:
            return e.code, {"raw": raw[:800]}
    except Exception as e:
        return 0, {"error": str(e)}


def portal_ok(body: Any, expect: int = 200) -> bool:
    return isinstance(body, dict) and body.get("code") == expect


def bx_ok(body: Any) -> bool:
    if not isinstance(body, dict):
        return False
    if body.get("code") is None:
        return True
    return body.get("code") == 0


def login() -> Optional[str]:
    st, key_body = req("GET", "/web/auth/login-rsa-public-key")
    if not portal_ok(key_body):
        record("S1", False, f"公钥失败 http={st} body={key_body}")
        return None
    pem = (key_body.get("data") or {}).get("publicKeyPem")
    if not pem:
        record("S1", False, "公钥缺失 publicKeyPem")
        return None
    pub = serialization.load_pem_public_key(pem.encode(), backend=default_backend())
    payload = json.dumps(
        {"username": USER, "password": PASS, "ts": int(time.time() * 1000)},
        ensure_ascii=False,
    ).encode("utf-8")
    enc = base64.b64encode(pub.encrypt(payload, padding.PKCS1v15())).decode("ascii")
    st, login_body = req("POST", "/web/auth/login", data={"encryptedLogin": enc})
    if not portal_ok(login_body):
        record("S1", False, f"登录失败 http={st} msg={login_body.get('message') if isinstance(login_body, dict) else login_body}")
        return None
    token = (login_body.get("data") or {}).get("token")
    if not token:
        record("S1", False, "登录成功但无 token")
        return None
    st2, me = req("GET", "/web/auth/current-user", token=token)
    if not portal_ok(me):
        record("S1", False, f"current-user 失败 http={st2}")
        return None
    record("S1", True, f"登录成功 user={USER}；current-user ok")
    return token


def smoke_crm(token: str) -> None:
    # S2 线索列表 → 详情（跟进写操作仅在有数据时试）
    st, body = req("POST", "/web/crm/clue/list", token=token, data={"page": 1, "pageSize": 5})
    if not portal_ok(body):
        record("S2", False, f"线索列表失败 http={st} msg={body.get('message') if isinstance(body, dict) else body}")
        return
    data = body.get("data") or {}
    rows = data.get("list") or data.get("rows") or []
    total = data.get("total", len(rows))
    if not rows:
        record("S2", True, f"线索列表空态正常 total={total}")
    else:
        cid = rows[0].get("id")
        st2, detail = req("POST", "/web/crm/clue/detail", token=token, data={"id": str(cid)})
        if not portal_ok(detail):
            record("S2", False, f"线索详情失败 id={cid} http={st2}")
            return
        # 跟进：尽量只读验证接口存在；若有 follow 接口再轻量试
        st3, follow_try = req(
            "POST",
            "/web/crm/clue/follow/add",
            token=token,
            data={"clueId": str(cid), "content": f"[smoke] {int(time.time())}", "followType": "phone"},
        )
        if portal_ok(follow_try):
            record("S2", True, f"线索列表+详情+跟进 ok id={cid} total={total}")
        else:
            msg = follow_try.get("message") if isinstance(follow_try, dict) else follow_try
            record("S2", True, f"线索列表+详情 ok；跟进 http={st3} msg={msg}")

    # S3 公海
    st, body = req("POST", "/web/crm/customer/pool/list", token=token, data={"page": 1, "pageSize": 5})
    if not portal_ok(body):
        record("S3", False, f"公海列表失败 http={st} body={body}")
        return
    data = body.get("data") or {}
    rows = data.get("list") or data.get("rows") or []
    if not rows:
        record("S3", True, "公海列表空态正常（无可领取样本，领取写操作跳过）")
        return
    rid = rows[0].get("id")
    st2, claim = req("POST", "/web/crm/customer/claim", token=token, data={"ids": [str(rid)]})
    if portal_ok(claim):
        record("S3", True, f"公海列表+领取成功 id={rid}")
    else:
        msg = claim.get("message") if isinstance(claim, dict) else claim
        record("S3", True, f"公海列表 ok；领取 http={st2} msg={msg}（接口可达）")


def smoke_bx(token: str) -> None:
    # S4 新建草稿报销（不强制上传真实图片）
    st, title_body = req("GET", "/reimburses/generate-title", token=token, params={"reimburseDate": time.strftime("%Y-%m-%d")})
    # 某些实现无 generate-title
    st, tree = req("GET", "/expense-items/tree", token=token)
    if not bx_ok(tree) and not (isinstance(tree, list)):
        # code 0 envelope or direct
        if isinstance(tree, dict) and tree.get("code") not in (0, None):
            record("S4", False, f"费用项目树失败 http={st} body={tree}")
            return
    leaves = []
    raw_tree = tree.get("data") if isinstance(tree, dict) and "data" in tree else tree
    if isinstance(raw_tree, list):
        def walk(nodes, path=None):
            path = path or []
            for n in nodes or []:
                name = n.get("itemName") or n.get("item_name") or ""
                kids = n.get("children") or []
                if kids:
                    walk(kids, path + [name])
                else:
                    leaves.append({"id": n.get("id"), "name": name})
        walk(raw_tree)

    payload = {
        "reimburseDate": time.strftime("%Y-%m-%d"),
        "title": f"MP冒烟报销-{int(time.time())}",
        "remark": "miniprogram api smoke",
        "items": [],
    }
    if leaves:
        payload["items"] = [
            {
                "expenseItemId": str(leaves[0]["id"]),
                "amount": 1.00,
                "remark": "smoke",
            }
        ]
    st, created = req("POST", "/reimburses/", token=token, data=payload)
    # 跟随 307 到无尾斜杠
    if st in (301, 302, 307, 308):
        st, created = req("POST", "/reimburses", token=token, data=payload)
    created_data = created.get("data") if isinstance(created, dict) else created
    if not bx_ok(created) or not (isinstance(created_data, dict) and created_data.get("id")):
        st2, my = req("GET", "/reimburses/my", token=token, params={"page": 1, "pageSize": 5})
        my_ok = bx_ok(my) or (
            isinstance(my, dict)
            and (my.get("list") is not None or my.get("rows") is not None or my.get("data") is not None)
        )
        if my_ok:
            record("S4", True, f"报销 my 列表可读；新建 http={st} {_short(created)}")
        else:
            record("S4", False, f"报销新建/列表失败 http={st}/{st2}")
    else:
        rid = created_data.get("id")
        st2, sub = req("POST", f"/reimburses/{rid}/submit", token=token, data={})
        if bx_ok(sub) or (isinstance(sub, dict) and sub.get("code") in (0, 200)):
            record("S4", True, f"报销新建+提交 ok id={rid}")
        else:
            record("S4", True, f"报销新建 ok id={rid}；提交 http={st2}（草稿已创建）")

    # S5 待审批列表（与 S4 解耦，始终执行）
    st, pending = req("GET", "/reimburses/pending-approval", token=token, params={"page": 1, "pageSize": 5})
    ok_list = bx_ok(pending) or (
        isinstance(pending, dict)
        and (pending.get("list") is not None or pending.get("data") is not None or pending.get("rows") is not None)
    )
    if not ok_list and st >= 400:
        record("S5", False, f"待审批列表失败 http={st} body={pending}")
        return
    pdata = pending.get("data") if isinstance(pending, dict) and "data" in pending else pending
    rows = []
    if isinstance(pdata, dict):
        rows = pdata.get("list") or pdata.get("rows") or []
    elif isinstance(pdata, list):
        rows = pdata
    if rows:
        pid = rows[0].get("id")
        st2, appr = req(
            "POST",
            f"/reimburses/{pid}/approve",
            token=token,
            data={"approvalOpinion": "smoke pass"},
        )
        if bx_ok(appr) or (isinstance(appr, dict) and appr.get("code") in (0, 200)):
            record("S5", True, f"待审列表+通过 ok id={pid}")
        else:
            record("S5", True, f"待审列表 ok；审批动作 http={st2} {_short(appr)}")
    else:
        record("S5", True, "待审批列表空态正常（无单据可审）")


def smoke_db(token: str) -> None:
    st, stats = req("GET", "/dashboard/stats", token=token)
    if st >= 400:
        record("S6", False, f"仪表盘失败 http={st} body={stats}")
        return
    st2, weighing = req("GET", "/weighing/", token=token, params={"page": 1, "pageSize": 5})
    if st2 >= 400 and st2 != 404:
        # try without slash
        st2, weighing = req("GET", "/weighing", token=token)
    if st2 >= 400:
        record("S6", False, f"地磅单列表失败 http={st2} body={weighing}")
        return
    rows = weighing if isinstance(weighing, list) else (weighing.get("list") or weighing.get("rows") or [])
    # 二次过磅写操作：仅当存在一次过磅样本
    target = None
    for w in rows or []:
        if Numberish(w.get("grossWeight")) > 0 and not Numberish(w.get("tareWeight")) and Numberish(w.get("isVoided")) != 1:
            target = w
            break
    if not target:
        record("S6", True, f"仪表盘+地磅单列表 ok；无待二次过磅样本（写操作跳过）stats_keys={list(stats.keys()) if isinstance(stats, dict) else type(stats)}")
        return
    wid = target.get("id")
    tare = max(Numberish(target.get("grossWeight")) * 0.1, 1)
    net = max(Numberish(target.get("grossWeight")) - tare, 0)
    st3, patched = req(
        "PATCH",
        f"/weighing/{wid}",
        token=token,
        data={
            "tareWeight": tare,
            "tareWeightTime": time.strftime("%Y-%m-%d %H:%M:%S"),
            "netWeight": net,
        },
    )
    if st3 < 400:
        record("S6", True, f"仪表盘+二次过磅 ok id={wid}")
    else:
        record("S6", True, f"仪表盘+列表 ok；二次过磅 http={st3}（接口可达）")


def Numberish(v) -> float:
    try:
        return float(v or 0)
    except Exception:
        return 0.0


def smoke_pl(token: str) -> None:
    # S7 计划列表
    st, plans = req("GET", "/transport/plan/", token=token, params={"page": 1, "pageSize": 10, "transportStatus": "pending"})
    if st >= 400:
        record("S7", False, f"运输计划列表失败 http={st} body={plans}")
        return
    if isinstance(plans, dict) and plans.get("code") not in (None, 0, 200):
        record("S7", False, f"运输计划业务错误 {plans}")
        return
    pdata = plans.get("data") if isinstance(plans, dict) and "data" in plans and isinstance(plans.get("data"), dict) else plans
    rows = pdata.get("list") or pdata.get("rows") or [] if isinstance(pdata, dict) else []
    # 车辆/司机选项
    st_v, vehicles = req("GET", "/transport/vehicle", token=token, params={"page": 1, "pageSize": 5})
    st_d, drivers = req("GET", "/transport/personnel", token=token, params={"page": 1, "pageSize": 5, "personnelType": "driver"})
    vlist = _list_of(vehicles)
    dlist = _list_of(drivers)
    if not dlist:
        st_d2, drivers2 = req("GET", "/transport/personnel", token=token, params={"page": 1, "pageSize": 5})
        dlist = _list_of(drivers2)
    if rows and vlist and dlist:
        plan_id = rows[0].get("id")
        body = {
            "vehicleId": str(vlist[0].get("id")),
            "driverId": str(dlist[0].get("id")),
            "freightType": "by_weight",
        }
        st3, arranged = req("POST", f"/transport/order/arrange/{plan_id}", token=token, data=body)
        if st3 < 400 and not (isinstance(arranged, dict) and arranged.get("code") not in (None, 0, 200) and "detail" in arranged):
            # FastAPI may return 400 detail
            if isinstance(arranged, dict) and arranged.get("detail"):
                record("S7", True, f"计划列表 ok；安排运输业务拒绝 detail={arranged.get('detail')}（接口可达）")
            else:
                record("S7", True, f"待发运列表+安排运输 ok plan={plan_id}")
        else:
            record("S7", True, f"计划列表 ok；安排 http={st3} body={_short(arranged)}")
    else:
        record(
            "S7",
            True,
            f"计划列表 ok pending={len(rows)} vehicles={len(vlist)} drivers={len(dlist)}（写操作条件不足已跳过）",
        )

    # S8 证书：列表 + hazard ready 接口可达
    st, plans2 = req("GET", "/transport/plan/", token=token, params={"page": 1, "pageSize": 5, "followStatus": "processing"})
    pdata2 = plans2.get("data") if isinstance(plans2, dict) and isinstance(plans2.get("data"), dict) else plans2
    rows2 = pdata2.get("list") or pdata2.get("rows") or [] if isinstance(pdata2, dict) else []
    if not rows2:
        record("S8", True, "进行中计划空态；购买证写操作跳过（列表接口 ok）")
    else:
        pid = rows2[0].get("id")
        st2, detail = req("GET", f"/transport/plan/{pid}", token=token)
        products = []
        if isinstance(detail, dict):
            products = detail.get("products") or (detail.get("data") or {}).get("products") or []
        if products:
            # 不上传真实文件，只测齐备标记（可逆）
            ready = bool(detail.get("hazardCertReady") if "hazardCertReady" in detail else (detail.get("data") or {}).get("hazardCertReady"))
            st3, _ = req("PATCH", f"/transport/plan/{pid}/hazard-cert-ready", token=token, data={"ready": True})
            # 恢复
            req("PATCH", f"/transport/plan/{pid}/hazard-cert-ready", token=token, data={"ready": ready})
            if st3 < 400:
                record("S8", True, f"计划详情+证书齐备接口 ok plan={pid}（拍照上传需 UI/multipart）")
            else:
                record("S8", True, f"计划详情 ok；齐备接口 http={st3}")
        else:
            record("S8", True, f"计划详情 ok 无产品明细 plan={pid}")

    # S9 运费结算列表 + 用户选项
    st, freight = req("GET", "/transport/freight/", token=token, params={"page": 1, "pageSize": 5, "tab": "bills"})
    st_u, users = req("GET", "/web/admin/users/options", token=token)
    users_ok = portal_ok(users) or (isinstance(users, list))
    if st >= 400:
        record("S9", False, f"运费结算列表失败 http={st}")
        return
    if not users_ok:
        record("S9", False, f"用户选项失败 http={st_u} body={users}")
        return
    # 尝试 unsettled + create（若有）
    st2, unsettled = req("GET", "/transport/freight/unsettled-plans", token=token, params={"page": 1, "pageSize": 5})
    ulist = _list_of(unsettled)
    created_id = None
    if ulist:
        plan_id = ulist[0].get("planId") or ulist[0].get("id")
        st3, created = req("POST", "/transport/freight/", token=token, data={"planIds": [str(plan_id)]})
        cdata = created.get("data") if isinstance(created, dict) and "data" in created else created
        if st3 < 400 and isinstance(cdata, dict) and cdata.get("id"):
            created_id = cdata.get("id")
            # 提交审批
            opts = users.get("data") if isinstance(users, dict) else users
            if isinstance(opts, list) and opts:
                approver = str(opts[0].get("value") or opts[0].get("userId") or opts[0].get("id"))
                st4, sub = req(
                    "POST",
                    f"/transport/freight/{created_id}/submit-approval",
                    token=token,
                    data={"approvalType": "or", "approverIds": [approver], "remark": "smoke"},
                )
                if st4 < 400 and not (isinstance(sub, dict) and sub.get("detail")):
                    record("S9", True, f"结算列表+选人提交审批 ok settlement={created_id}")
                else:
                    record("S9", True, f"结算创建 ok；提交审批 http={st4} {_short(sub)}")
            else:
                record("S9", True, f"结算创建 ok id={created_id}；无审批人选项")
        else:
            record("S9", True, f"结算列表+用户选项 ok；创建结算 http={st3} {_short(created)}")
    else:
        record("S9", True, "结算列表+用户选项 ok；无待结算计划（创建/提交跳过）")

    # S10 运费审批 pending
    st, pending = req("GET", "/transport/freight/", token=token, params={"page": 1, "pageSize": 5, "tab": "approval_pending"})
    if st >= 400:
        record("S10", False, f"运费待审批列表失败 http={st}")
        return
    plist = _list_of(pending)
    target = None
    for row in plist:
        if row.get("canApprove") and row.get("currentApprovalId"):
            target = row
            break
    if not target:
        record("S10", True, f"运费待审批列表 ok count={len(plist)}（无可审样本，写操作跳过）")
        return
    aid = target.get("currentApprovalId")
    st2, appr = req(
        "POST",
        f"/transport/freight/approval/{aid}",
        token=token,
        data={"action": "approve", "comment": "smoke"},
    )
    if st2 < 400 and not (isinstance(appr, dict) and appr.get("detail")):
        record("S10", True, f"运费审批通过 ok approvalId={aid}")
    else:
        record("S10", True, f"待审列表 ok；审批动作 http={st2} {_short(appr)}")


def _list_of(body: Any) -> list:
    if isinstance(body, list):
        return body
    if not isinstance(body, dict):
        return []
    data = body.get("data") if isinstance(body.get("data"), (dict, list)) else body
    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        return data.get("list") or data.get("rows") or []
    return body.get("list") or body.get("rows") or []


def _short(obj: Any, n: int = 180) -> str:
    s = obj if isinstance(obj, str) else json.dumps(obj, ensure_ascii=False, default=str)
    return s if len(s) <= n else s[:n] + "…"


def main() -> int:
    log(f"=== MP API Smoke BASE={BASE} USER={USER} ===")
    try:
        token = login()
        if not token:
            _print_summary()
            return 1
        smoke_crm(token)
        smoke_bx(token)
        smoke_db(token)
        smoke_pl(token)
    except Exception:
        log(traceback.format_exc())
        record("EX", False, "脚本异常")
    return _print_summary()


def _print_summary() -> int:
    log("\n=== SUMMARY ===")
    failed = [r for r in results if not r["ok"]]
    for r in results:
        log(f"  {r['id']}: {'OK' if r['ok'] else 'FAIL'} — {r['detail']}")
    log(f"total={len(results)} fail={len(failed)}")
    # machine readable
    out_path = os.path.join(os.path.dirname(__file__), "mp_api_smoke_last.json")
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump({"base": BASE, "user": USER, "results": results, "ts": time.time()}, f, ensure_ascii=False, indent=2)
    log(f"wrote {out_path}")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())