feat: add RBAC control for outbound list module

Co-authored-by: aider (openai/DeepSeek-V3.2-Thinking) <aider@aider.chat>
This commit is contained in:
dxc
2026-02-27 13:57:59 +08:00
parent 3714dd180b
commit 5065410662
2 changed files with 117 additions and 19 deletions

View File

@ -8,6 +8,66 @@ import traceback
outbound_bp = Blueprint('outbound', __name__, url_prefix='/outbound')
# ==============================================================================
# 辅助函数:获取当前用户的完整权限列表(基于角色查询)
# ==============================================================================
def get_current_user_permissions():
"""
返回当前用户拥有的所有权限码列表(包括菜单和元素)
此函数根据角色查询数据库得到权限。
"""
from flask_jwt_extended import get_jwt
from app.services.auth_service import AuthService
claims = get_jwt()
user_role = claims.get('role')
if not user_role:
return []
# 超级管理员返回所有字段权限
if user_role == 'super_admin':
return ['outbound_list:*']
perm_dict = AuthService.get_user_permissions(user_role)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms
def filter_item_by_permissions(item_dict, user_permissions):
"""
根据用户权限过滤 item 字典,无权限的字段值置为 None
"""
# 字段名到权限码的映射(与前端 permissionMap 保持一致)
field_to_perm = {
'outbound_no': 'outbound_list:outbound_no',
'outbound_time': 'outbound_list:outbound_time',
'outbound_type': 'outbound_list:outbound_type',
'total_amount': 'outbound_list:total_amount',
'consumer_name': 'outbound_list:consumer_name',
'operator_name': 'outbound_list:operator_name',
'remark': 'outbound_list:remark',
'signature_path': 'outbound_list:signature_path',
# 明细字段
'sku': 'outbound_list:sku',
'name': 'outbound_list:name',
'material_type': 'outbound_list:material_type',
'category': 'outbound_list:category',
'spec_model': 'outbound_list:spec_model',
'quantity': 'outbound_list:quantity',
'unit_price': 'outbound_list:unit_price',
'subtotal': 'outbound_list:subtotal',
}
# 如果用户是超级管理员且有 'outbound_list:*',则不过滤
if 'outbound_list:*' in user_permissions:
return item_dict
for field, perm_code in field_to_perm.items():
if field in item_dict and perm_code not in user_permissions:
item_dict[field] = None
# 如果 item_dict 中包含 items 列表,递归处理每个子项
if 'items' in item_dict and isinstance(item_dict['items'], list):
for sub_item in item_dict['items']:
filter_item_by_permissions(sub_item, user_permissions)
return item_dict
# --------------------------------------------------------
# 1. 扫码查询库存接口 (关联三个库存表)
# GET /api/v1/outbound/scan?barcode=...
@ -105,7 +165,7 @@ def create_outbound():
# --------------------------------------------------------
@outbound_bp.route('', methods=['GET'])
@jwt_required()
@permission_required('outbound_selection')
@permission_required('outbound_list')
def get_outbound_list():
try:
page = int(request.args.get('page', 1))
@ -116,6 +176,11 @@ def get_outbound_list():
# ★ [修改] 调用分组查询服务
result = OutboundService.get_grouped_list(page, limit, keyword)
# 字段级脱敏
user_permissions = get_current_user_permissions()
if result.get('items'):
result['items'] = [filter_item_by_permissions(item, user_permissions) for item in result['items']]
return jsonify({
'code': 200,
'msg': '获取成功',