240 lines
7.3 KiB
Python
240 lines
7.3 KiB
Python
"""历史记录管理路由，提供调用历史查询、统计和导出功能"""
|
||
|
||
from fastapi import APIRouter, HTTPException, status, Depends
|
||
from typing import List, Dict, Any, Optional
|
||
from datetime import datetime
|
||
import json
|
||
|
||
from app.services.history_manager import history_manager
|
||
from app.models.database import get_db
|
||
from app.routes.user import get_current_active_user
|
||
|
||
router = APIRouter(prefix="/history", tags=["history"])
|
||
|
||
|
||
@router.get("/user-calls")
|
||
async def get_user_call_history(
|
||
algorithm_id: Optional[str] = None,
|
||
start_date: Optional[str] = None,
|
||
end_date: Optional[str] = None,
|
||
status: Optional[str] = None,
|
||
skip: int = 0,
|
||
limit: int = 100,
|
||
current_user: dict = Depends(get_current_active_user),
|
||
db = Depends(get_db)
|
||
):
|
||
"""获取用户的调用历史"""
|
||
# 解析日期参数
|
||
start_dt = None
|
||
end_dt = None
|
||
if start_date:
|
||
try:
|
||
start_dt = datetime.fromisoformat(start_date.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
raise HTTPException(status_code=400, detail="Invalid start_date format")
|
||
|
||
if end_date:
|
||
try:
|
||
end_dt = datetime.fromisoformat(end_date.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
raise HTTPException(status_code=400, detail="Invalid end_date format")
|
||
|
||
# 普通用户只能查看自己的历史，管理员可以查看所有用户历史
|
||
user_id = current_user.get("id")
|
||
if current_user.get("role") == "admin":
|
||
# 管理员可以指定用户ID，否则查看所有用户
|
||
user_id = None # 这样会返回所有用户的记录
|
||
|
||
history = history_manager.get_user_call_history(
|
||
db=db,
|
||
user_id=user_id or current_user.get("id"),
|
||
algorithm_id=algorithm_id,
|
||
start_date=start_dt,
|
||
end_date=end_dt,
|
||
status=status,
|
||
skip=skip,
|
||
limit=limit
|
||
)
|
||
|
||
return {
|
||
"history": [call.__dict__ for call in history],
|
||
"count": len(history),
|
||
"skip": skip,
|
||
"limit": limit
|
||
}
|
||
|
||
|
||
@router.get("/algorithm-calls/{algorithm_id}")
|
||
async def get_algorithm_call_history(
|
||
algorithm_id: str,
|
||
user_id: Optional[str] = None,
|
||
start_date: Optional[str] = None,
|
||
end_date: Optional[str] = None,
|
||
status: Optional[str] = None,
|
||
skip: int = 0,
|
||
limit: int = 100,
|
||
current_user: dict = Depends(get_current_active_user),
|
||
db = Depends(get_db)
|
||
):
|
||
"""获取特定算法的调用历史"""
|
||
# 验证权限：用户必须有权访问该算法
|
||
# 在实际实现中，这里应该检查用户是否有权访问该算法
|
||
# 为简化，我们只检查是否为管理员或查看自己的记录
|
||
|
||
# 解析日期参数
|
||
start_dt = None
|
||
end_dt = None
|
||
if start_date:
|
||
try:
|
||
start_dt = datetime.fromisoformat(start_date.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
raise HTTPException(status_code=400, detail="Invalid start_date format")
|
||
|
||
if end_date:
|
||
try:
|
||
end_dt = datetime.fromisoformat(end_date.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
raise HTTPException(status_code=400, detail="Invalid end_date format")
|
||
|
||
history = history_manager.get_algorithm_call_history(
|
||
db=db,
|
||
algorithm_id=algorithm_id,
|
||
user_id=user_id,
|
||
start_date=start_dt,
|
||
end_date=end_dt,
|
||
status=status,
|
||
skip=skip,
|
||
limit=limit
|
||
)
|
||
|
||
return {
|
||
"history": [call.__dict__ for call in history],
|
||
"count": len(history),
|
||
"skip": skip,
|
||
"limit": limit
|
||
}
|
||
|
||
|
||
@router.get("/statistics")
|
||
async def get_call_statistics(
|
||
user_id: Optional[str] = None,
|
||
algorithm_id: Optional[str] = None,
|
||
current_user: dict = Depends(get_current_active_user),
|
||
db = Depends(get_db)
|
||
):
|
||
"""获取调用统计信息"""
|
||
# 权限检查
|
||
if current_user.get("role") != "admin":
|
||
# 普通用户只能查看自己的统计
|
||
if user_id and user_id != current_user.get("id"):
|
||
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
||
user_id = current_user.get("id")
|
||
|
||
stats = history_manager.get_call_statistics(
|
||
db=db,
|
||
user_id=user_id,
|
||
algorithm_id=algorithm_id
|
||
)
|
||
|
||
return stats
|
||
|
||
|
||
@router.post("/compare")
|
||
async def get_comparison_data(
|
||
call_ids: List[str],
|
||
current_user: dict = Depends(get_current_active_user),
|
||
db = Depends(get_db)
|
||
):
|
||
"""获取用于对比的历史数据"""
|
||
# 权限检查：用户只能对比自己的调用记录
|
||
# 获取调用记录
|
||
calls = db.query(AlgorithmCall).filter(AlgorithmCall.id.in_(call_ids)).all()
|
||
|
||
# 检查权限：用户只能对比自己的记录
|
||
for call in calls:
|
||
if call.user_id != current_user.get("id") and current_user.get("role") != "admin":
|
||
raise HTTPException(status_code=403, detail="Insufficient permissions to access call data")
|
||
|
||
comparison_data = history_manager.get_comparison_data(db, call_ids)
|
||
|
||
return {
|
||
"comparison_data": comparison_data,
|
||
"count": len(comparison_data)
|
||
}
|
||
|
||
|
||
@router.get("/export")
|
||
async def export_history(
|
||
algorithm_id: Optional[str] = None,
|
||
start_date: Optional[str] = None,
|
||
end_date: Optional[str] = None,
|
||
format_type: str = "json",
|
||
current_user: dict = Depends(get_current_active_user),
|
||
db = Depends(get_db)
|
||
):
|
||
"""导出历史记录"""
|
||
# 解析日期参数
|
||
start_dt = None
|
||
end_dt = None
|
||
if start_date:
|
||
try:
|
||
start_dt = datetime.fromisoformat(start_date.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
raise HTTPException(status_code=400, detail="Invalid start_date format")
|
||
|
||
if end_date:
|
||
try:
|
||
end_dt = datetime.fromisoformat(end_date.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
raise HTTPException(status_code=400, detail="Invalid end_date format")
|
||
|
||
file_path = history_manager.export_history(
|
||
db=db,
|
||
user_id=current_user.get("id"),
|
||
algorithm_id=algorithm_id,
|
||
start_date=start_dt,
|
||
end_date=end_dt,
|
||
format_type=format_type
|
||
)
|
||
|
||
if file_path:
|
||
return {
|
||
"success": True,
|
||
"file_path": file_path,
|
||
"download_url": f"/api/files/download/{file_path}",
|
||
"message": "History exported successfully"
|
||
}
|
||
else:
|
||
raise HTTPException(status_code=500, detail="Failed to export history")
|
||
|
||
|
||
@router.delete("/cleanup")
|
||
async def cleanup_old_history(
|
||
days_old: int,
|
||
algorithm_id: Optional[str] = None,
|
||
current_user: dict = Depends(get_current_active_user),
|
||
db = Depends(get_db)
|
||
):
|
||
"""清理旧的历史记录"""
|
||
# 只有管理员可以清理历史记录
|
||
if current_user.get("role") != "admin":
|
||
raise HTTPException(status_code=403, detail="Only administrators can clean up history")
|
||
|
||
# 确保天数为正数
|
||
if days_old <= 0:
|
||
raise HTTPException(status_code=400, detail="days_old must be positive")
|
||
|
||
deleted_count = history_manager.delete_old_history(
|
||
db=db,
|
||
days_old=days_old,
|
||
algorithm_id=algorithm_id
|
||
)
|
||
|
||
return {
|
||
"message": f"Cleaned up {deleted_count} old history records",
|
||
"deleted_count": deleted_count
|
||
}
|
||
|
||
|
||
# 导入需要的模型
|
||
from app.models.models import AlgorithmCall |