Mycontent/scripts/test/soul_api/test_ckb_leads_submitted.py

# -*- coding: utf-8 -*-
"""
测试 GET /api/db/ckb-leads?mode=submitted 是否返回 phone、wechatId 字段。

背景：存客宝工作台「加入/匹配提交」表格读这两个字段；旧版 API 未返回会导致全程显示「—」。

用法（仓库根目录）::

    pip install -r scripts/test/requirements-test.txt

    # 任选其一提供管理端 Token
    set ADMIN_TOKEN=eyJhbGciOi...   # 浏览器 localStorage「admin_token」
    # 或不设 ADMIN_TOKEN，用账号密码（与 scripts/test/config.py 一致）
    set SOUL_ADMIN_USERNAME=admin
    set SOUL_ADMIN_PASSWORD=admin123

    # 可选：SOUL_API_BASE=http://127.0.0.1:8080  或 SOUL_TEST_ENV=local
    python scripts/test/soul_api/test_ckb_leads_submitted.py

    python scripts/test/soul_api/test_ckb_leads_submitted.py --page-size 5

退出码：0 表示每条记录均含 phone/wechatId 键（值可为空）；1 表示请求失败或缺少字段。
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

import requests

# 复用 scripts/test 下的 config / util
_TEST_ROOT = Path(__file__).resolve().parent.parent
if str(_TEST_ROOT) not in sys.path:
    sys.path.insert(0, str(_TEST_ROOT))

from config import (  # noqa: E402
    ADMIN_PASSWORD,
    ADMIN_USERNAME,
    API_BASE,
    ENV_LABEL,
    get_env_banner,
)


def _admin_token(base_url: str) -> str:
    t = os.environ.get("ADMIN_TOKEN", "").strip()
    if t:
        return t
    try:
        r = requests.post(
            f"{base_url}/api/admin",
            json={"username": ADMIN_USERNAME, "password": ADMIN_PASSWORD},
            timeout=15,
        )
        data = r.json()
        if data.get("success") and data.get("token"):
            return str(data["token"])
    except Exception as e:
        print(f"[错误] 管理端登录失败: {e}", file=sys.stderr)
    return ""


def _headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}", "Accept": "application/json"}


def main() -> int:
    parser = argparse.ArgumentParser(description="测试 ckb-leads submitted 接口字段")
    parser.add_argument("--page-size", type=int, default=20, help="每页条数，默认 20")
    args = parser.parse_args()

    base = os.environ.get("SOUL_API_BASE", "").strip().rstrip("/") or API_BASE
    print(get_env_banner())
    print(f"请求基址: {base}（当前解析环境: {ENV_LABEL}）")

    token = _admin_token(base)
    if not token:
        print(
            "[错误] 无管理端 Token。请设置环境变量 ADMIN_TOKEN，"
            "或配置 SOUL_ADMIN_USERNAME / SOUL_ADMIN_PASSWORD 并能 POST /api/admin 登录。",
            file=sys.stderr,
        )
        return 1

    url = f"{base}/api/db/ckb-leads"
    params = {"mode": "submitted", "page": 1, "pageSize": min(max(1, args.page_size), 100)}
    try:
        r = requests.get(url, headers=_headers(token), params=params, timeout=20)
    except requests.RequestException as e:
        print(f"[错误] 请求异常: {e}", file=sys.stderr)
        return 1

    if r.status_code == 401:
        print("[错误] 401 未授权：ADMIN_TOKEN 无效或已过期。", file=sys.stderr)
        return 1
    if r.status_code != 200:
        print(f"[错误] HTTP {r.status_code}: {r.text[:500]}", file=sys.stderr)
        return 1

    try:
        data = r.json()
    except json.JSONDecodeError:
        print(f"[错误] 响应非 JSON: {r.text[:300]}", file=sys.stderr)
        return 1

    if not data.get("success"):
        print(f"[错误] success=false: {data}", file=sys.stderr)
        return 1

    records = data.get("records") or []
    total = data.get("total")
    print(f"\n[INFO] submitted 总数约: {total!r}，本页记录数: {len(records)}")

    if not records:
        print("[通过] 无 join/match 记录，跳过字段检查。")
        return 0

    missing = []
    sample_rows = []
    for rec in records:
        rid = rec.get("id")
        uid = rec.get("userId", "")
        mtype = rec.get("matchType", "")
        has_phone_key = "phone" in rec
        has_wechat_key = "wechatId" in rec
        if not has_phone_key or not has_wechat_key:
            missing.append(
                {"id": rid, "userId": uid, "matchType": mtype, "missing": [k for k, ok in [("phone", has_phone_key), ("wechatId", has_wechat_key)] if not ok]}
            )
        sample_rows.append(
            {
                "id": rid,
                "userId": uid,
                "matchType": mtype,
                "phone": rec.get("phone", "<无此键>"),
                "wechatId": rec.get("wechatId", "<无此键>"),
            }
        )

    print("\n前几条（节选）:")
    for row in sample_rows[: min(8, len(sample_rows))]:
        print(f"  id={row['id']} type={row['matchType']} userId={row['userId']!r} "
              f"phone={row['phone']!r} wechatId={row['wechatId']!r}")

    if missing:
        print("\n[失败] 下列记录缺少 API 字段（需部署含 phone/wechatId 的 soul-api）:", file=sys.stderr)
        for m in missing[:15]:
            print(f"  {m}", file=sys.stderr)
        if len(missing) > 15:
            print(f"  … 另有 {len(missing) - 15} 条", file=sys.stderr)
        return 1

    non_empty_contact = sum(1 for rec in records if (rec.get("phone") or rec.get("wechatId")))
    print(f"\n[通过] 本页每条均含 phone、wechatId 键；其中有非空联系方式的条目: {non_empty_contact}/{len(records)}")
    print("（值为空仅代表库内该 join/match 行未写入手机/微信，与「工作台是否展示 —」已不是同一类问题）")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())