soul-yongping/scripts/test/process/test_backfill_persons_ckb_api_key.py

# -*- coding: utf-8 -*-
"""
流程测试：从存客宝获取所有计划的 apiKey，补齐本地 persons.ckb_api_key

场景：persons 表有 ckb_plan_id 但 ckb_api_key 为空时，调用存客宝 plan/detail 获取 apiKey 并更新本地

前置条件：
- 测试环境（SOUL_TEST_ENV=souldev 或 local）
- soul-api 可连通，CKB_OPEN_API_KEY、CKB_OPEN_ACCOUNT 已配置（soul-api/.env）
"""
import hashlib
import time

import pytest
import requests

from config import SOUL_API_ENV
from util import admin_headers

CKB_OPEN_BASE = "https://ckbapi.quwanzhi.com"


def _ckb_open_sign(account: str, ts: int, api_key: str) -> str:
    """存客宝开放 API 签名：sign = MD5(MD5(account+timestamp) + apiKey)"""
    plain = account + str(ts)
    first = hashlib.md5(plain.encode()).hexdigest()
    return hashlib.md5((first + api_key).encode()).hexdigest()


def _ckb_get_token(api_key: str, account: str) -> str:
    """获取存客宝开放 API JWT"""
    ts = int(time.time())
    sign = _ckb_open_sign(account, ts, api_key)
    r = requests.post(
        f"{CKB_OPEN_BASE}/v1/open/auth/token",
        json={"apiKey": api_key, "account": account, "timestamp": ts, "sign": sign},
        timeout=15,
    )
    data = r.json()
    if data.get("code") != 200:
        raise RuntimeError(f"存客宝鉴权失败: {data.get('message', r.text)}")
    token = (data.get("data") or {}).get("token")
    if not token:
        raise RuntimeError("存客宝返回无 token")
    return token


def _ckb_get_plan_api_key(token: str, plan_id: int) -> str:
    """调用 plan/detail 获取计划级 apiKey"""
    r = requests.get(
        f"{CKB_OPEN_BASE}/v1/plan/detail",
        params={"planId": plan_id},
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        timeout=15,
    )
    data = r.json()
    if data.get("code") != 200:
        raise RuntimeError(f"获取计划详情失败 planId={plan_id}: {data.get('message', r.text)}")
    api_key = (data.get("data") or {}).get("apiKey")
    if not api_key:
        raise RuntimeError(f"计划 {plan_id} 详情中无 apiKey")
    return api_key


def _load_ckb_config() -> tuple[str, str]:
    """从 soul-api/.env 加载 CKB 配置"""
    def _parse_env(path):
        out = {}
        if not path.exists():
            return out
        for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            out[k.strip()] = v.strip().strip('"').strip("'")
        return out

    for name in [".env", ".env.development", ".env.production"]:
        env_path = SOUL_API_ENV / name
        loaded = _parse_env(env_path)
        api_key = (loaded.get("CKB_OPEN_API_KEY") or "").strip()
        account = (loaded.get("CKB_OPEN_ACCOUNT") or "").strip()
        if api_key and account:
            return api_key, account
    return "", ""


def test_backfill_persons_ckb_api_key(admin_token, base_url):
    """
    从存客宝获取所有计划的 apiKey，补齐本地 persons.ckb_api_key 为空的记录
    """
    if not admin_token:
        pytest.skip("admin 登录失败，跳过")

    ckb_api_key, ckb_account = _load_ckb_config()
    if not ckb_api_key or not ckb_account:
        pytest.skip("CKB_OPEN_API_KEY 或 CKB_OPEN_ACCOUNT 未配置，跳过")

    # 1. 拉取 persons 列表
    r = requests.get(
        f"{base_url}/api/db/persons",
        headers=admin_headers(admin_token),
        timeout=10,
    )
    assert r.status_code == 200, f"拉取 persons 失败: {r.text}"
    data = r.json()
    assert data.get("success") is True, f"拉取 persons 失败: {data}"
    persons = data.get("persons") or []

    # 2. 筛选需要补全的：ckb_plan_id > 0 且 ckb_api_key 为空
    need_backfill = [
        p
        for p in persons
        if (p.get("ckbPlanId") or 0) > 0
        and not (p.get("ckbApiKey") or "").strip()
    ]

    if not need_backfill:
        pytest.skip("无需要补全 ckb_api_key 的 Person，跳过")

    # 3. 获取存客宝 JWT
    ckb_token = _ckb_get_token(ckb_api_key, ckb_account)

    # 4. 逐个补全
    updated = 0
    failed = []
    for p in need_backfill:
        plan_id = p.get("ckbPlanId") or 0
        person_id = p.get("personId") or ""
        name = p.get("name") or ""
        try:
            api_key = _ckb_get_plan_api_key(ckb_token, plan_id)
        except Exception as e:
            failed.append((name, str(e)))
            continue

        # 5. 调用 soul-api 更新 Person（POST 带 personId 为更新，传完整字段避免覆盖）
        payload = {
            "personId": person_id,
            "name": name,
            "label": (p.get("label") or ""),
            "ckbApiKey": api_key,
            "greeting": (p.get("greeting") or ""),
            "tips": (p.get("tips") or ""),
            "remarkType": (p.get("remarkType") or ""),
            "remarkFormat": (p.get("remarkFormat") or ""),
            "startTime": (p.get("startTime") or "09:00"),
            "endTime": (p.get("endTime") or "18:00"),
        }
        if p.get("addFriendInterval"):
            payload["addFriendInterval"] = p["addFriendInterval"]
        r_update = requests.post(
            f"{base_url}/api/db/persons",
            headers=admin_headers(admin_token),
            json=payload,
            timeout=15,
        )
        if r_update.status_code == 200 and r_update.json().get("success"):
            updated += 1
        else:
            failed.append((name, r_update.text or "更新失败"))

    assert not failed, f"补全失败: {failed}"
    assert updated > 0, f"应至少补全 1 条，实际补全 {updated} 条"
    print(f"\n[backfill] 成功补全 {updated} 条 persons.ckb_api_key")