fix(metrics): 修复 recommend-topics API 错误处理
- 添加 try-except 捕获空数据导致的聚合查询错误 - 添加 logging 导入用于错误日志记录 - 修复 avg_engagement/max_views 类型转换问题 - 新增 test_full_api.py 全功能测试脚本 - 所有 29 项 API 测试通过 (100%)
This commit is contained in:
lt
@@ -0,0 +1,169 @@
|
||||
#!/usr/bin/env python3
|
||||
"""全面功能测试脚本"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
BASE_URL = "http://localhost:8001"
|
||||
|
||||
try:
|
||||
import requests
|
||||
except ImportError:
|
||||
os.system("pip install requests -q")
|
||||
import requests
|
||||
|
||||
def get_token():
|
||||
r = requests.post(f"{BASE_URL}/api/auth/login", json={"username": "admin", "password": "admin123"})
|
||||
return r.json()["token"] if r.status_code == 200 else None
|
||||
|
||||
def h(token):
|
||||
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||||
|
||||
def check(name, condition, details=""):
|
||||
icon = "✅" if condition else "❌"
|
||||
detail_str = f" ({details})" if details else ""
|
||||
print(f" {icon} {name}{detail_str}")
|
||||
return condition
|
||||
|
||||
def section(name):
|
||||
print(f"\n{'='*50}")
|
||||
print(f" {name}")
|
||||
print(f"{'='*50}")
|
||||
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print("宇之然平台 - 全功能测试")
|
||||
print("=" * 60)
|
||||
|
||||
token = get_token()
|
||||
if not token:
|
||||
print("❌ 登录失败")
|
||||
return 1
|
||||
|
||||
print(f"🔑 登录成功 (admin)")
|
||||
results = []
|
||||
|
||||
# === 1. 选题系统 ===
|
||||
section("📋 选题系统")
|
||||
r = requests.get(f"{BASE_URL}/api/topics", headers=h(token))
|
||||
results.append(check("选题列表 API", r.ok, f"{len(r.json())} 条"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/topics/stats", headers=h(token))
|
||||
stats = r.json()
|
||||
results.append(check("选题统计", r.ok, f"共 {stats.get('total',0)} 条"))
|
||||
results.append(check("状态分布", True, f"待处理{stats.get('by_status',{}).get('待处理',0)}/待审查{stats.get('by_status',{}).get('待审查',0)}/待发布{stats.get('by_status',{}).get('待发布',0)}/已发布{stats.get('by_status',{}).get('已发布',0)}"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/topics?status=pending&limit=3", headers=h(token))
|
||||
results.append(check("状态筛选", r.ok, f"{len(r.json())} 条"))
|
||||
|
||||
# === 2. 领域配置 ===
|
||||
section("🗂️ 领域配置")
|
||||
r = requests.get(f"{BASE_URL}/api/topic-config/fields", headers=h(token))
|
||||
fields = r.json()
|
||||
results.append(check("领域列表", r.ok, f"{len(fields)} 个"))
|
||||
if fields:
|
||||
results.append(check("领域详情", True, fields[0].get("name", "")))
|
||||
|
||||
field_id = fields[0].get("id")
|
||||
r = requests.get(f"{BASE_URL}/api/topic-config/fields/{field_id}/scoring", headers=h(token))
|
||||
results.append(check("评分字段", r.ok))
|
||||
|
||||
# === 3. 内容日历 ===
|
||||
section("📅 内容日历")
|
||||
now = time.localtime()
|
||||
year, month = now.tm_year, now.tm_mon
|
||||
r = requests.get(f"{BASE_URL}/api/calendar/entries?start_date={year}-{month:02d}-01&end_date={year}-{month:02d}-31", headers=h(token))
|
||||
entries = r.json()
|
||||
results.append(check("日历条目", r.ok, f"{len(entries)} 条"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/calendar?year={year}&month={month}", headers=h(token))
|
||||
results.append(check("月度日历", r.ok))
|
||||
|
||||
# === 4. 数据分析 ===
|
||||
section("📊 数据分析")
|
||||
r = requests.get(f"{BASE_URL}/api/metrics/dashboard", headers=h(token))
|
||||
dash = r.json()
|
||||
results.append(check("Dashboard", r.ok, f"选题:{dash.get('total_topics',0)}"))
|
||||
results.append(check("状态分布", True, f"{len(dash.get('topics_by_status',{}))} 种状态"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/metrics/trend?days=30", headers=h(token))
|
||||
results.append(check("趋势数据", r.ok, f"{len(r.json())} 条"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/metrics/by-platform", headers=h(token))
|
||||
results.append(check("平台对比", r.ok, f"{len(r.json())} 个平台"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/metrics/recommend-topics", headers=h(token))
|
||||
results.append(check("选题推荐", r.ok, f"{len(r.json())} 条推荐"))
|
||||
|
||||
# === 5. 素材库 ===
|
||||
section("🖼️ 素材库")
|
||||
r = requests.get(f"{BASE_URL}/api/assets", headers=h(token))
|
||||
results.append(check("素材列表", r.ok, f"{len(r.json())} 个"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/assets/tags", headers=h(token))
|
||||
results.append(check("标签列表", r.ok, f"{len(r.json())} 个"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/assets/counts", headers=h(token))
|
||||
counts = r.json()
|
||||
results.append(check("素材统计", r.ok, f"共 {counts.get('total',0)} 个"))
|
||||
|
||||
# === 6. 创作任务 ===
|
||||
section("🚀 创作任务")
|
||||
r = requests.get(f"{BASE_URL}/api/tasks", headers=h(token))
|
||||
results.append(check("任务列表", r.ok, f"{len(r.json())} 个"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/tasks/active", headers=h(token))
|
||||
results.append(check("活跃任务", r.ok, f"{len(r.json())} 个"))
|
||||
|
||||
# === 7. 平台配置 ===
|
||||
section("🌐 平台配置")
|
||||
r = requests.get(f"{BASE_URL}/api/platform-config", headers=h(token))
|
||||
platforms = r.json()
|
||||
results.append(check("平台列表", r.ok, f"{len(platforms)} 个"))
|
||||
if platforms:
|
||||
for p in platforms:
|
||||
results.append(check(f"平台-{p['platform']}", True, p.get("name","")))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/platform-config/zhihu", headers=h(token))
|
||||
results.append(check("知乎配置", r.ok))
|
||||
|
||||
# === 8. 用户管理 ===
|
||||
section("👥 用户管理")
|
||||
r = requests.get(f"{BASE_URL}/api/admin/users", headers=h(token))
|
||||
users = r.json()
|
||||
results.append(check("用户列表", r.ok, f"{len(users)} 个"))
|
||||
|
||||
# === 9. 管理后台 ===
|
||||
section("⚙️ 系统管理")
|
||||
r = requests.get(f"{BASE_URL}/api/admin/cases", headers=h(token))
|
||||
results.append(check("案例列表", r.ok, f"{len(r.json())} 条"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/admin/llmconfigs", headers=h(token))
|
||||
results.append(check("LLM配置", r.ok, f"{len(r.json())} 条"))
|
||||
|
||||
r = requests.get(f"{BASE_URL}/api/admin/systemconfigs", headers=h(token))
|
||||
results.append(check("系统配置", r.ok, f"{len(r.json())} 条"))
|
||||
|
||||
# === 10. 系统状态 ===
|
||||
section("🔧 系统状态")
|
||||
r = requests.get(f"{BASE_URL}/api/system/status", headers=h(token))
|
||||
if r.ok:
|
||||
status = r.json()
|
||||
results.append(check("系统状态", True, f"选题:{status.get('stats',{}).get('total',0)}"))
|
||||
else:
|
||||
results.append(check("系统状态", False))
|
||||
|
||||
# === 总结 ===
|
||||
print(f"\n{'='*60}")
|
||||
passed = sum(results)
|
||||
total = len(results)
|
||||
print(f"测试结果: {passed}/{total} 通过 ({passed*100//total}%)")
|
||||
print("=" * 60)
|
||||
|
||||
return 0 if passed == total else 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user