Neo
6f8f12314f
包含多个会话的累积代码变更： - backend: AI 聊天服务、触发器调度、认证增强、WebSocket、调度器最小间隔 - admin-web: ETL 状态页、任务管理、调度配置、登录优化 - miniprogram: 看板页面、聊天集成、UI 组件、导航更新 - etl: DWS 新任务（finance_area_daily/board_cache）、连接器增强 - tenant-admin: 项目初始化 - db: 19 个迁移脚本（etl_feiqiu 11 + zqyy_app 8） - packages/shared: 枚举和工具函数更新 - tools: 数据库工具、报表生成、健康检查 - docs: PRD/架构/部署/合约文档更新 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
# -*- 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())
|