44 lines
1.3 KiB
Python
44 lines
1.3 KiB
Python
"""审计日志写入助手（复用 audit_log 表，参照 service/customer_agent/audit.py 范式）。
|
|
|
|
工作台审计留痕统一走本模块：敏感信息查看、签约/结束服务等合规动作全程可追溯。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from utils.request_id import get_request_id
|
|
|
|
|
|
async def write_audit(
|
|
db: AsyncSession,
|
|
*,
|
|
user_id: int | None,
|
|
username: str | None,
|
|
module: str,
|
|
action: str,
|
|
target: str | None = None,
|
|
detail: dict | None = None,
|
|
status: str = "成功",
|
|
) -> None:
|
|
"""插入一条审计日志并提交（trace_id 自动取当前请求链路）。"""
|
|
await db.execute(
|
|
text(
|
|
"INSERT INTO audit_log (user_id, username, module, action, target, detail, trace_id, status) "
|
|
"VALUES (:user_id, :username, :module, :action, :target, :detail, :trace_id, :status)"
|
|
),
|
|
{
|
|
"user_id": user_id,
|
|
"username": username,
|
|
"module": module,
|
|
"action": action,
|
|
"target": target,
|
|
"detail": json.dumps(detail, ensure_ascii=False) if detail is not None else None,
|
|
"trace_id": get_request_id(),
|
|
"status": status,
|
|
},
|
|
)
|
|
await db.commit()
|