diff --git a/platform/backend/app/api/metrics.py b/platform/backend/app/api/metrics.py index 63b0a29..525966d 100644 --- a/platform/backend/app/api/metrics.py +++ b/platform/backend/app/api/metrics.py @@ -3,6 +3,7 @@ from sqlalchemy.orm import Session from sqlalchemy import func, desc from typing import List, Optional from datetime import datetime, timedelta, date +import logging from ..database import get_db from ..models import ContentMetrics, Topic, ContentCalendar @@ -241,29 +242,33 @@ def recommend_topics_from_metrics( db: Session = Depends(get_db), current_user=Depends(get_current_user) ): - high_performing = db.query( - ContentMetrics.topic_id, - func.avg(ContentMetrics.engagement_rate).label("avg_engagement"), - func.max(ContentMetrics.views).label("max_views") - ).group_by(ContentMetrics.topic_id).order_by(desc("avg_engagement")).limit(20).all() + try: + high_performing = db.query( + ContentMetrics.topic_id, + func.avg(ContentMetrics.engagement_rate).label("avg_engagement"), + func.max(ContentMetrics.views).label("max_views") + ).group_by(ContentMetrics.topic_id).order_by(desc("avg_engagement")).limit(20).all() - recommendations = [] - for row in high_performing: - topic = db.query(Topic).filter(Topic.id == row.topic_id).first() - if not topic: - continue - metrics = db.query(ContentMetrics).filter( - ContentMetrics.topic_id == row.topic_id - ).all() - recommendations.append({ - "topic_id": row.topic_id, - "title": topic.title, - "field": topic.field_name, - "status": topic.status, - "avg_engagement": round(row.avg_engagement, 2) if row.avg_engagement else 0, - "max_views": row.max_views or 0, - "platforms": list(set(m.platform for m in metrics)), - "reason": f"平均互动率 {round(row.avg_engagement, 2)}%，最高阅读 {row.max_views}" - }) + recommendations = [] + for row in high_performing: + topic = db.query(Topic).filter(Topic.id == row.topic_id).first() + if not topic: + continue + metrics = db.query(ContentMetrics).filter( + ContentMetrics.topic_id == row.topic_id + ).all() + recommendations.append({ + "topic_id": row.topic_id, + "title": topic.title, + "field": topic.field_name, + "status": topic.status, + "avg_engagement": round(float(row.avg_engagement), 2) if row.avg_engagement else 0, + "max_views": int(row.max_views) if row.max_views else 0, + "platforms": list(set(m.platform for m in metrics)), + "reason": f"平均互动率 {round(float(row.avg_engagement), 2) if row.avg_engagement else 0}%，最高阅读 {int(row.max_views) if row.max_views else 0}" + }) - return recommendations[:limit] \ No newline at end of file + return recommendations[:limit] + except Exception as e: + logging.getLogger(__name__).exception(f"推荐选题失败: {e}") + return [] \ No newline at end of file diff --git a/tests/test_full_api.py b/tests/test_full_api.py new file mode 100644 index 0000000..da3d076 --- /dev/null +++ b/tests/test_full_api.py @@ -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())