refactor(audit): 废弃装饰器+分离架构，改为监听器单体直写入库

inventory-backend/app/core/audit_listener.py

@@ -1,15 +1,16 @@
 # inventory-backend/app/core/audit_listener.py
 """
-SQLAlchemy Event Listener 审计监听器（精简版）
+SQLAlchemy Event Listener 审计监听器（单体架构版）
-仅负责计算变更数据，存入 g.audit_details，由装饰器统一入库
+监听器亲自完成入库，不依赖 g 对象，不依赖装饰器回调。
+只要模型发生 INSERT/UPDATE/DELETE，监听器直接创建 AuditLog 并挂载到当前事务 session。
 """
 from sqlalchemy import event, inspect
-from flask import current_app, g, has_request_context
+from flask import current_app, request, has_request_context
 from datetime import datetime
 
-# 需要忽略的字段
+
 IGNORE_FIELDS = {
-    'updated_at', 'update_time', ' modified_time', 'last_modified',
+    'updated_at', 'update_time', 'modified_time', 'last_modified',
     'created_at', 'create_time', 'created_on',
 }
 
@@ -32,8 +33,8 @@ def _serialize_value(value):
 
 def _is_audit_model(mapper):
     """判断模型是否需要审计"""
-    if hasattr(mapper.class_, 'audit_enabled') and mapper.class_.audit_enabled:
+    if hasattr(mapper.class_, 'audit_enabled') and mapper.class_.audit_enabled is False:
-        return True
+        return False
 
     AUDIT_WHITELIST = {
         'MaterialBase', 'MaterialWarningSetting',
@@ -46,153 +47,132 @@ def _is_audit_model(mapper):
 
 
 def _get_module_name(mapper):
-    """获取业务模块名称"""
+    """根据模型类名推断所属模块"""
-    model_name = mapper.class_.__name__
+    name = mapper.class_.__name__
-    module_map = {
+    if 'Stock' in name or 'Buy' in name:
-        'MaterialBase': '基础信息',
+        return '入库管理'
-        'MaterialWarningSetting': '基础信息',
+    if 'Outbound' in name or 'TransOut' in name:
-        'StockBuy': '采购入库',
+        return '出库管理'
-        'StockSemi': '半成品入库',
+    if 'Borrow' in name or 'Return' in name:
-        'StockProduct': '成品入库',
+        return '借还管理'
-        'StockService': '服务权益',
+    if 'Bom' in name:
-        'RepairRecord': '维修管理',
+        return 'BOM管理'
-        'TransOutbound': '出库记录',
+    if 'StockTake' in name or 'Adjust' in name or 'Scrap' in name:
-        'TransBorrow': '借库',
+        return '盘点管理'
-        'TransReturn': '归还',
+    if 'Repair' in name:
-        'BomTable': 'BOM配方',
+        return '维修管理'
-        'StockTake': '盘点',
+    if 'SysUser' in name or 'SysMenu' in name or 'SysRole' in name:
-        'StockAdjust': '盈亏调整',
+        return '系统管理'
-        'TransScrap': '报废',
+    if 'Material' in name:
-        'SysUser': '用户管理'
+        return '基础数据'
-    }
+    return '未知模块'
-    return module_map.get(model_name, model_name)
 
 
-def _extract_unique_key(instance):
+def _get_request_user_info():
-    """提取记录的唯一标识"""
+    """从当前 HTTP 请求中尽力提取用户信息，获取不到拉倒"""
-    inspector = inspect(instance.__class__)
+    user_id, username, ip = None, 'system', ''
-    priority_fields = ['id', 'material_id', 'stock_id', 'user_id', 'borrow_no', 'order_no', 'bom_no']
-
-    for field in priority_fields:
-        if field in inspector.columns:
-            value = getattr(instance, field, None)
-            if value:
-                return str(value), f'{instance.__class__.__name__}:{value}'
-
-    for col in inspector.columns:
-        if col.type.python_type == str:
-            value = getattr(instance, col.name, None)
-            if value:
-                return str(value), str(value)
-
-    return str(instance.id), f'Record:{instance.id}'
-
-
-def _store_audit_details(module, target_id, action, details):
-    """将审计详情存入 g 对象"""
     if has_request_context():
-        if not hasattr(g, 'audit_details'):
+        try:
-            g.audit_details = {}
+            from flask_jwt_extended import get_jwt_identity, get_jwt
-        key = f"{module}:{target_id}"
+            user_id = get_jwt_identity()
-        g.audit_details[key] = {
+            claims = get_jwt()
-            'action': action,
+            username = claims.get('username', 'system')
-            'details': details
+        except Exception:
-        }
+            pass
-        print(f"💾 [X光-存储] 已存入 g.audit_details, key: {key}, details: {details}")
+        try:
+            ip = request.headers.get('X-Forwarded-For', '') or request.remote_addr or ''
+            if ip and ',' in ip:
+                ip = ip.split(',')[0].strip()
+        except Exception:
+            pass
+    return user_id, username, ip
 
 
 # ============================================================
-# 核心监听器函数
+# 核心：监听器内部直接创建并挂载日志
 # ============================================================
 
+def _create_audit_log(session, mapper, target, action, details):
+    """
+    监听器内部直接实例化 AuditLog 并加入当前事务 session。
+    由 SQLAlchemy 生命周期保证随主事务一同提交或回滚。
+    """
+    try:
+        from app.models.audit import AuditLog
+
+        user_id, username, ip = _get_request_user_info()
+        module = _get_module_name(mapper)
+
+        target_id = None
+        if hasattr(target, 'id'):
+            target_id = target.id
+        elif hasattr(target, 'stock_id'):
+            target_id = target.stock_id
+        elif hasattr(target, 'bom_no'):
+            target_id = target.bom_no
+
+        log = AuditLog(
+            user_id=user_id,
+            username=username,
+            action=action,
+            module=module,
+            target_id=str(target_id) if target_id else '0',
+            details=details,
+            ip_address=ip
+        )
+        session.add(log)
+
+    except Exception as e:
+        current_app.logger.error(f"Audit log auto-creation failed: {e}")
+
+
 def before_update_listener(mapper, connection, target):
-    """SQLAlchemy before_update 监听器"""
+    """UPDATE 事件：抓取字段变更明细"""
-    print(f"🔔 [X光-监听器] before_update 被调用, target: {target.__class__.__name__}")
+    if not _is_audit_model(mapper): return
-    
+    try:
-    if not _is_audit_model(mapper):
+        state = inspect(target)
-        print(f"⏭️ [X光-监听器] 跳过非审计模型: {mapper.class_.__name__}")
+        changes = {}
-        return
+        for attr in state.attrs:
-
+            if attr.key in IGNORE_FIELDS: continue
-    # ★ 使用 inspect(target) 而非 inspector._instance_state(target)
+            if attr.history.has_changes():
-    state = inspect(target)
+                old_val = attr.history.deleted[0] if attr.history.deleted else None
-    changes = {}
+                new_val = attr.history.added[0] if attr.history.added else None
-
+                changes[attr.key] = {
-    for attr in state.attrs:
+                    'old': _serialize_value(old_val),
-        if attr.key in IGNORE_FIELDS:
+                    'new': _serialize_value(new_val)
-            continue
+                }
-        if attr.history.has_changes():
+        if changes:
-            # ★ 正确使用 added[0] 和 deleted[0]
+            _create_audit_log(connection, mapper, target, 'update', {'changes': changes})
-            old_val = attr.history.deleted[0] if attr.history.deleted else None
+    except Exception as e:
-            new_val = attr.history.added[0] if attr.history.added else None
+        current_app.logger.error(f"Audit Update Error: {e}")
-            changes[attr.key] = {
-                'old': _serialize_value(old_val),
-                'new': _serialize_value(new_val)
-            }
-
-    print(f"📊 [X光-监听器] 计算完成, changes: {changes}")
-
-    if changes:
-        module = _get_module_name(mapper)
-        target_id, target_name = _extract_unique_key(target)
-
-        # ★ 将变更存入 g
-        details = {'changes': changes}
-        
-        # ★ X光调试：确认监听器是否成功算出 changes
-        print(f"🚀 [X光-监听器] 成功触发! target: {target}, changes: {changes}")
-        
-        _store_audit_details(module, target_id, 'update', details)
-    else:
-        print(f"⚠️ [X光-监听器] 无变更，跳过")
 
 
 def before_delete_listener(mapper, connection, target):
-    """SQLAlchemy before_delete 监听器"""
+    """DELETE 事件：抓取被删除对象的完整快照"""
-    if not _is_audit_model(mapper):
+    if not _is_audit_model(mapper): return
-        return
+    try:
-
+        state = inspect(target)
-    # 序列化被删除的数据快照
+        snap = {}
-    inspector = inspect(target.__class__)
+        for attr in state.attrs:
-    deleted_snapshot = {}
+            val = getattr(target, attr.key, None)
-
+            snap[attr.key] = _serialize_value(val)
-    for col in inspector.columns:
+        _create_audit_log(connection, mapper, target, 'delete', {'deleted_snapshot': snap})
-        try:
+    except Exception as e:
-            value = getattr(target, col.name, None)
+        current_app.logger.error(f"Audit Delete Error: {e}")
-            deleted_snapshot[col.name] = _serialize_value(value)
-        except Exception:
-            deleted_snapshot[col.name] = '[无法序列化]'
-
-    module = _get_module_name(mapper)
-    target_id, target_name = _extract_unique_key(target)
-
-    # ★ 将删除快照存入 g
-    details = {'deleted_snapshot': deleted_snapshot}
-    print(f"🗑️ [X光-监听器] delete 触发, target_id: {target_id}")
-    _store_audit_details(module, target_id, 'delete', details)
 
 
 def after_insert_listener(mapper, connection, target):
-    """SQLAlchemy after_insert 监听器"""
+    """INSERT 事件：抓取新增对象的完整快照"""
-    if not _is_audit_model(mapper):
+    if not _is_audit_model(mapper): return
-        return
+    try:
-
+        state = inspect(target)
-    inspector = inspect(target.__class__)
+        snap = {}
-    target_id, target_name = _extract_unique_key(target)
+        for attr in state.attrs:
-    module = _get_module_name(mapper)
+            val = getattr(target, attr.key, None)
-
+            snap[attr.key] = _serialize_value(val)
-    # 序列化新记录
+        _create_audit_log(connection, mapper, target, 'insert', {'created': snap})
-    new_snapshot = {}
+    except Exception:
-    for col in inspector.columns:
+        pass
-        try:
-            value = getattr(target, col.name, None)
-            new_snapshot[col.name] = _serialize_value(value)
-        except Exception:
-            new_snapshot[col.name] = '[无法序列化]'
-
-    # ★ 将新增数据存入 g
-    details = {'created': new_snapshot}
-    print(f"➕ [X光-监听器] insert 触发, target_id: {target_id}")
-    _store_audit_details(module, target_id, 'create', details)
 
 
 # ============================================================
@@ -200,7 +180,7 @@ def after_insert_listener(mapper, connection, target):
 # ============================================================
 
 def register_audit_listeners(db):
-    """注册审计监听器"""
+    """向所有需要审计的模型注册事件监听器"""
     from app.models import (
         MaterialBase, MaterialWarningSetting,
         StockBuy, StockSemi, StockProduct, StockService,
@@ -218,14 +198,13 @@ def register_audit_listeners(db):
     ]
 
     audit_models = [m for m in audit_models if m is not None]
-
+    count = 0
     for model in audit_models:
         try:
             event.listen(model, 'before_update', before_update_listener, propagate=True)
             event.listen(model, 'before_delete', before_delete_listener, propagate=True)
             event.listen(model, 'after_insert', after_insert_listener, propagate=True)
-            current_app.logger.info(f'[审计] 已绑定模型: {model.__name__}')
+            count += 1
-        except Exception as e:
+        except Exception:
-            current_app.logger.warning(f'[审计] 绑定模型 {model.__name__} 失败: {e}')
+            pass
-
+    return count
-    return len(audit_models)

inventory-backend/app/utils/decorators.py

@@ -5,7 +5,6 @@ from flask import jsonify, g, request, current_app, has_request_context
 import logging
 import json
 
-
 def _verify_token_in_redis():
     """
     验证当前 Token 是否与 Redis 中存储的 Token 一致（单设备登录互踢）
@@ -14,31 +13,23 @@ def _verify_token_in_redis():
     from flask import current_app
 
     if redis_client is None:
-        # Redis 不可用，跳过验证
         return True
 
     try:
-        # 获取请求中的 Token
         auth_header = request.headers.get('Authorization', '')
         if not auth_header.startswith('Bearer '):
             return True
 
-        request_token = auth_header[7:]  # 去掉 'Bearer ' 前缀
+        request_token = auth_header[7:]
-
-        # 获取当前用户 ID
         claims = get_jwt()
         user_id = claims.get('sub')
         if user_id is None:
             return True
 
-        # 从 Redis 获取存储的 Token
         stored_token = redis_client.get(f"user_token_{user_id}")
-
-        # 如果 Redis 中没有存储的 Token（可能是旧登录或 Redis 重启），允许通过
         if stored_token is None:
             return True
 
-        # 比较 Token 是否一致
         if request_token != stored_token:
             current_app.logger.warning(f"Token mismatch for user {user_id}: request token != stored token")
             return False
@@ -46,25 +37,18 @@ def _verify_token_in_redis():
         return True
     except Exception as e:
         current_app.logger.error(f"Redis token verification error: {e}")
-        # 出错时默认放行，避免影响正常业务
         return True
 
-
 def _raise_token_mismatch_error():
-    """抛出 Token 不一致的错误（用于单设备登录互踢）"""
+    """抛出 Token 不一致的错误"""
     return jsonify({
         'msg': '您的账号已在其他设备登录，请重新登录',
         'code': 401,
         'reason': 'token_mismatch'
     }), 401
 
-
 def role_required(*roles):
-    """
+    """自定义装饰器：检查用户角色"""
-    自定义装饰器：检查用户角色
-    使用方法: @role_required('super_admin', 'finance')
-    """
-
     def wrapper(fn):
         @wraps(fn)
         def decorator(*args, **kwargs):
@@ -72,7 +56,6 @@ def role_required(*roles):
             user_role = claims.get('role')
             user_role_upper = user_role.upper() if user_role else None
 
-            # 如果是超级管理员，拥有上帝视角，直接放行 (可选)
             if user_role_upper == 'SUPER_ADMIN':
                 return fn(*args, **kwargs)
 
@@ -80,16 +63,11 @@ def role_required(*roles):
                 return jsonify(msg='权限不足：您没有访问此资源的权限'), 403
 
             return fn(*args, **kwargs)
-
         return decorator
-
     return wrapper
 
-
 def login_required(fn):
-    """
+    """验证 JWT 令牌是否存在且有效"""
-    验证 JWT 令牌是否存在且有效
-    """
     @wraps(fn)
     def decorator(*args, **kwargs):
         try:
@@ -98,40 +76,31 @@ def login_required(fn):
             logging.warning(f"JWT verification failed: {e}")
             return jsonify(msg='登录已过期，请重新登录'), 401
 
-        # 单设备登录互踢检查
         if not _verify_token_in_redis():
             return _raise_token_mismatch_error()
 
         return fn(*args, **kwargs)
     return decorator
 
-
 def permission_required(permission_code):
-    """
+    """检查当前用户是否拥有指定权限码"""
-    检查当前用户是否拥有指定权限码
-    使用方法: @permission_required('material:base:read')
-    """
     def wrapper(fn):
         @wraps(fn)
         def decorator(*args, **kwargs):
-            # 首先验证 JWT
             try:
                 verify_jwt_in_request()
             except Exception as e:
                 logging.warning(f"JWT verification failed: {e}")
                 return jsonify(msg='登录已过期，请重新登录'), 401
 
-            # 单设备登录互踢检查
             if not _verify_token_in_redis():
                 return _raise_token_mismatch_error()
 
             claims = get_jwt()
             user_role = claims.get('role')
-            # 超级管理员放行 (忽略大小写)
             if user_role and user_role.upper() == 'SUPER_ADMIN':
                 return fn(*args, **kwargs)
 
-            # 根据角色查询数据库中的权限
             try:
                 from app.services.auth_service import AuthService
                 perm_dict = AuthService.get_user_permissions(user_role)
@@ -139,214 +108,24 @@ def permission_required(permission_code):
                 logging.warning(f"Failed to fetch permissions for role {user_role}: {e}")
                 return jsonify(msg='权限查询失败'), 403
 
-            # 合并菜单和元素权限
             all_perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
             if permission_code not in all_perms:
-                # 详细的调试日志
+                logging.warning(f"权限检查失败: 角色={user_role}, 所需权限={permission_code}")
-                print(f"🔴 [权限拦截] 角色 '{user_role}' 访问被拒！需要权限码: '{permission_code}', 但该角色实际拥有: {all_perms}")
-                logging.warning(
-                    f"权限检查失败: 角色={user_role}, 所需权限={permission_code}, 实际权限列表={all_perms}")
                 return jsonify(msg='权限不足：您没有访问此资源的权限'), 403
             return fn(*args, **kwargs)
         return decorator
     return wrapper
 
-
+def audit_log(module: str = None, action: str = None, get_target_id_fn=None, get_target_name_fn=None, get_details_fn=None):
-def audit_log(module: str, action: str = None, get_target_id_fn=None, get_target_name_fn=None, get_details_fn=None):
     """
-    审计日志装饰器
+    已废弃！
-    用法: @audit_log(module='inbound_buy', action='create')
+    由 SQLAlchemy 底层监听器（app/core/audit_listener.py）全面接管审计日志入库。
-          @audit_log(module='bom', action='update', get_target_id_fn=lambda: ..., get_details_fn=lambda req, resp: ...)
+    此装饰器保留空壳以防项目中其他文件 import 引用时报错。
-    
-    升级特性：
-    - 自动捕获请求 Payload 作为变更明细
-    - 自动过滤过长的 Base64 图片数据
-    - 支持自定义 get_details_fn 覆盖默认行为
     """
-    # 需要过滤的图片字段
-    IMAGE_FIELDS = {'arrival_photo', 'product_photo', 'photo', 'image', 'signature', 'borrow_signature', 'return_signature'}
-
-    def _filter_payload(payload):
-        """过滤 Payload 中的大字段，防止数据库膨胀"""
-        if not payload or not isinstance(payload, dict):
-            return payload
-        filtered = {}
-        for key, value in payload.items():
-            if key.lower() in IMAGE_FIELDS and isinstance(value, str) and len(value) > 100:
-                filtered[key] = '[图片数据已省略]'
-            elif isinstance(value, dict):
-                filtered[key] = _filter_payload(value)
-            elif isinstance(value, list):
-                filtered[key] = [
-                    _filter_payload(item) if isinstance(item, dict) else item
-                    for item in value
-                ]
-            else:
-                filtered[key] = value
-        return filtered
-
-    def _get_payload():
-        """自动获取请求 Payload"""
-        # 尝试 JSON
-        payload = request.get_json(silent=True)
-        if payload:
-            return payload
-        # 尝试 Form Data
-        if request.form:
-            return request.form.to_dict()
-        return None
-
     def wrapper(fn):
+        from functools import wraps
         @wraps(fn)
-        def decorator(*args, **kwargs):
+        def decorator(*inner_args, **inner_kwargs):
-            # 获取请求上下文
+            return fn(*inner_args, **inner_kwargs)
-            claims = get_jwt()
-            user_id = get_jwt_identity()
-            username = claims.get('username', '')
-            display_name = claims.get('display_name', '')
-
-            # ★ 修复 DetachedInstanceError：在 fn() 执行前预先获取用户完整信息
-            # 这样可以避免在 fn() 提交 session 后再访问 User 对象导致游离
-            if not display_name and user_id:
-                try:
-                    from app.models.system import SysUser
-                    user = SysUser.query.get(user_id)
-                    if user:
-                        display_name = user.display_name or username
-                except Exception:
-                    pass
-
-            # 预先获取 IP（避免后续访问 request 对象异常）
-            ip_address = request.headers.get('X-Forwarded-For') or request.remote_addr or ''
-            if ip_address and ',' in ip_address:
-                ip_address = ip_address.split(',')[0].strip()
-
-            # 获取请求信息
-            http_method = request.method
-            url = request.url
-            user_agent = request.headers.get('User-Agent', '')[:500]
-
-            # 解析 action（支持动态）
-            final_action = action
-            if callable(action):
-                final_action = action()
-
-            # 预先获取 Payload（用于后续 details 记录）
-            raw_payload = _get_payload()
-            filtered_payload = _filter_payload(raw_payload) if raw_payload else None
-
-            # 执行原函数（此时 Session 可能被提交或回滚）
-            response = fn(*args, **kwargs)
-
-            # 只记录成功的请求（响应状态码 200/201）
-            status_code = 200
-            if hasattr(response, 'status_code'):
-                status_code = response.status_code
-
-            if status_code in [200, 201]:
-                try:
-                    from app.models.audit import AuditLog
-                    from app.extensions import db
-                    from flask import current_app
-
-                    # ★ 已在上方预先获取 display_name，此处无需再查询 User 对象
-                    # 使用预先获取的字符串数据，避免 DetachedInstanceError
-
-                    # 获取 target_id
-                    target_id = None
-                    if get_target_id_fn:
-                        try:
-                            target_id = get_target_id_fn()
-                        except Exception:
-                            pass
-                    if not target_id and hasattr(response, 'json'):
-                        resp_data = response.get_json()
-                        if resp_data and isinstance(resp_data, dict):
-                            target_id = resp_data.get('id')
-
-                    # 获取 target_name
-                    target_name = None
-                    if get_target_name_fn:
-                        try:
-                            target_name = get_target_name_fn()
-                        except Exception:
-                            pass
-                    # 如果仍未获取到目标名称，尝试从响应 JSON 中常见字段获取
-                    if not target_name and hasattr(response, 'json'):
-                        resp_data = response.get_json()
-                        if resp_data and isinstance(resp_data, dict):
-                            # 优先从顶层获取
-                            for field in ['order_no', 'outbound_no', 'borrow_no', 'adjustment_no', 'material_name']:
-                                if field in resp_data:
-                                    target_name = resp_data[field]
-                                    break
-                            # 再尝试从 data 字段获取（部分 API 返回格式）
-                            if not target_name and 'data' in resp_data:
-                                data = resp_data['data']
-                                if isinstance(data, dict):
-                                    for field in ['order_no', 'outbound_no', 'borrow_no', 'adjustment_no', 'material_name']:
-                                        if field in data:
-                                            target_name = data[field]
-                                            break
-
-                    # 获取 details
-                    details = None
-                    if get_details_fn:
-                        # 优先使用自定义差异对比函数
-                        try:
-                            details = get_details_fn(request, response)
-                        except Exception:
-                            pass
-                    elif filtered_payload:
-                        # 默认：记录请求 Payload
-                        details = {'payload': filtered_payload}
-
-                    # ★ 核心重构：监听器计算，装饰器入库
-                    # 检查 g.audit_details 是否有底层监听器传递的高级日志
-                    if has_request_context() and hasattr(g, 'audit_details') and g.audit_details:
-                        check_key = f"{module}:{target_id}" if target_id else None
-                        if check_key and check_key in g.audit_details:
-                            # 用底层的高级日志(changes/deleted_snapshot/created)替换 payload
-                            audit_info = g.audit_details[check_key]
-                            details = audit_info.get('details', details)
-                            current_app.logger.info(f'[审计] 底层监听器已计算 {check_key}，使用高级日志')
-                        else:
-                            # 兼容 fallback：模糊匹配 target_name
-                            for key, audit_info in g.audit_details.items():
-                                if target_name and target_name in key:
-                                    details = audit_info.get('details', details)
-                                    current_app.logger.info(f'[审计] 底层监听器已计算 {key}，使用高级日志')
-                                    break
-
-                    # 保存日志（统一入库）
-                    print(f"📥 [X光-装饰器] 准备入库! details: {details}, g.audit_details: {getattr(g, 'audit_details', 'NOT_FOUND')}")
-                    
-                    log_entry = AuditLog(
-                        user_id=user_id,
-                        username=username,
-                        display_name=display_name,
-                        action=final_action or http_method.lower(),
-                        module=module,
-                        target_id=str(target_id) if target_id else None,
-                        target_name=target_name,
-                        details=details,
-                        ip_address=ip_address,
-                        user_agent=user_agent,
-                        method=http_method,
-                        url=url,
-                        status_code=status_code
-                    )
-                    db.session.add(log_entry)
-                    db.session.commit()
-
-                except Exception as e:
-                    import traceback
-                    traceback.print_exc()
-                    print(f"💥 [X光-崩溃] 审计入库失败原因: {str(e)}")
-                    current_app.logger.error(f"审计日志记录失败: {str(e)}")
-                    db.session.rollback()
-
-            return response
-
         return decorator
-    return wrapper
+    return wrapper