ad2d27cd72
问题：同一套「字段名 → 中文」映射存了三份手工同步的副本 ——
前端 AuditLog.vue 的 fieldMap（约 95 个字段）、后端 api/v1/audit.py 的
ACTION_ALIASES，另有若干内联的 status 映射散落在 service 层。
改一边漏一边就会漂移，页面上冒出英文列名或对不上的中文。
改动：
· 新增 app/utils/audit_labels.py 作为**唯一来源**，统一提供
操作类型归一化（canon_action）、字段名中文化（field_label）、
码值中文化（enum_label / bool_label / person_name_label）、
ID 指向登记（id_ref_of）。
· api/v1/audit.py 改为从该模块 import，删掉本地副本。
· 新增 GET /audit/labels 下发映射（静态标签，免权限码；审计页本身
已由 system_audit 把关）。
· 前端 AuditLog.vue 删除本地 95 行 fieldMap，改从接口拉取；
拉取失败时未命中项原样显示，是可控降级。
★ 码值必须**按模块**翻译：同名 status 在出库/借还/报废里含义完全不同
（出库的 3 是「已出库」，报废的 3 是「已执行(已报废)」），
不区分模块会把报废单显示成已出库。借还模块下还混着字符串状态
（borrowed/returned），因其两张表共用 status 列名。
故映射按 (模块, 字段) 匹配，未登记再退回字段名。
★ ID 字段同理：parent_id 在「系统管理」里是菜单的上级菜单，在「BOM管理」
里是物料节点 —— 指向完全不同的表。
验证：前端 vue-tsc 改前改后均为 565 个错误、去除行号后**报错集合完全相同**，
未引入新错误；GET /audit/labels 实测返回 200。
197 lines
8.0 KiB
Python
197 lines
8.0 KiB
Python
# inventory-backend/app/api/v1/audit.py
|
||
from flask import Blueprint, request, jsonify, current_app
|
||
from flask_jwt_extended import jwt_required, get_jwt
|
||
from app.utils.decorators import permission_required
|
||
from app.models.audit import AuditLog
|
||
from app.extensions import db
|
||
from sqlalchemy import or_
|
||
from datetime import datetime
|
||
import json
|
||
|
||
audit_bp = Blueprint('audit', __name__)
|
||
|
||
|
||
# =============================================================================
|
||
# 操作类型归一化 / 中文化标签
|
||
#
|
||
# ★ 映射表已统一收敛到 app/utils/audit_labels.py（**唯一来源**）。
|
||
# 本模块与日报服务共用同一份，前端经 GET /audit/labels 拉取同一份 ——
|
||
# 改一处即可，不会再出现"两边手工同步、改一边漏一边"的漂移。
|
||
#
|
||
# 历史问题（迁移前）：前端 AuditLog.vue 自带一份 fieldMap，后端这里自带一份
|
||
# ACTION_ALIASES，两份副本各自演化。
|
||
# =============================================================================
|
||
from app.utils.audit_labels import ( # noqa: E402
|
||
ACTION_ALIASES,
|
||
canon_action,
|
||
labels_payload,
|
||
)
|
||
|
||
|
||
@audit_bp.route('/logs', methods=['GET'])
|
||
@jwt_required()
|
||
@permission_required('system_audit')
|
||
def get_audit_logs():
|
||
"""获取审计日志列表（分页）"""
|
||
try:
|
||
# 分页参数
|
||
page = request.args.get('page', 1, type=int)
|
||
page_size = request.args.get('pageSize', 50, type=int)
|
||
|
||
# 筛选参数
|
||
username = request.args.get('username', '').strip()
|
||
module = request.args.get('module', '').strip()
|
||
action = request.args.get('action', '').strip()
|
||
target_id = request.args.get('target_id', '').strip()
|
||
start_date = request.args.get('start_date', '').strip()
|
||
end_date = request.args.get('end_date', '').strip()
|
||
|
||
# ★ 操作人类型：all(默认) / user(仅真实用户) / system(仅系统)
|
||
#
|
||
# 背景：改造前的全局监听器没有请求上下文守卫，系统初始化与后台定时任务
|
||
# 产生了大量 username='system' 的日志（历史存量约 1.8 万条），会把列表刷屏。
|
||
# 新监听器已加守卫不再产生此类记录，但存量数据仍需要能筛掉。
|
||
operator_type = request.args.get('operator_type', 'all').strip().lower()
|
||
if operator_type not in ('all', 'user', 'system'):
|
||
operator_type = 'all'
|
||
|
||
# 构建查询
|
||
query = AuditLog.query
|
||
|
||
if operator_type == 'user':
|
||
# 真实用户：排除 system 占位账号
|
||
query = query.filter(AuditLog.username != 'system')
|
||
elif operator_type == 'system':
|
||
query = query.filter(AuditLog.username == 'system')
|
||
|
||
if username:
|
||
query = query.filter(AuditLog.username.like(f'%{username}%'))
|
||
if module:
|
||
query = query.filter(AuditLog.module == module)
|
||
if action:
|
||
# ★ 兼容历史别名：先把任意写法（中文/小写）归一化为规范值，
|
||
# 再展开为该值的全部等价写法一起匹配。
|
||
# 否则选「新增」只能搜到早期中文 action 的数据（约 3-4 月），
|
||
# 会让用户误以为"没有最近的内容"。
|
||
canon = canon_action(action)
|
||
if canon in ACTION_ALIASES:
|
||
query = query.filter(AuditLog.action.in_(ACTION_ALIASES[canon]))
|
||
else:
|
||
query = query.filter(AuditLog.action == action)
|
||
if target_id:
|
||
query = query.filter(AuditLog.target_id == target_id)
|
||
if start_date:
|
||
try:
|
||
start_dt = datetime.strptime(start_date, '%Y-%m-%d')
|
||
query = query.filter(AuditLog.created_at >= start_dt)
|
||
except ValueError:
|
||
pass
|
||
if end_date:
|
||
try:
|
||
end_dt = datetime.strptime(end_date, '%Y-%m-%d')
|
||
# 包含当天结束时间
|
||
from datetime import timedelta
|
||
end_dt = end_dt + timedelta(days=1)
|
||
query = query.filter(AuditLog.created_at < end_dt)
|
||
except ValueError:
|
||
pass
|
||
|
||
# 【行级数据隔离】通过操作人 username 关联用户表过滤公司
|
||
from app.utils.decorators import get_current_company_filter
|
||
from app.models.system import SysUser
|
||
from sqlalchemy import func
|
||
company_limit = get_current_company_filter()
|
||
if company_limit is not None:
|
||
query = query.join(SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username) \
|
||
.filter(SysUser.department == company_limit)
|
||
|
||
# 排序
|
||
query = query.order_by(AuditLog.created_at.desc())
|
||
|
||
# 分页
|
||
pagination = query.paginate(page=page, per_page=page_size, error_out=False)
|
||
logs = pagination.items
|
||
|
||
# 序列化
|
||
data = [log.to_dict() for log in logs]
|
||
|
||
# 获取可用的模块和操作类型（同公司范围内）
|
||
modules_query = db.session.query(AuditLog.module).distinct()
|
||
actions_query = db.session.query(AuditLog.action).distinct()
|
||
if company_limit is not None:
|
||
modules_query = modules_query.join(SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username) \
|
||
.filter(SysUser.department == company_limit)
|
||
actions_query = actions_query.join(SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username) \
|
||
.filter(SysUser.department == company_limit)
|
||
modules = [m[0] for m in modules_query.all() if m[0]]
|
||
|
||
# ★ action 下拉项归一化：把 CREATE/create/新增 等别名合并为一个规范值，
|
||
# 避免下拉框出现「CREATE」与「新增」两个语义重复的选项。
|
||
actions = sorted({canon_action(a[0]) for a in actions_query.all() if a[0]})
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'msg': '获取成功',
|
||
'data': {
|
||
'list': data,
|
||
'total': pagination.total,
|
||
'page': page,
|
||
'pageSize': page_size,
|
||
'modules': modules,
|
||
'actions': actions
|
||
}
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取审计日志失败: {str(e)}")
|
||
return jsonify({'code': 500, 'msg': f'服务器内部错误: {str(e)}'}), 500
|
||
|
||
|
||
@audit_bp.route('/logs/<int:log_id>', methods=['GET'])
|
||
@jwt_required()
|
||
def get_audit_log_detail(log_id):
|
||
"""获取单条审计日志详情"""
|
||
try:
|
||
log = AuditLog.query.get(log_id)
|
||
if not log:
|
||
return jsonify({'code': 404, 'msg': '日志不存在'}), 404
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'msg': '获取成功',
|
||
'data': log.to_dict()
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取审计日志详情失败: {str(e)}")
|
||
return jsonify({'code': 500, 'msg': str(e)}), 500
|
||
|
||
|
||
@audit_bp.route('/modules', methods=['GET'])
|
||
@jwt_required()
|
||
def get_modules():
|
||
"""获取所有模块列表（用于筛选）"""
|
||
try:
|
||
modules = db.session.query(AuditLog.module).distinct().all()
|
||
modules = [m[0] for m in modules if m[0]]
|
||
return jsonify({'code': 200, 'data': modules}), 200
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取模块列表失败: {str(e)}")
|
||
return jsonify({'code': 500, 'msg': str(e)}), 500
|
||
|
||
|
||
@audit_bp.route('/labels', methods=['GET'])
|
||
@jwt_required()
|
||
def get_labels():
|
||
"""
|
||
下发操作类型与字段名的中文映射。
|
||
|
||
★ 后端是**唯一来源**，前端不再自带副本 —— 否则两份手工同步的映射必然漂移。
|
||
前端拉到之前（或拉取失败时）对未命中字段原样显示字段名，
|
||
是可控的降级，不会让页面崩掉。
|
||
|
||
免权限码：纯静态标签，不含任何业务数据；审计页本身已由
|
||
@permission_required('system_audit') 把关（见 /logs）。
|
||
"""
|
||
return jsonify({'code': 200, 'data': labels_payload()}), 200
|