huaqiapp/scripts/verify/api_flow_test.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""花期 APP · 全链路接口验证（M1–M9 全模块）

用法：
    HQ_TEST_ACCOUNT=demo HQ_TEST_PASSWORD=demo123456 python3 scripts/verify/api_flow_test.py
    （可选）HQ_TEST_ACCOUNT2 / HQ_TEST_PASSWORD2 用于家人绑定等双人链路
"""
import datetime
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

BASE = os.environ.get("HQ_API_BASE", "https://huaqiapp.gzyakj.com.cn")
ACCOUNT = os.environ.get("HQ_TEST_ACCOUNT", "")
PASSWORD = os.environ.get("HQ_TEST_PASSWORD", "")
ACCOUNT2 = os.environ.get("HQ_TEST_ACCOUNT2", "")
PASSWORD2 = os.environ.get("HQ_TEST_PASSWORD2", "")
if not ACCOUNT or not PASSWORD:
    print("请先设置测试账号环境变量：HQ_TEST_ACCOUNT / HQ_TEST_PASSWORD")
    sys.exit(2)

TOKEN = None
TOKEN2 = None
PASSED = []
FAILED = []


def call(path, params=None, method="GET", token=None):
    url = BASE + path
    data = None
    if method == "GET" and params:
        url += "?" + urllib.parse.urlencode(params)
    elif method == "POST":
        data = urllib.parse.urlencode(params or {}, doseq=True).encode()
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Accept", "application/json")
    tok = token if token else TOKEN
    if tok:
        req.add_header("token", tok)
    try:
        with urllib.request.urlopen(req, timeout=25) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return {"code": -e.code, "msg": e.read().decode("utf-8", "ignore")[:200]}
    except Exception as e:
        return {"code": -1, "msg": str(e)}


def check(name, cond, detail=""):
    if cond:
        PASSED.append(name)
        print(f"  ✅ {name}" + (f"  {detail}" if detail else ""))
    else:
        FAILED.append(f"{name} {detail}")
        print(f"  ❌ {name}  {detail}")


def login(account, password):
    r = call("/api/user/login", {"account": account, "password": password}, "POST", token="")
    info = (r.get("data") or {}).get("userinfo") or {}
    return info.get("token", "")


today = datetime.date.today()
start = today - datetime.timedelta(days=13)
month = today.strftime("%Y-%m")

print("=" * 72)
print("0. 登录鉴权")
r = call("/api/user/login", {"account": ACCOUNT, "password": PASSWORD}, "POST", token="")
check("登录 code=1", r.get("code") == 1, r.get("msg"))
info = (r.get("data") or {}).get("userinfo") or {}
TOKEN = info.get("token", "")
check("下发 token", bool(TOKEN), TOKEN[:12] + "…")

if ACCOUNT2 and PASSWORD2:
    TOKEN2 = login(ACCOUNT2, PASSWORD2)
    check("第二账号登录", bool(TOKEN2))

print("=" * 72)
print("1. M1 周期管理 + 智能提醒")
r = call("/api/cycle/save", {"start_date": str(start), "end_date": str(start + datetime.timedelta(days=4))}, "POST")
check("记录经期", r.get("code") == 1, r.get("msg"))
r = call("/api/cycle/current")
cur = r.get("data") or {}
check("当前周期阶段", cur.get("phase") in ("menstrual", "follicular", "ovulation", "luteal"), cur.get("phase_text"))
r = call("/api/cycle/index", {"month": month})
cal = r.get("data") or {}
import calendar as pycal
check("月历天数", len(cal.get("list") or []) == pycal.monthrange(today.year, today.month)[1])
r = call("/api/cycle/log", {"log_date": str(today), "flow": 2, "pain_level": 3, "symptom_ids": "1,2", "mood": -1, "sleep_hours": 6.5}, "POST")
check("每日记录", r.get("code") == 1)
r = call("/api/cycle/report", {"type": "month", "month": month})
check("健康报告", (r.get("data") or {}).get("stats", {}).get("log_days", 0) >= 1)

r = call("/api/setting/reminderSave", {"title": "复诊提醒", "remind_time": "09:30", "repeat_rule": "daily"}, "POST")
check("新增自定义提醒", r.get("code") == 1, r.get("msg"))
reminder_id = (r.get("data") or {}).get("id", 0)
r = call("/api/setting/reminderList")
rst = (r.get("data") or {}).get("list") or []
check("提醒列表包含新增", any(x.get("id") == reminder_id for x in rst), f"{len(rst)} 条")
r = call("/api/home/index")
home = r.get("data") or {}
check("首页聚合提醒合并", any(x.get("title") == "复诊提醒" for x in (home.get("reminders") or [])))

print("=" * 72)
print("2. M2 情绪疗愈 + 急救包")
r = call("/api/mood/save", {"mood": "anxious", "intensity": 3, "triggers": "工作", "content": "全链路测试"}, "POST")
check("情绪记录", r.get("code") == 1, r.get("msg"))
r = call("/api/mood/trend", {"days": 7})
check("情绪趋势", len((r.get("data") or {}).get("list") or []) == 7)
r = call("/api/mood/firstaid")
fa = r.get("data") or {}
check("急救包 ≥4 个工具", len(fa.get("list") or []) >= 4, f"{len(fa.get('list') or [])} 个")
check("急救包含热线", bool((fa.get("hotline") or {}).get("phone")), str((fa.get("hotline") or {}).get("phone")))

print("=" * 72)
print("3. M3 科学备孕")
r = call("/api/prepare/index")
prep = r.get("data") or {}
check("备孕总览", bool(prep.get("cycle")) and "lh_quota" in prep)
check("排卵与易孕窗口", prep.get("ovulation") is not None or prep.get("cycle", {}).get("has_data") is False, str(prep.get("ovulation")))

r = call("/api/prepare/bbtSave", {"log_date": str(today), "temperature": 36.62}, "POST")
check("体温记录", r.get("code") == 1, r.get("msg"))
r = call("/api/prepare/bbtList", {"days": 14})
bbt = r.get("data") or {}
check("体温曲线含今日", any(x.get("temperature") == 36.62 for x in (bbt.get("list") or [])), f"{bbt.get('count')} 天")

r = call("/api/prepare/lhtest", {"test_type": "lh", "log_date": str(today), "level": 3}, "POST")
if r.get("code") == 1:
    lh = r.get("data") or {}
    check("试纸识别", lh.get("result") == "strong", lh.get("result_text"))
    check("识别置信度", (lh.get("confidence") or 0) > 80, str(lh.get("confidence")))
    check("额度扣减", (lh.get("quota") or {}).get("used", 0) >= 1, str((lh.get("quota") or {}).get("used")))
else:
    # 免费额度用尽（每月 5 次）时应给出明确提示，属预期行为
    check("试纸识别额度控制", ("用完" in (r.get("msg") or "")) or ("会员" in (r.get("msg") or "")), r.get("msg"))

r = call("/api/prepare/taskSave", {"title": "预约孕前检查", "scope": "prepare", "category": "check"}, "POST")
check("备孕任务新增", r.get("code") == 1, r.get("msg"))
task_id = (r.get("data") or {}).get("id", 0)
r = call("/api/prepare/taskDone", {"id": task_id}, "POST")
check("备孕任务完成", (r.get("data") or {}).get("done") == 1)
r = call("/api/prepare/guide")
gd = r.get("data") or {}
check("同房指导锁态", "locked" in gd, f"locked={gd.get('locked')}")

print("=" * 72)
print("4. M4 家人共护")
r = call("/api/family/index")
fam = r.get("data") or {}
check("家庭总览", "binds" in fam and "points" in fam, f"绑定 {len(fam.get('binds') or [])} 人")
r = call("/api/family/invite", {"relation": "partner", "member_name": "小满的伴侣"}, "POST")
invite = r.get("data") or {}
check("生成邀请码", bool(invite.get("invite_code")), invite.get("invite_code"))
code = invite.get("invite_code", "")
bind_id = invite.get("id", 0)

if TOKEN2 and code:
    r = call("/api/family/confirm", {"invite_code": code}, "POST", token=TOKEN2)
    check("第二账号确认绑定", r.get("code") == 1, r.get("msg"))
    r = call("/api/family/messageSend", {"bind_id": bind_id, "content": "多喝温水，暖贴放包里了"}, "POST")
    check("留言发送", r.get("code") == 1, r.get("msg"))
    r = call("/api/family/messages", {"bind_id": bind_id})
    check("留言可回读", len((r.get("data") or {}).get("list") or []) >= 1)

r = call("/api/family/careAction", method="POST")
care = r.get("data") or {}
check("关怀行动积分", (care.get("points") or {}).get("total", 0) >= 2, str((care.get("points") or {}).get("total")))
r = call("/api/family/scripts")
check("话术库", len((r.get("data") or {}).get("list") or []) >= 3)
r = call("/api/family/healthView")
hv = r.get("data") or {}
check("家庭健康视图", "cycle" in hv and "privacy_note" in hv)

print("=" * 72)
print("5. M5 健康建议")
r = call("/api/health/index")
h = r.get("data") or {}
check("建议总览 4 类", len(h.get("groups") or []) == 4, ",".join(g.get("key") for g in (h.get("groups") or [])))
check("建议含免责声明", "仅供参考" in (h.get("disclaimer") or ""))
r = call("/api/health/detail", {"type": "diet"})
check("饮食建议列表", (r.get("data") or {}).get("total", 0) >= 1, str((r.get("data") or {}).get("total")))

print("=" * 72)
print("6. M7 两性关系")
r = call("/api/relation/index")
rel = r.get("data") or {}
check("关系总览", "assessments" in rel and "courses" in rel)
check("关系量表", len(rel.get("assessments") or []) >= 1)
r = call("/api/relation/cards")
check("沟通引导卡", (r.get("data") or {}).get("total", 0) >= 5)
r = call("/api/relation/conflicts")
check("冲突工具", (r.get("data") or {}).get("total", 0) >= 3)
r = call("/api/relation/courses")
check("关系课程", len((r.get("data") or {}).get("list") or []) >= 3)
course_id = ((r.get("data") or {}).get("list") or [{}])[0].get("id")
r = call("/api/relation/courseBuy", {"id": course_id}, "POST")
check("课程下单", bool((r.get("data") or {}).get("order_no")) or (r.get("data") or {}).get("purchased"), r.get("msg"))

r = call("/api/relation/taskSave", {"title": "一起散步 20 分钟", "category": "daily"}, "POST")
check("双人任务新增", r.get("code") == 1)
rel_task = (r.get("data") or {}).get("id", 0)
r = call("/api/relation/taskDone", {"id": rel_task}, "POST")
check("双人任务完成+记忆联动", (r.get("data") or {}).get("done") == 1)
r = call("/api/relation/memorySave", {"title": "在一起", "memory_date": "2024-05-20", "type": "anniversary"}, "POST")
check("纪念日新增", r.get("code") == 1)
mem_id = (r.get("data") or {}).get("id", 0)
r = call("/api/relation/memories")
check("纪念日列表含倒数", any(x.get("days_to_next") is not None for x in (r.get("data") or {}).get("list") or []))
r = call("/api/relation/memoryRemove", {"id": mem_id}, "POST")
check("纪念日删除", r.get("code") == 1)

print("=" * 72)
print("7. M8 亲子教育")
r = call("/api/parenting/babySave", {"nickname": "小豆", "gender": 2, "birthday": str(today - datetime.timedelta(days=200))}, "POST")
check("宝宝档案保存", (r.get("data") or {}).get("month_age") is not None, str((r.get("data") or {}).get("month_age")))
r = call("/api/parenting/index")
par = r.get("data") or {}
check("亲子总览", bool(par.get("baby")) and "vaccines" in par)
check("月龄计算", isinstance(par.get("month_age"), int), str(par.get("month_age")))
r = call("/api/parenting/growthSave", {"record_date": str(today), "height": 66.5, "weight": 7.8, "milestone": "会翻身"}, "POST")
check("成长记录", (r.get("code") == 1), r.get("msg"))
r = call("/api/parenting/activities")
act = r.get("data") or {}
check("亲子活动按月龄", len(act.get("list") or []) >= 1, f"{len(act.get('list') or [])} 个")
r = call("/api/parenting/coTaskSave", {"title": "周三晚哄睡", "category": "care"}, "POST")
check("共育任务新增", r.get("code") == 1)
co_id = (r.get("data") or {}).get("id", 0)
r = call("/api/parenting/coTaskDone", {"id": co_id}, "POST")
check("共育任务完成", (r.get("data") or {}).get("done") == 1)
r = call("/api/parenting/emotions")
check("育儿情绪内容", (r.get("data") or {}).get("total", 0) >= 3)
r = call("/api/parenting/vaccines")
vac = (r.get("data") or {}).get("list") or []
check("疫苗排期按生日推算", len(vac) >= 10 and any(v.get("due_date") for v in vac), f"{len(vac)} 项")
r = call("/api/parenting/courses")
check("育儿课程", len((r.get("data") or {}).get("list") or []) >= 3)

print("=" * 72)
print("8. M9 心理咨询（预约表单）+ 心理测评")
r = call("/api/consult/index")
con = r.get("data") or {}
check("咨询总览", "categories" in con and "bookings" in con and "assessments" in con)
check("咨询分类 11 个", len(con.get("categories") or []) == 11, f"{len(con.get('categories') or [])} 个")
check("危机信息含热线", "12356" in ((con.get("crisis") or {}).get("text") or ""))

# 清理历史待处理预约，保证可重复执行
for b in (con.get("bookings") or []):
    if b.get("status") in ("pending", "contacted"):
        call("/api/consult/bookingCancel", {"id": b["id"]}, "POST")

r = call("/api/consult/booking", {
    "name": "小满", "mobile": "13900002222", "category": "emotion",
    "prefer_time": "工作日 19:00-21:00", "description": "最近工作压力大，睡眠不好，想找人聊聊",
}, "POST")
booking = r.get("data") or {}
check("提交咨询预约", r.get("code") == 1 and bool(booking.get("id")), r.get("msg"))
check("预约分类中文", booking.get("category") == "情绪疏导", str(booking.get("category")))
booking_id = booking.get("id", 0)

r = call("/api/consult/bookings")
blist = (r.get("data") or {}).get("list") or []
check("我的预约列表", any(x.get("id") == booking_id for x in blist), f"{len(blist)} 条")
if blist:
    check("手机号脱敏", "****" in (blist[0].get("mobile") or ""))

r = call("/api/consult/bookingCancel", {"id": booking_id}, "POST")
check("取消预约", r.get("code") == 1, r.get("msg"))

r = call("/api/consult/assessments", {"scope": "mood"})
alist = (r.get("data") or {}).get("list") or []
codes = {a.get("code"): a for a in alist}
check("量表列表含 SDS", "dep_sds" in codes)
check("量表列表含 SAS", "anx_sas" in codes)
check("量表列表含 16PF", "pf16" in codes)
sds = codes.get("dep_sds") or {}
check("SDS 需付费 29.9", sds.get("need_pay") is True and abs((sds.get("price") or 0) - 29.9) < 0.01, f"¥{sds.get('price')}")
check("SDS 题数 20", sds.get("questions") == 20, str(sds.get("questions")))

r = call("/api/consult/assessmentDetail", {"id": sds.get("id")})
detail = r.get("data") or {}
check("SDS 题库加载", len(detail.get("items") or []) == 20 and len(detail.get("options") or []) == 4)

# 免费量表（PMS）计分仍正常
pms = next((a for a in alist if a.get("code") == "pms"), None)
check("免费量表 PMS 存在", pms is not None)
if pms:
    r = call("/api/consult/assessmentSubmit", {"id": pms["id"], "answers": json.dumps([0, 1, 1, 0, 1])}, "POST")
    res = r.get("data") or {}
    check("PMS 计分 3", res.get("score") == 3, str(res.get("score")))
    check("PMS 分级 low", res.get("result") == "low", str(res.get("result")))

# 产后情绪高风险 → 危机干预
pdp = next((a for a in alist if a.get("code") == "pdp"), None)
if pdp:
    r = call("/api/consult/assessmentSubmit", {"id": pdp["id"], "answers": json.dumps([3, 3, 3, 3, 3])}, "POST")
    res = r.get("data") or {}
    check("高风险触发危机干预", res.get("risk_flag") == 1 and bool((res.get("crisis") or {}).get("hotlines")))
    check("危机文案含热线", "12356" in ((res.get("crisis") or {}).get("text") or ""))

print("=" * 72)
print("9. M6 会员 / 设置 / 数据备份")
r = call("/api/member/plans")
plans = (r.get("data") or {}).get("list") or []
check("会员套餐 3 档", len(plans) == 3)
r = call("/api/member/status")
check("会员状态", "member" in (r.get("data") or {}))
r = call("/api/setting/save", {"app_lock": "password", "app_lock_code": "1234", "local_only": 1,
                               "remind_period": 1, "remind_time": "20:30"}, "POST")
check("设置保存", r.get("code") == 1, r.get("msg"))
r = call("/api/setting/index")
check("设置回读", (r.get("data") or {}).get("has_lock_code") is True)

r = call("/api/setting/backupCreate", method="POST")
bk = r.get("data") or {}
check("创建备份", bool(bk.get("id")) and (bk.get("size") or 0) > 0, f"{bk.get('size')} bytes")
backup_id = bk.get("id", 0)
r = call("/api/setting/backupList")
blist = (r.get("data") or {}).get("list") or []
check("备份列表", len(blist) >= 1, f"{len(blist)} 份")
r = call("/api/setting/backupRestore", {"id": backup_id}, "POST")
check("备份恢复", r.get("code") == 1, r.get("msg"))

print("=" * 72)
print("10. 内容与通用能力回归")
r = call("/api/content/index", {"type": "care", "limit": 5})
check("话术内容可读", (r.get("data") or {}).get("total", 0) >= 5)
r = call("/api/content/index", {"type": "course", "part": "relation", "limit": 10})
check("课程按 part 过滤", (r.get("data") or {}).get("total", 0) >= 3)
r = call("/api/user/detail")
check("个人中心聚合", "summary" in (r.get("data") or {}))

print("=" * 72)
print(f"通过 {len(PASSED)} 项，失败 {len(FAILED)} 项")
if FAILED:
    for f in FAILED:
        print("  ❌ " + f)
    sys.exit(1)
print("🎉 M1–M9 全链路验证通过")