293 lines
13 KiB
Python
293 lines
13 KiB
Python
|
"""审计日志 API —— 查看系统操作审计记录
|
|||
|
|
|
|||
|
|
与 MOM(KCGL) /audit/logs 的接口保持同构的筛选维度（操作人/模块/动作/目标/
|
|||
|
|
时间区间），便于两端运维习惯统一；额外提供 request_id 筛选，可凭它直接跳到
|
|||
|
|
结构化日志里的那一次请求。
|
|||
|
|
|
|||
|
|
另提供两个 CSV 导出端点（审计明细 / 日活统计），均支持按列导出。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import csv
|
|||
|
|
import io
|
|||
|
|
from datetime import datetime, time, timedelta
|
|||
|
|
from typing import Any, Callable
|
|||
|
|
|
|||
|
|
from fastapi import APIRouter, Depends, Query, Response
|
|||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
|
|
|||
|
|
from app.core.database import get_db
|
|||
|
|
from app.core.deps import require_admin
|
|||
|
|
from app.core.time_utils import BEIJING_TZ, get_beijing_time
|
|||
|
|
from app.schemas.audit import (
|
|||
|
|
AuditLogListResponse,
|
|||
|
|
AuditLogResponse,
|
|||
|
|
AuditOption,
|
|||
|
|
AuditOptionsResponse,
|
|||
|
|
DailyUsageResponse,
|
|||
|
|
DailyUsageRow,
|
|||
|
|
)
|
|||
|
|
from app.services import audit_service
|
|||
|
|
from app.services.audit_service import ACTION_LABELS, MODULE_LABELS
|
|||
|
|
|
|||
|
|
router = APIRouter(prefix="/audit", tags=["审计日志"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_day(value: str | None, *, end_of_day: bool = False) -> datetime | None:
|
|||
|
|
"""解析 YYYY-MM-DD 为北京时间。
|
|||
|
|
|
|||
|
|
结束日期取次日 00:00 作为上界（配合 < 判断）—— 直接取当天 23:59:59 会
|
|||
|
|
漏掉该秒内的记录，是日期区间筛选最常见的差一错误。
|
|||
|
|
"""
|
|||
|
|
if not value:
|
|||
|
|
return None
|
|||
|
|
try:
|
|||
|
|
day = datetime.strptime(value, "%Y-%m-%d").date()
|
|||
|
|
except ValueError:
|
|||
|
|
return None
|
|||
|
|
if end_of_day:
|
|||
|
|
return datetime.combine(day + timedelta(days=1), time.min, tzinfo=BEIJING_TZ)
|
|||
|
|
return datetime.combine(day, time.min, tzinfo=BEIJING_TZ)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/logs", response_model=AuditLogListResponse)
|
|||
|
|
async def get_audit_logs(
|
|||
|
|
user_id: str | None = Query(None, description="操作人账号（模糊匹配）"),
|
|||
|
|
module: str | None = Query(None, description="业务模块"),
|
|||
|
|
action: str | None = Query(None, description="动作类型"),
|
|||
|
|
target_id: str | None = Query(None, description="目标ID"),
|
|||
|
|
request_id: str | None = Query(None, description="请求ID（与接口日志对账）"),
|
|||
|
|
status_code: int | None = Query(None, description="响应状态码"),
|
|||
|
|
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD"),
|
|||
|
|
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（含当天）"),
|
|||
|
|
page: int = Query(1, ge=1),
|
|||
|
|
page_size: int = Query(50, ge=1, le=200),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
current_user: dict = Depends(require_admin),
|
|||
|
|
) -> AuditLogListResponse:
|
|||
|
|
"""审计日志分页查询（按时间倒序）"""
|
|||
|
|
start = _parse_day(start_date)
|
|||
|
|
# 结束日期用「次日 00:00」作为开区间上界，避免漏掉当天最后几条
|
|||
|
|
end_exclusive = _parse_day(end_date, end_of_day=True)
|
|||
|
|
|
|||
|
|
rows, total = await audit_service.list_audit_logs(
|
|||
|
|
db,
|
|||
|
|
user_id=user_id,
|
|||
|
|
module=module,
|
|||
|
|
action=action,
|
|||
|
|
target_id=target_id,
|
|||
|
|
request_id=request_id,
|
|||
|
|
status_code=status_code,
|
|||
|
|
start=start,
|
|||
|
|
end=end_exclusive - timedelta(microseconds=1) if end_exclusive else None,
|
|||
|
|
skip=(page - 1) * page_size,
|
|||
|
|
limit=page_size,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
items = []
|
|||
|
|
for row in rows:
|
|||
|
|
item = AuditLogResponse.model_validate(row)
|
|||
|
|
# 中文标签由服务端补，避免前端为每个枚举再维护一份映射
|
|||
|
|
item.module_label = MODULE_LABELS.get(row.module, row.module)
|
|||
|
|
item.action_label = ACTION_LABELS.get(row.action, row.action)
|
|||
|
|
items.append(item)
|
|||
|
|
|
|||
|
|
return AuditLogListResponse(items=items, total=total)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/options", response_model=AuditOptionsResponse)
|
|||
|
|
async def get_audit_options(
|
|||
|
|
current_user: dict = Depends(require_admin),
|
|||
|
|
) -> AuditOptionsResponse:
|
|||
|
|
"""筛选项：模块与动作的中文下拉；顺带下发导出可选列"""
|
|||
|
|
return AuditOptionsResponse(
|
|||
|
|
modules=[AuditOption(value=k, label=v) for k, v in MODULE_LABELS.items()],
|
|||
|
|
actions=[AuditOption(value=k, label=v) for k, v in ACTION_LABELS.items()],
|
|||
|
|
log_export_columns=[
|
|||
|
|
AuditOption(value=k, label=v[0]) for k, v in _AUDIT_LOG_COLUMNS.items()
|
|||
|
|
],
|
|||
|
|
usage_export_columns=[
|
|||
|
|
AuditOption(value=k, label=v[0]) for k, v in _DAILY_USAGE_COLUMNS.items()
|
|||
|
|
],
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================================================
|
|||
|
|
# CSV 导出
|
|||
|
|
# ============================================================
|
|||
|
|
|
|||
|
|
def _bj(dt: datetime | None) -> str:
|
|||
|
|
"""时间列统一按北京时间输出（与列表页、日活分日口径一致）。
|
|||
|
|
|
|||
|
|
直接输出 UTC 会让导出文件里 01:00 的操作显示成前一天 17:00，
|
|||
|
|
与网页上看到的对不上 —— 导出与页面不一致是最容易被质疑的那种问题。
|
|||
|
|
"""
|
|||
|
|
if dt is None:
|
|||
|
|
return ""
|
|||
|
|
return dt.astimezone(BEIJING_TZ).strftime("%Y-%m-%d %H:%M:%S")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _actor(log) -> str:
|
|||
|
|
"""操作人：优先中文名，退化为账号（与列表页的展示规则一致）"""
|
|||
|
|
if not log.user_id and not log.display_name:
|
|||
|
|
return "未认证"
|
|||
|
|
return f"{log.display_name}({log.user_id})" if log.display_name else (log.user_id or "")
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 列定义：key → (表头, 取值函数)。
|
|||
|
|
# 前端只传 key 列表，中文表头与取值口径都由后端统一维护，
|
|||
|
|
# 避免两端各写一份导致"导出的列和页面上的对不上"。
|
|||
|
|
_AUDIT_LOG_COLUMNS: dict[str, tuple[str, Callable[[Any], Any]]] = {
|
|||
|
|
"time": ("时间", lambda r: _bj(r.created_at)),
|
|||
|
|
"user": ("操作人", _actor),
|
|||
|
|
"role": ("角色", lambda r: r.role or ""),
|
|||
|
|
"module": ("模块", lambda r: MODULE_LABELS.get(r.module, r.module)),
|
|||
|
|
"action": ("动作", lambda r: ACTION_LABELS.get(r.action, r.action)),
|
|||
|
|
"method": ("方法", lambda r: r.method or ""),
|
|||
|
|
"url": ("请求路径", lambda r: r.url or ""),
|
|||
|
|
"status": ("结果", lambda r: r.status_code if r.status_code is not None else ""),
|
|||
|
|
"ip": ("来源IP", lambda r: r.ip_address or ""),
|
|||
|
|
"target": ("目标", lambda r: f"{r.target_type or ''}:{r.target_id or ''}".strip(":")),
|
|||
|
|
"error": ("错误信息", lambda r: r.error_message or ""),
|
|||
|
|
"request_id": ("请求ID", lambda r: r.request_id or ""),
|
|||
|
|
"user_agent": ("User-Agent", lambda r: r.user_agent or ""),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
_DAILY_USAGE_COLUMNS: dict[str, tuple[str, Callable[[dict], Any]]] = {
|
|||
|
|
"day": ("日期", lambda r: r["day"]),
|
|||
|
|
"user": ("操作人", lambda r: f"{r['display_name']}({r['user_id']})" if r["display_name"] else (r["user_id"] or "")),
|
|||
|
|
"role": ("角色", lambda r: r["role"] or ""),
|
|||
|
|
# 上线/下线时间 = 当天首次/末次活动（非登录时间），
|
|||
|
|
# 登录/登出次数单独成列，两者不再混为一谈
|
|||
|
|
"first_active": ("上线时间", lambda r: _bj(r["first_active_at"])),
|
|||
|
|
"last_active": ("下线时间", lambda r: _bj(r["last_active_at"])),
|
|||
|
|
"login_count": ("登录次数", lambda r: r["login_count"]),
|
|||
|
|
"logout_count": ("登出次数", lambda r: r["logout_count"]),
|
|||
|
|
"op_count": ("操作次数", lambda r: r["op_count"]),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _csv_response(
|
|||
|
|
columns: dict[str, tuple[str, Callable]], keys: list[str], rows: list, filename: str,
|
|||
|
|
) -> Response:
|
|||
|
|
"""把行数据渲染成 CSV 响应。
|
|||
|
|
|
|||
|
|
⚠️ 必须带 UTF-8 BOM：Excel 靠它识别编码，否则中文表头与内容全是乱码。
|
|||
|
|
这是 CSV 导出最常见、也最容易被忽略的坑。
|
|||
|
|
"""
|
|||
|
|
buf = io.StringIO()
|
|||
|
|
writer = csv.writer(buf)
|
|||
|
|
writer.writerow([columns[k][0] for k in keys])
|
|||
|
|
for row in rows:
|
|||
|
|
writer.writerow([columns[k][1](row) for k in keys])
|
|||
|
|
|
|||
|
|
return Response(
|
|||
|
|
content=b"\xef\xbb\xbf" + buf.getvalue().encode("utf-8"),
|
|||
|
|
media_type="text/csv; charset=utf-8",
|
|||
|
|
# 文件名用纯 ASCII：中文文件名要走 RFC 5987，各浏览器行为不一致，
|
|||
|
|
# 内部系统没必要为它引入兼容成本。
|
|||
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _resolve_keys(raw: str | None, columns: dict) -> list[str]:
|
|||
|
|
"""解析前端传来的列 key。缺省 = 全部列；未知 key 直接忽略（不报错）。"""
|
|||
|
|
if not raw:
|
|||
|
|
return list(columns)
|
|||
|
|
keys = [k.strip() for k in raw.split(",") if k.strip() in columns]
|
|||
|
|
return keys or list(columns)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/logs/export")
|
|||
|
|
async def export_audit_logs(
|
|||
|
|
user_id: str | None = Query(None, description="操作人账号（模糊匹配）"),
|
|||
|
|
module: str | None = Query(None, description="业务模块"),
|
|||
|
|
action: str | None = Query(None, description="动作类型"),
|
|||
|
|
target_id: str | None = Query(None, description="目标ID"),
|
|||
|
|
request_id: str | None = Query(None, description="请求ID"),
|
|||
|
|
status_code: int | None = Query(None, description="响应状态码"),
|
|||
|
|
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD"),
|
|||
|
|
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（含当天）"),
|
|||
|
|
columns: str | None = Query(None, description="导出列，逗号分隔；缺省=全部"),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
current_user: dict = Depends(require_admin),
|
|||
|
|
) -> Response:
|
|||
|
|
"""审计明细 CSV 导出 —— 筛选维度与 /logs 完全一致，保证"看到什么就能导出什么"。"""
|
|||
|
|
start = _parse_day(start_date)
|
|||
|
|
end_exclusive = _parse_day(end_date, end_of_day=True)
|
|||
|
|
|
|||
|
|
rows, truncated = await audit_service.export_audit_logs(
|
|||
|
|
db,
|
|||
|
|
user_id=user_id, module=module, action=action, target_id=target_id,
|
|||
|
|
request_id=request_id, status_code=status_code,
|
|||
|
|
start=start,
|
|||
|
|
end=end_exclusive - timedelta(microseconds=1) if end_exclusive else None,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
keys = _resolve_keys(columns, _AUDIT_LOG_COLUMNS)
|
|||
|
|
resp = _csv_response(_AUDIT_LOG_COLUMNS, keys, rows, "audit_logs.csv")
|
|||
|
|
if truncated:
|
|||
|
|
# 用响应头传递"已截断"，前端据此提示用户收窄筛选条件
|
|||
|
|
resp.headers["X-Export-Truncated"] = "1"
|
|||
|
|
resp.headers["X-Export-Max-Rows"] = str(audit_service.EXPORT_MAX_ROWS)
|
|||
|
|
resp.headers["Access-Control-Expose-Headers"] = "X-Export-Truncated, X-Export-Max-Rows"
|
|||
|
|
return resp
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/daily-usage/export")
|
|||
|
|
async def export_daily_usage(
|
|||
|
|
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD（北京时间），默认今天"),
|
|||
|
|
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（北京时间），默认同起始日"),
|
|||
|
|
columns: str | None = Query(None, description="导出列，逗号分隔；缺省=全部"),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
current_user: dict = Depends(require_admin),
|
|||
|
|
) -> Response:
|
|||
|
|
"""日活统计 CSV 导出 —— 每人一行：上线/下线次数与时间、操作次数。"""
|
|||
|
|
start = _parse_day(start_date) or datetime.combine(
|
|||
|
|
get_beijing_time().date(), time.min, tzinfo=BEIJING_TZ,
|
|||
|
|
)
|
|||
|
|
end = _parse_day(end_date, end_of_day=True) or (start + timedelta(days=1))
|
|||
|
|
|
|||
|
|
items = await audit_service.get_daily_usage(db, start=start, end=end)
|
|||
|
|
keys = _resolve_keys(columns, _DAILY_USAGE_COLUMNS)
|
|||
|
|
return _csv_response(_DAILY_USAGE_COLUMNS, keys, items, "daily_usage.csv")
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/daily-usage", response_model=DailyUsageResponse)
|
|||
|
|
async def get_daily_usage(
|
|||
|
|
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD（北京时间），默认今天"),
|
|||
|
|
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（北京时间），默认同起始日"),
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
current_user: dict = Depends(require_admin),
|
|||
|
|
) -> DailyUsageResponse:
|
|||
|
|
"""日活 / 使用统计 —— 按【北京时间自然日 × 操作人】聚合。
|
|||
|
|
|
|||
|
|
回答的是「每天有哪些人用了系统、用了多少」：
|
|||
|
|
· 上线时间 / 下线时间：当天**首次 / 末次活动**时间（任意审计记录）
|
|||
|
|
· 操作次数：当天该用户的全部审计记录数（使用深度）
|
|||
|
|
· 登录次数 / 登出次数：真实的手动登录 / 登出行为计数
|
|||
|
|
|
|||
|
|
⚠️ 上线时间【不取登录时间】：token 有效期内（refresh 7 天）用户不重新登录，
|
|||
|
|
按登录算会让「周一登录、周二继续用」的周二变成"登录次数 0、上线时间空，
|
|||
|
|
但操作次数 35"——报表自相矛盾。改用活动口径后，当天的第一次操作即上线时间。
|
|||
|
|
|
|||
|
|
⚠️ 登出次数天然小于登录次数：用户直接关浏览器、断网、token 过期都不会
|
|||
|
|
产生登出记录。这是真实情况，不做任何"补齐"推算。
|
|||
|
|
"""
|
|||
|
|
# 起始日：未传则取北京的今天。_parse_day 返回的是北京时间当日 00:00。
|
|||
|
|
start = _parse_day(start_date) or datetime.combine(
|
|||
|
|
get_beijing_time().date(), time.min, tzinfo=BEIJING_TZ,
|
|||
|
|
)
|
|||
|
|
# 结束日：_parse_day(end_of_day=True) 已给出「次日 00:00」，正好当作半开上界。
|
|||
|
|
# 未传则默认单日查询（= 起始日当天）。
|
|||
|
|
end = _parse_day(end_date, end_of_day=True) or (start + timedelta(days=1))
|
|||
|
|
|
|||
|
|
items = await audit_service.get_daily_usage(db, start=start, end=end)
|
|||
|
|
|
|||
|
|
return DailyUsageResponse(
|
|||
|
|
start_date=start.astimezone(BEIJING_TZ).strftime("%Y-%m-%d"),
|
|||
|
|
end_date=(end - timedelta(days=1)).astimezone(BEIJING_TZ).strftime("%Y-%m-%d"),
|
|||
|
|
items=[DailyUsageRow(**row) for row in items],
|
|||
|
|
total=len(items),
|
|||
|
|
)
|