Enhance profile editing and sharing functionality
- Added a new feature for sharing profile cards, including special handling for forwarding to friends and displaying a canvas cover with user information. - Updated the mini program's profile-edit page to generate a shareable card with a structured layout, including user avatar, nickname, and additional information. - Improved the documentation to reflect the new sharing capabilities and updated the last modified date for relevant entries.
This commit is contained in:
166
scripts/test/process/test_backfill_persons_ckb_api_key.py
Normal file
166
scripts/test/process/test_backfill_persons_ckb_api_key.py
Normal file
@@ -0,0 +1,166 @@
|
||||
# -*- 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")
|
||||
Reference in New Issue
Block a user