KCGL/inventory-backend/app/utils/audit_events.py

# inventory-backend/app/utils/audit_events.py
"""
[已停用 — DEPRECATED] 全局审计日志拦截器

★ 本模块自 v3.x 起不再注册、不再生效。
   审计日志已统一由 app/core/audit_listener.py 接管（白名单制 + 请求上下文守卫）。

停用原因（原实现的三个设计缺陷）：
  1. 监听 db.Model 全局，无白名单 —— 系统表/草稿表/向量表被一并审计，
     产生大量无业务价值的记录；
  2. 无 has_request_context() 守卫 —— 系统初始化与后台定时任务产生的变更
     被记为 username='system' 的日志（历史数据中占 32%）；
  3. target_name 兜底为 `{英文表名} ID:{id}`（历史数据中占 72%），
     对用户完全不可读。

保留文件仅为兼容可能的历史 import；register_audit_events() 已改为空操作，
调用它不会产生任何监听器。
"""
import json
from datetime import datetime, date
from decimal import Decimal
from flask import request, has_request_context
from sqlalchemy import event, text


class AuditJSONEncoder(json.JSONEncoder):
    """JSON 序列化增强器，支持 datetime/Decimal 等特殊类型"""
    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        return str(obj)


def model_to_dict(obj):
    """将 SQLAlchemy 模型实例转换为字典"""
    return {c.name: getattr(obj, c.name) for c in obj.__table__.columns}


def get_current_user_info():
    """
    从当前 HTTP 请求上下文中提取用户信息
    兼容 JWT 和匿名访问
    """
    user_info = {
        'user_id': 'system',
        'username': 'system',
        'display_name': 'System',
        'ip_address': '127.0.0.1',
        'method': 'SYSTEM',
        'url': ''
    }

    if has_request_context():
        # 获取 IP 地址
        user_info['ip_address'] = request.headers.get('X-Forwarded-For', '') or request.remote_addr or '127.0.0.1'
        if ',' in user_info['ip_address']:
            user_info['ip_address'] = user_info['ip_address'].split(',')[0].strip()

        user_info['method'] = request.method
        user_info['url'] = request.path

        # 尝试从 JWT 获取用户信息
        try:
            from flask_jwt_extended import get_jwt_identity, get_jwt
            user_id = get_jwt_identity()
            claims = get_jwt()

            if user_id:
                user_info['user_id'] = str(user_id)
            if claims:
                user_info['username'] = claims.get('username', 'unknown')
                user_info['display_name'] = claims.get('display_name', claims.get('username', 'Unknown'))
        except Exception:
            pass

    return user_info


def serialize_value(value):
    """序列化单个值，确保 JSON 兼容"""
    if value is None:
        return None
    if isinstance(value, (datetime, date)):
        return value.strftime('%Y-%m-%d %H:%M:%S')
    if isinstance(value, Decimal):
        return float(value)
    if isinstance(value, (bytes, bytearray)):
        try:
            return value.decode('utf-8')
        except Exception:
            return '[二进制数据]'
    return value


# 需要忽略的审计字段（时间戳等自动维护字段）
IGNORE_FIELDS = {
    'updated_at', 'update_time', 'modified_time', 'last_modified',
    'created_at', 'create_time', 'created_on', 'version',
}

# 审计日志表名
AUDIT_TABLE = 'audit_logs'

# 不需要审计的表
IGNORE_TABLES = {'audit_logs', 'sys_log', 'syslog', 'alembic_version'}


def insert_audit_log(connection, action, target, details):
    """
    使用 connection.execute 直接插入审计日志
    避免干扰当前 session 事务，自动随主事务一起提交/回滚
    """
    tablename = target.__tablename__

    # 严禁监听日志表本身，防止无限递归
    if tablename in IGNORE_TABLES:
        return

    # 获取目标 ID
    target_id = ''
    if hasattr(target, 'id'):
        target_id = str(target.id)
    elif hasattr(target, 'stock_id'):
        target_id = str(target.stock_id)
    elif hasattr(target, 'uuid'):
        target_id = str(target.uuid)
    elif hasattr(target, 'bom_no'):
        target_id = str(target.bom_no)

    # 获取目标名称（用于展示）
    target_name = ''
    for name_field in ['name', 'title', 'material_name', 'product_name', 'display_name', 'username']:
        if hasattr(target, name_field):
            val = getattr(target, name_field)
            if val:
                target_name = str(val)
                break

    # 如果当前表没名字，但它有关联的物料对象 (比如 material.name)
    if not target_name and hasattr(target, 'material') and target.material:
        target_name = getattr(target.material, 'name', '')

    # 如果当前表有 material_id，尝试从关联的 material 表查询名称
    if not target_name and hasattr(target, 'material_id') and target.material_id:
        try:
            # 使用 connection 查询物料表获取名称
            result = connection.execute(
                text("SELECT name FROM material_base WHERE id = :id"),
                {'id': target.material_id}
            ).fetchone()
            if result:
                target_name = str(result[0])
        except Exception:
            pass

    # 如果实在找不到名字，再用 表名 + ID 兜底
    if not target_name:
        target_name = f"{tablename} ID:{target_id}"

    user_info = get_current_user_info()

    # 推断模块名称
    module = _infer_module_name(tablename, target)

    # 使用原始 SQL 插入，确保事务一致性
    sql = text("""
        INSERT INTO audit_logs
        (user_id, username, display_name, action, module, target_id, target_name, details, ip_address, method, url, created_at)
        VALUES
        (:user_id, :username, :display_name, :action, :module, :target_id, :target_name, :details, :ip_address, :method, :url, :created_at)
    """)

    connection.execute(sql, {
        'user_id': user_info['user_id'],
        'username': user_info['username'],
        'display_name': user_info['display_name'],
        'action': action,
        'module': module,
        'target_id': target_id,
        'target_name': target_name,
        'details': json.dumps(details, cls=AuditJSONEncoder),
        'ip_address': user_info['ip_address'],
        'method': user_info['method'],
        'url': user_info['url'],
        'created_at': datetime.now()
    })


def _infer_module_name(tablename, target):
    """根据表名或模型类推断所属模块"""
    class_name = target.__class__.__name__

    if any(kw in class_name for kw in ['Stock', 'Buy', 'Inbound']):
        return '入库管理'
    if any(kw in class_name for kw in ['Outbound']):
        return '出库管理'
    if any(kw in class_name for kw in ['Borrow', 'Return']):
        return '借还管理'
    if any(kw in class_name for kw in ['Repair']):
        return '维修管理'
    if any(kw in class_name for kw in ['Scrap']):
        return '报废管理'
    if any(kw in class_name for kw in ['Bom', 'BOM']):
        return 'BOM管理'
    if any(kw in class_name for kw in ['StockTake', 'StockAdjust', 'Adjustment']):
        return '盘点管理'
    if any(kw in class_name for kw in ['Material', 'Base']):
        return '基础数据'
    if any(kw in class_name for kw in ['SysUser', 'SysMenu', 'SysRole', 'SysPermission']):
        return '系统管理'
    if any(kw in class_name for kw in ['Warehouse', 'Location']):
        return '库位管理'

    return tablename or '未知模块'


def _has_changes(history):
    """检查历史记录对象是否有变更"""
    return history.has_changes()

def register_audit_events(db):
    """
    [已停用] 空操作，保留仅为兼容历史调用。

    审计日志已由 app/core/audit_listener.py 接管（白名单制 + 请求上下文守卫）。
    调用本函数不会注册任何监听器。
    """
    return False