KCGL/inventory-backend/app/api/v1/audit.py

# 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/<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