PayrollManagementSystem/backend/app/services/audit_service.py

"""操作日志审计服务。

需求 2.5.1 要求记录用户的操作行为。与登录行为的分工：登录在 `login_logs`
单独记录（失败也要留痕且不依赖认证上下文），本模块记录**已认证用户的管理类
操作**。

设计要点：

- **成功与失败都留痕** —— 失败的操作（如试图删除不存在的用户）同样是审计
  对象，模型里专门有 `success` / `error_message` 两列承载它
- **审计失败不影响业务** —— 落库过程整体兜底，失败只记应用日志、绝不外抛
  （与 `auth_service._record` 同一原则："审计可丢，业务不可断"）
- **独立提交** —— 日志单独 commit，不与业务同事务。同事务时业务失败回滚会
  连带丢掉最该留痕的失败记录

用法（上下文管理器）::

    target = user_repository.get_by_id(session, user_id)   # 只读预取，供摘要
    with audit(session, operator=current_user,
               record=AuditRecord(MODULE_USER, ACTION_DELETE,
                                  f"删除用户 {target.username}"),
               meta=meta):
        ...业务操作与 session.commit()...
        return out

约定：

- 业务代码在上下文内自行 commit（事务边界仍在 service 层）。**commit 之后
  的任何数据库查询、ORM 属性读取都必须移出上下文**（提到 commit 之前或
  用已固化的局部变量）—— 业务已提交却因后置查询抛错被记成「失败」，会
  产生与事实相反的审计记录
- 上下文内不得吞掉异常后正常退出——吞掉后会被记为成功
- 摘要需要"目标实体名"时，应在进入上下文**之前**只读预取，
  这样目标不存在（NotFound 失败）时也能留下可追溯的记录
- 不支持嵌套（在 audit 内再调用会自行 commit 的服务）——进入侦测到
  未提交变更会记 warning
"""

import logging
import time
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass

from sqlalchemy.orm import Session

from app.core.exceptions import AppException
from app.models import User
from app.repositories import operation_log_repository
from app.repositories.operation_log_repository import OperationLogEntry

logger = logging.getLogger(__name__)


@dataclass(frozen=True, slots=True)
class RequestMeta:
    """HTTP 请求元信息。由路由层依赖 `RequestMetaDep` 提取后传入 service。"""

    ip: str | None = None
    user_agent: str | None = None
    method: str | None = None
    path: str | None = None


@dataclass(frozen=True, slots=True)
class AuditRecord:
    """审计的业务侧内容（模块 / 操作 / 摘要 / 请求参数）。"""

    module: str
    action: str
    summary: str
    request_data: dict | None = None


@contextmanager
def audit(
    session: Session,
    *,
    operator: User,
    record: AuditRecord,
    meta: RequestMeta | None = None,
) -> Iterator[None]:
    """审计上下文：正常退出记成功；抛异常记失败并**原样重抛**。"""
    started = time.perf_counter()
    meta = meta or RequestMeta()

    # 固化操作人为普通标量，之后不再触碰 ORM 对象 —— 失败路径先 rollback
    # 会使其全部过期，再读 id/username 将触发额外 SELECT；若账号恰被并发
    # 删除，ObjectDeletedError 还会被下面的兜底吞掉、失败审计静默丢失
    base: dict = {
        "user_id": operator.id,
        "username": operator.username,
        "module": record.module,
        "action": record.action,
        "summary": record.summary,
        "method": meta.method,
        "path": meta.path,
        "ip": meta.ip,
        "user_agent": meta.user_agent,
        "request_data": record.request_data,
    }

    # 进入审计时应无未提交变更（业务写操作在上下文内才发生）。出现变更
    # 意味着调用方可能嵌套了会自行 commit 的服务：失败路径的回滚将无法
    # 撤销已提交的部分，审计与数据状态会不一致
    if session.new or session.dirty or session.deleted:
        logger.warning(
            "进入审计上下文时存在未提交的变更 | module=%s | action=%s",
            record.module,
            record.action,
        )

    try:
        yield
    except Exception as exc:
        # 先回滚再写日志：业务异常可能源于 IntegrityError 等数据库错误，
        # 不回滚的话 session 处于「必须回滚」状态，写日志会抛
        # PendingRollbackError —— 最该留下的失败记录反而被丢掉
        _rollback_quietly(session)
        _safe_write(
            session,
            entry=OperationLogEntry(
                **base,
                success=False,
                error_message=_error_text(exc),
                duration_ms=_elapsed_ms(started),
            ),
        )
        raise
    else:
        _safe_write(
            session,
            entry=OperationLogEntry(**base, success=True, duration_ms=_elapsed_ms(started)),
        )


def _safe_write(session: Session, *, entry: OperationLogEntry) -> None:
    """落库并独立提交。整体兜底：任何失败只记应用日志，绝不外抛。"""
    try:
        operation_log_repository.add(session, entry)
        session.commit()
    except Exception:
        logger.warning(
            "操作日志写入失败（不影响业务） | module=%s | action=%s | summary=%s",
            entry.module,
            entry.action,
            entry.summary,
            exc_info=True,
        )
        _rollback_quietly(session)


def _rollback_quietly(session: Session) -> None:
    try:
        session.rollback()
    except Exception:
        logger.warning("会话回滚失败", exc_info=True)


def _error_text(exc: Exception) -> str:
    """失败原因文案。

    业务异常取其面向用户的消息；未预期异常返回通用文案 —— 完整堆栈由全局
    异常处理器记入应用日志，审计表不落内部细节。
    """
    if isinstance(exc, AppException):
        return exc.message
    return "服务端异常，请稍后重试"


def _elapsed_ms(started: float) -> int:
    """度量业务处理耗时（不含响应序列化）。"""
    return max(0, int((time.perf_counter() - started) * 1000))