H3ConuMS-v2/backend/app/api/v1/audit.py

"""审计日志 API"""
from datetime import datetime
from typing import Optional
from fastapi import APIRouter, Depends, Query
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.middleware.permission_middleware import require_permission
from app.models.audit_log import AuditLog
from app.services.audit_service import query_logs
import csv
import io

router = APIRouter(prefix="/api/audit", tags=["审计日志"])


@router.get("/logs")
def get_audit_logs(
    start_time: Optional[datetime] = Query(None),
    end_time: Optional[datetime] = Query(None),
    user_id: Optional[str] = Query(None),
    username: Optional[str] = Query(None),
    action_type: Optional[str] = Query(None),
    resource_type: Optional[str] = Query(None),
    status: Optional[str] = Query(None),
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    db: Session = Depends(get_db),
    _: dict = Depends(require_permission('*')),
):
    """查询审计日志（仅管理员）"""
    total, items = query_logs(
        db,
        start_time=start_time,
        end_time=end_time,
        user_id=user_id,
        username=username,
        action_type=action_type,
        resource_type=resource_type,
        status=status,
        page=page,
        page_size=page_size,
    )
    return {
        "total": total,
        "page": page,
        "page_size": page_size,
        "items": [_fmt(r) for r in items],
    }


@router.get("/logs/{log_id}")
def get_audit_log_detail(
    log_id: int,
    db: Session = Depends(get_db),
    _: dict = Depends(require_permission('*')),
):
    """获取单条审计日志详情"""
    log = db.query(AuditLog).filter(AuditLog.id == log_id).first()
    if not log:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="日志不存在")
    return _fmt(log, detail=True)


@router.get("/stats")
def get_audit_stats(
    days: int = Query(7, ge=1, le=90),
    db: Session = Depends(get_db),
    _: dict = Depends(require_permission('*')),
):
    """审计日志统计（最近N天）"""
    from datetime import timedelta
    from sqlalchemy import func
    since = datetime.utcnow() - timedelta(days=days)
    rows = (
        db.query(AuditLog.action_type, AuditLog.status, func.count().label("cnt"))
        .filter(AuditLog.action_time >= since)
        .group_by(AuditLog.action_type, AuditLog.status)
        .all()
    )
    total = db.query(func.count(AuditLog.id)).filter(AuditLog.action_time >= since).scalar()
    by_type = {}
    for row in rows:
        if row.action_type not in by_type:
            by_type[row.action_type] = {"success": 0, "failed": 0, "error": 0}
        by_type[row.action_type][row.status] = row.cnt
    return {"total": total, "days": days, "by_type": by_type}


@router.get("/logs/export/csv")
def export_audit_logs(
    start_time: Optional[datetime] = Query(None),
    end_time: Optional[datetime] = Query(None),
    action_type: Optional[str] = Query(None),
    status: Optional[str] = Query(None),
    db: Session = Depends(get_db),
    _: dict = Depends(require_permission('*')),
):
    """导出审计日志为 CSV"""
    _, items = query_logs(
        db,
        start_time=start_time,
        end_time=end_time,
        action_type=action_type,
        status=status,
        page=1,
        page_size=5000,
    )

    def generate():
        buf = io.StringIO()
        writer = csv.writer(buf)
        writer.writerow(["时间", "用户", "角色", "操作类型", "子类型", "路径", "状态码", "状态", "IP", "描述"])
        for r in items:
            writer.writerow([
                r.action_time.strftime("%Y-%m-%d %H:%M:%S") if r.action_time else "",
                r.username, r.user_role, r.action_type, r.action_subtype or "",
                f"{r.request_method} {r.request_path}", r.status_code, r.status,
                r.ip_address or "", r.description,
            ])
        yield buf.getvalue().encode("utf-8-sig")

    filename = f"audit_{datetime.now().strftime('%Y%m%d%H%M%S')}.csv"
    return StreamingResponse(
        generate(),
        media_type="text/csv",
        headers={"Content-Disposition": f"attachment; filename={filename}"},
    )


def _fmt(r: AuditLog, detail: bool = False) -> dict:
    base = {
        "id": r.id,
        "action_time": r.action_time.isoformat() if r.action_time else None,
        "user_id": r.user_id,
        "username": r.username,
        "user_role": r.user_role,
        "action_type": r.action_type,
        "action_subtype": r.action_subtype,
        "request_method": r.request_method,
        "request_path": r.request_path,
        "status": r.status,
        "status_code": r.status_code,
        "resource_type": r.resource_type,
        "resource_id": r.resource_id,
        "resource_name": r.resource_name,
        "description": r.description,
        "ip_address": r.ip_address,
    }
    if detail:
        base["request_params"] = r.request_params
        base["response_data"] = r.response_data
        base["error_message"] = r.error_message
        base["user_agent"] = r.user_agent
    return base