Neo-ZQYY/scripts/ops/api_health_check.py

# -*- coding: utf-8 -*-
"""API 健康检查脚本

任务 1.3：登录获取 JWT → 验证任务注册表 → 执行 sync-check
"""

import json
import sys

import requests

BASE_URL = "http://localhost:8000"
ADMIN_USER = "admin"
ADMIN_PASS = "admin123"


def main() -> int:
    ok = True

    # ── 1. 登录获取 JWT ──────────────────────────────────────
    print("=" * 60)
    print("[1/3] POST /api/auth/login — 登录获取 JWT")
    print("=" * 60)
    try:
        resp = requests.post(
            f"{BASE_URL}/api/auth/login",
            json={"username": ADMIN_USER, "password": ADMIN_PASS},
            timeout=10,
        )
    except requests.ConnectionError:
        print("✗ 无法连接后端服务，请确认 uvicorn 已在 :8000 启动")
        return 1

    if resp.status_code != 200:
        print(f"✗ 登录失败: HTTP {resp.status_code}")
        print(f"  响应: {resp.text[:500]}")
        return 1

    tokens = resp.json()
    jwt = tokens["access_token"]
    print(f"✓ 登录成功，获取 JWT（前 40 字符）: {jwt[:40]}...")
    print(f"  token_type: {tokens['token_type']}")
    print()

    headers = {"Authorization": f"Bearer {jwt}"}

    # ── 2. 获取任务注册表 ────────────────────────────────────
    print("=" * 60)
    print("[2/3] GET /api/tasks/registry — 验证任务注册表")
    print("=" * 60)
    resp = requests.get(f"{BASE_URL}/api/tasks/registry", headers=headers, timeout=10)

    if resp.status_code != 200:
        print(f"✗ 获取注册表失败: HTTP {resp.status_code}")
        print(f"  响应: {resp.text[:500]}")
        ok = False
    else:
        data = resp.json()
        groups = data.get("groups", {})
        total_tasks = sum(len(tasks) for tasks in groups.values())
        common_tasks = sum(
            1 for tasks in groups.values() for t in tasks if t.get("is_common")
        )

        if total_tasks == 0:
            print("✗ 任务注册表为空！")
            ok = False
        else:
            print(f"✓ 任务注册表非空")
            print(f"  业务域数量: {len(groups)}")
            print(f"  总任务数:   {total_tasks}")
            print(f"  常用任务数: {common_tasks}")
            print(f"  业务域列表: {', '.join(sorted(groups.keys()))}")
            # 按域打印任务数
            for domain in sorted(groups.keys()):
                tasks = groups[domain]
                n_common = sum(1 for t in tasks if t.get("is_common"))
                print(f"    {domain}: {len(tasks)} 个任务（{n_common} 个常用）")
    print()

    # ── 3. Sync-Check ────────────────────────────────────────
    print("=" * 60)
    print("[3/3] GET /api/tasks/sync-check — 后端与 ETL 注册表同步检查")
    print("=" * 60)
    resp = requests.get(f"{BASE_URL}/api/tasks/sync-check", headers=headers, timeout=30)

    if resp.status_code != 200:
        print(f"✗ sync-check 请求失败: HTTP {resp.status_code}")
        print(f"  响应: {resp.text[:500]}")
        ok = False
    else:
        sc = resp.json()
        if sc.get("error"):
            print(f"⚠ sync-check 返回错误: {sc['error']}")
            ok = False
        elif sc.get("in_sync"):
            print("✓ 后端注册表与 ETL 真实注册表完全同步 (in_sync=true)")
        else:
            print("✗ 后端注册表与 ETL 真实注册表不同步 (in_sync=false)")
            if sc.get("backend_only"):
                print(f"  仅后端有（ETL 缺失）: {sc['backend_only']}")
            if sc.get("etl_only"):
                print(f"  仅 ETL 有（后端缺失）: {sc['etl_only']}")
            ok = False
    print()

    # ── 汇总 ─────────────────────────────────────────────────
    print("=" * 60)
    if ok:
        print("✓ API 健康检查全部通过")
    else:
        print("✗ API 健康检查存在问题，请查看上方详情")
    print("=" * 60)

    return 0 if ok else 1


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