feat(audit): 日活使用统计 + 双 CSV 导出（审计/上下线，列可自定义）

【日活统计：GET /audit/daily-usage】按【北京时间自然日 × 操作人】聚合，单个 GROUP BY 完成（count(*) FILTER），不用窗口函数。指标：上线/下线时间、操作次数、登录/登出次数。 ⚠️ 上线/下线时间取【当天首次/末次活动】，刻意不取登录时间： Refresh Token 有效期 7 天，用户不必每天重新登录。按登录算会出现「登录次数 0、上线时间空，但操作次数 35」——报表自相矛盾。时间一律按 +08:00 分日与渲染，否则早班（00:00~08:00）操作会掉到前一天。【CSV 导出：两个端点 + 列自定义】 - /audit/logs/export 整体审计导出，筛选维度与列表页完全一致 - /audit/daily-usage/export 上下线导出，每人一行 - 列清单由后端统一维护并经 /audit/options 下发（log_export_columns / usage_export_columns），前端不硬编码表头，避免两端漂移 - ⚠️ 响应带 UTF-8 BOM：Excel 靠它识别编码，否则中文表头全乱码 - 单次上限 5 万行，超出经 X-Export-Truncated 头告知前端明确提示（静默截断比报错更危险） - list_audit_logs 与 export_audit_logs 共用 _log_filters，保证「看到的」与「导出的」永远是同一批数据【前端】 - 审计页页头新增「人员统计」「导出 CSV」两个按钮，现有表格与筛选零改动 - 人员统计走抽屉（AuditUsagePanel）：日期范围+快捷键、日活表格、上下线次数彩色标签、北京时间渲染、导出前弹列勾选面板 - ExportColumnsModal 为两处导出共用，默认全选
2026-09-21 12:06:43 +08:00
parent 5290d83463
commit 1fea30b03b
7 changed files with 825 additions and 17 deletions
--- a/backend/app/services/audit_service.py
+++ b/backend/app/services/audit_service.py
@ -18,7 +18,7 @@ import logging
 import uuid
 from datetime import datetime

-from sqlalchemy import func, select
+from sqlalchemy import and_, func, select
 from sqlalchemy.ext.asyncio import AsyncSession

 from app.core.database import AsyncSessionLocal
