# 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__) # ============================================================================= # 操作类型归一化 # # 问题背景：历史数据里 action 有两套写法 —— 早期装饰器（已废弃）写入中文 # （新增/修改/删除/批量删除…），现行监听器写入大写英文（CREATE/UPDATE/DELETE）。 # 前端下拉框直接取 DISTINCT action，于是同时出现「CREATE」和「新增」两个选项， # 而表格里二者又都显示为「新增」（actionMap 做了映射），用户无法分辨。 # # 后果：用户选了看得懂的中文项，只能搜到 3-4 月的历史数据，误以为"没有最近的内容"。 # # 处理：对外只暴露规范值（CREATE/UPDATE/DELETE），筛选时自动展开到全部别名， # 历史数据无需迁移即可被正确检索。 # ============================================================================= ACTION_ALIASES = { 'CREATE': ('CREATE', 'create', 'INSERT', 'insert', '新增', '批量生成'), 'UPDATE': ('UPDATE', 'update', '修改', '分配', '归还'), 'DELETE': ('DELETE', 'delete', '删除', '批量删除'), } # 反向索引：任意别名 → 规范值 _ALIAS_TO_CANON = { alias: canon for canon, aliases in ACTION_ALIASES.items() for alias in aliases } def canon_action(action): """把任意写法的 action 归一化为规范值；无法识别时原样返回""" return _ALIAS_TO_CANON.get((action or '').strip(), (action or '').strip()) @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/', 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