998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
150 lines
5.8 KiB
Python
150 lines
5.8 KiB
Python
#!/usr/bin/env python
|
||
"""
|
||
测试统计分析API
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, '/Users/nongjun/Desktop/Ai公司/本地开发与测试/kaopeilian-backend')
|
||
|
||
from app.services.statistics_service import StatisticsService
|
||
from app.core.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
# 测试数据库连接字符串（使用本地测试环境）
|
||
DATABASE_URL = "mysql+aiomysql://root:nj861021@localhost:3306/kaopeilian"
|
||
|
||
|
||
async def test_statistics_service():
|
||
"""测试统计服务"""
|
||
print("=" * 50)
|
||
print("开始测试统计分析服务")
|
||
print("=" * 50)
|
||
|
||
# 创建数据库连接
|
||
engine = create_async_engine(DATABASE_URL, echo=False)
|
||
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
||
try:
|
||
async with async_session() as session:
|
||
# 测试用户ID（使用实际存在的用户）
|
||
test_user_id = 2 # admin用户
|
||
|
||
print(f"\n📊 测试用户ID: {test_user_id}")
|
||
print("-" * 50)
|
||
|
||
# 1. 测试关键指标
|
||
print("\n1️⃣ 测试关键指标...")
|
||
try:
|
||
metrics = await StatisticsService.get_key_metrics(
|
||
db=session,
|
||
user_id=test_user_id,
|
||
period="month"
|
||
)
|
||
print("✓ 关键指标获取成功:")
|
||
print(f" - 学习效率: {metrics['learningEfficiency']['value']}%")
|
||
print(f" - 知识覆盖率: {metrics['knowledgeCoverage']['value']}%")
|
||
print(f" - 平均用时: {metrics['avgTimePerQuestion']['value']} 分/题")
|
||
print(f" - 进步速度: {metrics['progressSpeed']['value']}%")
|
||
except Exception as e:
|
||
print(f"✗ 关键指标获取失败: {e}")
|
||
|
||
# 2. 测试成绩分布
|
||
print("\n2️⃣ 测试成绩分布...")
|
||
try:
|
||
distribution = await StatisticsService.get_score_distribution(
|
||
db=session,
|
||
user_id=test_user_id,
|
||
period="month"
|
||
)
|
||
print("✓ 成绩分布获取成功:")
|
||
print(f" - 优秀: {distribution['excellent']}")
|
||
print(f" - 良好: {distribution['good']}")
|
||
print(f" - 中等: {distribution['medium']}")
|
||
print(f" - 及格: {distribution['pass']}")
|
||
print(f" - 不及格: {distribution['fail']}")
|
||
except Exception as e:
|
||
print(f"✗ 成绩分布获取失败: {e}")
|
||
|
||
# 3. 测试难度分析
|
||
print("\n3️⃣ 测试难度分析...")
|
||
try:
|
||
difficulty = await StatisticsService.get_difficulty_analysis(
|
||
db=session,
|
||
user_id=test_user_id,
|
||
period="month"
|
||
)
|
||
print("✓ 难度分析获取成功:")
|
||
for key, value in difficulty.items():
|
||
print(f" - {key}: {value}%")
|
||
except Exception as e:
|
||
print(f"✗ 难度分析获取失败: {e}")
|
||
|
||
# 4. 测试知识点掌握度
|
||
print("\n4️⃣ 测试知识点掌握度...")
|
||
try:
|
||
mastery = await StatisticsService.get_knowledge_mastery(
|
||
db=session,
|
||
user_id=test_user_id
|
||
)
|
||
print(f"✓ 知识点掌握度获取成功 (共{len(mastery)}个知识点):")
|
||
for item in mastery[:3]: # 只显示前3个
|
||
print(f" - {item['name']}: {item['mastery']}%")
|
||
except Exception as e:
|
||
print(f"✗ 知识点掌握度获取失败: {e}")
|
||
|
||
# 5. 测试学习时长统计
|
||
print("\n5️⃣ 测试学习时长统计...")
|
||
try:
|
||
time_stats = await StatisticsService.get_study_time_stats(
|
||
db=session,
|
||
user_id=test_user_id,
|
||
period="week"
|
||
)
|
||
print(f"✓ 学习时长统计获取成功:")
|
||
print(f" - 日期数: {len(time_stats['labels'])}")
|
||
print(f" - 总学习时长: {sum(time_stats['studyTime'])} 小时")
|
||
print(f" - 总练习时长: {sum(time_stats['practiceTime'])} 小时")
|
||
except Exception as e:
|
||
print(f"✗ 学习时长统计获取失败: {e}")
|
||
|
||
# 6. 测试详细数据
|
||
print("\n6️⃣ 测试详细数据...")
|
||
try:
|
||
detail = await StatisticsService.get_detail_data(
|
||
db=session,
|
||
user_id=test_user_id,
|
||
period="month"
|
||
)
|
||
print(f"✓ 详细数据获取成功 (共{len(detail)}条记录):")
|
||
if detail:
|
||
first = detail[0]
|
||
print(f" - 最近日期: {first['date']}")
|
||
print(f" - 考试次数: {first['examCount']}")
|
||
print(f" - 平均分: {first['avgScore']}")
|
||
print(f" - 正确率: {first['accuracy']}%")
|
||
except Exception as e:
|
||
print(f"✗ 详细数据获取失败: {e}")
|
||
|
||
print("\n" + "=" * 50)
|
||
print("✅ 所有测试完成!")
|
||
print("=" * 50)
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 测试过程中发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("\n🚀 启动统计分析服务测试...")
|
||
asyncio.run(test_statistics_service())
|
||
|