@ -165,6 +165,54 @@ async def list_audit_logs(

    真实总数走独立 COUNT —— 前端分页器依赖它，不能用 len(当前页)。
    """
+    filters = _log_filters(
+        user_id=user_id, module=module, action=action, target_id=target_id,
+        request_id=request_id, status_code=status_code, start=start, end=end,
+    )
+
+    total = await db.scalar(
+        select(func.count()).select_from(AuditLog).where(*filters)
+    ) or 0
+
+    rows = (
+        await db.execute(
+            select(AuditLog)
+            .where(*filters)
+            .order_by(AuditLog.created_at.desc())
+            .offset(skip)
+            .limit(limit)
+        )
+    ).scalars().all()
+
+    return list(rows), total
+
+
+# ============================================================
+# 导出
+# ============================================================
+
+# 单次导出的行数上限。审计表只增不减，全量导出迟早会撑爆内存与浏览器，
+# 故设硬上限；超出时向上层返回 truncated=True，由前端明确提示「已截断」——
+# 静默截断会让使用者以为导全了，比报错更危险。
+EXPORT_MAX_ROWS = 50000
+
+
+def _log_filters(
+    *,
+    user_id: str | None = None,
+    module: str | None = None,
+    action: str | None = None,
+    target_id: str | None = None,
+    request_id: str | None = None,
+    status_code: int | None = None,
+    start: datetime | None = None,
+    end: datetime | None = None,
+) -> list:
+    """审计日志的筛选条件 —— list_audit_logs 与 export_audit_logs 共用。
+
+    抽出来的唯一目的：保证「列表看到的」和「导出出去的」永远是同一批数据。
+    两处各写一份迟早会漂移，而导出与列表不一致是最让人不信任的那种 bug。
+    """
    filters = []
    if user_id:
        filters.append(AuditLog.user_id.ilike(f"%{user_id}%"))
@ -182,19 +230,110 @@ async def list_audit_logs(
        filters.append(AuditLog.created_at >= start)
    if end:
        filters.append(AuditLog.created_at <= end)
+    return filters

-    total = await db.scalar(
-        select(func.count()).select_from(AuditLog).where(*filters)
-    ) or 0

+async def export_audit_logs(
+    db: AsyncSession, *, limit: int = EXPORT_MAX_ROWS, **kwargs,
+) -> tuple[list[AuditLog], bool]:
+    """导出用：按筛选条件取全部记录（不分页）。返回 (rows, truncated)。
+
+    多取一行来判断是否被截断 —— 比再跑一次 COUNT 便宜。
+    """
    rows = (
        await db.execute(
            select(AuditLog)
-            .where(*filters)
+            .where(*_log_filters(**kwargs))
            .order_by(AuditLog.created_at.desc())
-            .offset(skip)
-            .limit(limit)
+            .limit(limit + 1)
        )
    ).scalars().all()
+    truncated = len(rows) > limit
+    return list(rows[:limit]), truncated

-    return list(rows), total
+
+# ============================================================
+# 日活 / 使用统计
+# ============================================================
+
+# 成功 = 2xx/3xx。登录失败（401）也要留痕，但不应计入"上线次数"。
+_OK_STATUS_UPPER = 400
+
+
+async def get_daily_usage(
+    db: AsyncSession, *, start: datetime, end: datetime,
+) -> list[dict]:
+    """按【北京时间自然日 × 操作人】聚合用量 —— 日活报表的数据源。
+
+    start/end 为半开区间 [start, end)，调用方按北京时间日界传入。
+
+    全部指标由**一个 GROUP BY 查询**算出，不用窗口函数：
+      · 登录/登出次数 = 成功登录 / 成功登出数（最终凭证是 login_count，不是"上线次数"）
+      · 操作频次 = 当天该用户的全部审计记录数（代表系统使用深度）
+      · 上线/下线时间 = 当天**首次 / 末次活动**时间（任意审计记录）
+
+    ⚠️ 上线/下线时间【不能】取登录/登出时间。
+       Access/Refresh Token 有效期内（refresh 7 天）用户无需重新登录，
+       于是"周一登录、周二到周日继续用"会导致周二~周日：
+       登录次数=0、登录时间=空，但操作次数却是几十 —— 报表自相矛盾。
+       改用活动口径后，工人当天的第一次操作就是真实上线时间。
+       （审计中间件是全站的，移动端的接收/转交/完工同样入账，故自动覆盖移动端，
+         无需前端上报心跳。）
+
+    为什么用 `count(*) FILTER (WHERE ...)`：分组内一次扫描同时算出多个条件计数，
+    比多次子查询或 UNION 简单得多，且语义一眼可读。Postgres 原生支持。
+
+    ⚠️ 按【北京时间】分日：created_at 是 timestamptz（实存 UTC），
+       直接按 UTC 分日会让 00:00~08:00 的早班操作掉到前一天。
+    """
+    day_col = func.date(func.timezone("Asia/Shanghai", AuditLog.created_at))
+
+    login_ok = and_(
+        AuditLog.action == "login", AuditLog.status_code < _OK_STATUS_UPPER,
+    )
+    logout_ok = and_(
+        AuditLog.action == "logout", AuditLog.status_code < _OK_STATUS_UPPER,
+    )
+
+    stmt = (
+        select(
+            day_col.label("day"),
+            AuditLog.user_id.label("user_id"),
+            # 同一用户的 display_name / role 是一致的，取 max 只是为了
+            # 在 GROUP BY 下拿到一个非空代表值（避免再套一层 DISTINCT ON）
+            func.max(AuditLog.display_name).label("display_name"),
+            func.max(AuditLog.role).label("role"),
+            func.count().filter(login_ok).label("login_count"),
+            func.count().filter(logout_ok).label("logout_count"),
+            func.count().label("op_count"),
+            # 上线/下线时间取「任意记录」的首末，而不是登录/登出的首末（原因见 docstring）
+            func.min(AuditLog.created_at).label("first_active_at"),
+            func.max(AuditLog.created_at).label("last_active_at"),
+        )
+        .where(
+            AuditLog.created_at >= start,
+            AuditLog.created_at < end,
+            # 只统计"人"：未认证请求（如登录前的探测、refresh）没有操作人，
+            # 混进来会让"日活人数"虚高。若要排查匿名异常流量，走日志列表页按
+            # 结果/来源 IP 过滤更合适。
+            AuditLog.user_id.isnot(None),
+        )
+        .group_by(day_col, AuditLog.user_id)
+        .order_by(day_col.desc(), func.count().desc())
+    )
+
+    rows = (await db.execute(stmt)).all()
+    return [
+        {
+            "day": r.day.strftime("%Y-%m-%d") if hasattr(r.day, "strftime") else str(r.day),
+            "user_id": r.user_id,
+            "display_name": r.display_name,
+            "role": r.role,
+            "login_count": r.login_count or 0,
+            "logout_count": r.logout_count or 0,
+            "op_count": r.op_count or 0,
+            "first_active_at": r.first_active_at,
+            "last_active_at": r.last_active_at,
+        }
+        for r in rows
+    ]