66 lines
3.4 KiB
Python
66 lines
3.4 KiB
Python
"""操作日志与统计 API（仅管理员）。"""
|
|
import csv
|
|
import io
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..deps import get_current_user
|
|
from ..models import OperationLog
|
|
|
|
router = APIRouter(prefix="/api/operations", tags=["operations"])
|
|
|
|
|
|
def _require_admin(user: dict):
|
|
if not user.get("is_admin"):
|
|
raise HTTPException(status_code=403, detail="需要管理员权限")
|
|
|
|
|
|
def _query_logs(db: Session, days: int, username: str = "", action: str = ""):
|
|
since = datetime.utcnow() - timedelta(days=max(1, min(days, 365)))
|
|
query = db.query(OperationLog).filter(OperationLog.created_at >= since)
|
|
if username:
|
|
query = query.filter(OperationLog.username == username)
|
|
if action:
|
|
query = query.filter(OperationLog.action == action)
|
|
return query
|
|
|
|
|
|
@router.get("/summary")
|
|
async def operation_summary(days: int = Query(30, ge=1, le=365), user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
_require_admin(user)
|
|
query = _query_logs(db, days)
|
|
total = query.count()
|
|
action_counts = dict(query.with_entities(OperationLog.action, func.count()).group_by(OperationLog.action).all())
|
|
users = query.with_entities(OperationLog.username, func.count()).group_by(OperationLog.username).order_by(func.count().desc()).limit(10).all()
|
|
apps = query.filter(OperationLog.app_name.isnot(None)).with_entities(OperationLog.app_name, func.count()).group_by(OperationLog.app_name).order_by(func.count().desc()).limit(10).all()
|
|
return {"days": days, "total": total, "action_counts": action_counts,
|
|
"top_users": [{"name": name, "count": count} for name, count in users],
|
|
"top_apps": [{"name": name, "count": count} for name, count in apps]}
|
|
|
|
|
|
@router.get("")
|
|
async def list_operations(days: int = Query(30, ge=1, le=365), username: str = "", action: str = "", limit: int = Query(100, ge=1, le=500), user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
_require_admin(user)
|
|
logs = _query_logs(db, days, username, action).order_by(OperationLog.created_at.desc()).limit(limit).all()
|
|
return [{"id": log.id, "created_at": log.created_at, "username": log.username,
|
|
"action": log.action, "app_name": log.app_name, "task_id": log.task_id,
|
|
"resource_type": log.resource_type, "resource_name": log.resource_name,
|
|
"detail": json.loads(log.detail_json or "{}"), "client_ip": log.client_ip} for log in logs]
|
|
|
|
|
|
@router.get("/export")
|
|
async def export_operations(days: int = Query(30, ge=1, le=365), user: dict = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
_require_admin(user)
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow(["时间", "用户", "操作", "App", "资源类型", "资源名称", "IP", "详情"])
|
|
for log in _query_logs(db, days).order_by(OperationLog.created_at.desc()).all():
|
|
writer.writerow([log.created_at, log.username, log.action, log.app_name or "", log.resource_type or "", log.resource_name or "", log.client_ip or "", log.detail_json or ""])
|
|
return StreamingResponse(iter(["\ufeff" + output.getvalue()]), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=operation-logs.csv"})
|