156 lines
5.0 KiB
Python
156 lines
5.0 KiB
Python
"""
|
||
知识库操作审计日志服务
|
||
|
||
action 枚举值
|
||
-------------
|
||
upload 上传文件到知识库
|
||
download 下载/查看文件（含 URL 预览）
|
||
delete 删除文件
|
||
archive 归档文件（预留）
|
||
create_kb 创建知识库
|
||
delete_kb 删除知识库
|
||
permission_change 修改下级用户权限（如关闭/开启上传权限）
|
||
"""
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
import asyncpg
|
||
|
||
from logger.logging import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
VALID_ACTIONS = frozenset({
|
||
"upload",
|
||
"download",
|
||
"delete",
|
||
"archive",
|
||
"create_kb",
|
||
"delete_kb",
|
||
"permission_change",
|
||
})
|
||
|
||
|
||
class AuditService:
|
||
|
||
@staticmethod
|
||
async def write(
|
||
conn: asyncpg.Connection,
|
||
*,
|
||
enterprise_id: int,
|
||
actor_id: int,
|
||
action: str,
|
||
target_user_id: Optional[int] = None,
|
||
department_id: Optional[int] = None,
|
||
kb_id: Optional[int] = None,
|
||
file_id: Optional[int] = None,
|
||
ip: Optional[str] = None,
|
||
user_agent: Optional[str] = None,
|
||
metadata: Optional[Dict[str, Any]] = None,
|
||
) -> None:
|
||
"""写入一条审计日志（非阻塞，异常仅记录不抛出）。"""
|
||
try:
|
||
import json
|
||
meta_json = json.dumps(metadata, ensure_ascii=False) if metadata else None
|
||
await conn.execute(
|
||
"""
|
||
INSERT INTO kb_audit_log
|
||
(enterprise_id, actor_id, action, target_user_id, department_id,
|
||
kb_id, file_id, ip, user_agent, metadata)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb)
|
||
""",
|
||
enterprise_id,
|
||
actor_id,
|
||
action,
|
||
target_user_id,
|
||
department_id,
|
||
kb_id,
|
||
file_id,
|
||
ip,
|
||
user_agent,
|
||
meta_json,
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"写入审计日志失败（不影响主流程）: {e}")
|
||
|
||
@staticmethod
|
||
async def list_logs(
|
||
conn: asyncpg.Connection,
|
||
enterprise_id: int,
|
||
*,
|
||
department_ids: Optional[List[int]] = None, # None 表示不过滤（admin 全量）
|
||
actor_id: Optional[int] = None,
|
||
kb_id: Optional[int] = None,
|
||
action: Optional[str] = None,
|
||
page: int = 1,
|
||
page_size: int = 20,
|
||
) -> Tuple[List[dict], int]:
|
||
"""
|
||
查询审计日志。
|
||
- department_ids=None -> 全企业（admin 用）
|
||
- department_ids=[...] -> 限制在给定部门内（leader 用，按 actor 所在部门过滤）
|
||
"""
|
||
offset = (page - 1) * page_size
|
||
conds = ["l.enterprise_id = $1"]
|
||
params: list = [enterprise_id]
|
||
|
||
if department_ids is not None:
|
||
if not department_ids:
|
||
return [], 0
|
||
conds.append(f"l.department_id = ANY(${len(params)+1}::int[])")
|
||
params.append(department_ids)
|
||
|
||
if actor_id is not None:
|
||
conds.append(f"l.actor_id = ${len(params)+1}")
|
||
params.append(actor_id)
|
||
|
||
if kb_id is not None:
|
||
conds.append(f"l.kb_id = ${len(params)+1}")
|
||
params.append(kb_id)
|
||
|
||
if action is not None:
|
||
conds.append(f"l.action = ${len(params)+1}")
|
||
params.append(action)
|
||
|
||
where = " AND ".join(conds)
|
||
count_sql = f"SELECT COUNT(*) FROM kb_audit_log l WHERE {where}"
|
||
total = await conn.fetchval(count_sql, *params)
|
||
|
||
lim_ph = len(params) + 1
|
||
off_ph = len(params) + 2
|
||
params.extend([page_size, offset])
|
||
|
||
rows = await conn.fetch(
|
||
f"""
|
||
SELECT
|
||
l.id, l.enterprise_id, l.actor_id, l.action,
|
||
l.target_user_id, l.department_id, l.kb_id, l.file_id,
|
||
l.ip, l.metadata, l.created_at,
|
||
COALESCE(NULLIF(TRIM(u.display_name),''), u.username) AS actor_name,
|
||
COALESCE(NULLIF(TRIM(tu.display_name),''), tu.username) AS target_name,
|
||
d.name AS department_name,
|
||
kb.name AS kb_name,
|
||
kf.file_name
|
||
FROM kb_audit_log l
|
||
LEFT JOIN user_list u ON u.id = l.actor_id
|
||
LEFT JOIN user_list tu ON tu.id = l.target_user_id
|
||
LEFT JOIN department d ON d.id = l.department_id
|
||
LEFT JOIN knowledge_base kb ON kb.id = l.kb_id
|
||
LEFT JOIN knowledge_base_file kf ON kf.id = l.file_id
|
||
WHERE {where}
|
||
ORDER BY l.created_at DESC
|
||
LIMIT ${lim_ph} OFFSET ${off_ph}
|
||
""",
|
||
*params,
|
||
)
|
||
import json as _json
|
||
result = []
|
||
for r in rows:
|
||
row = dict(r)
|
||
meta = row.get("metadata")
|
||
if isinstance(meta, str):
|
||
try:
|
||
row["metadata"] = _json.loads(meta)
|
||
except Exception:
|
||
row["metadata"] = None
|
||
result.append(row)
|
||
return result, int(total or 0)
|