From 379bc5786f5b1565ab831786c65aedde6c70692f Mon Sep 17 00:00:00 2001 From: dxc Date: Fri, 27 Feb 2026 11:48:33 +0800 Subject: [PATCH] feat: implement RBAC for inbound buy module with field-level permissions Co-authored-by: aider (openai/DeepSeek-V3.2-Thinking) --- inventory-backend/app/api/v1/inbound/buy.py | 91 ++++++++++++++++++- inventory-web/src/views/stock/inbound/buy.vue | 87 +++++++++++++++++- 2 files changed, 172 insertions(+), 6 deletions(-) diff --git a/inventory-backend/app/api/v1/inbound/buy.py b/inventory-backend/app/api/v1/inbound/buy.py index 7ccbd51..59915bb 100644 --- a/inventory-backend/app/api/v1/inbound/buy.py +++ b/inventory-backend/app/api/v1/inbound/buy.py @@ -1,14 +1,89 @@ from flask import Blueprint, request, jsonify from app.services.inbound.buy_service import BuyInboundService +from app.utils.decorators import permission_required import traceback inbound_buy_bp = Blueprint('stock_buy', __name__) +# ============================================================================== +# 辅助函数：获取当前用户的完整权限列表（基于角色查询） +# ============================================================================== +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': + # 返回所有以 inbound_buy: 开头的权限码（这里我们返回一个特殊标记，表示全部） + # 为了简单，我们返回 ['inbound_buy:*']，在过滤函数中特殊处理 + return ['inbound_buy:*'] + 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 = { + 'id': 'inbound_buy:id', + 'base_id': 'inbound_buy:base_id', + 'global_print_id': 'inbound_buy:global_print_id', + 'sku': 'inbound_buy:sku', + 'barcode': 'inbound_buy:barcode', + 'in_date': 'inbound_buy:in_date', + 'serial_number': 'inbound_buy:serial_number', + 'batch_number': 'inbound_buy:batch_number', + 'status': 'inbound_buy:status', + 'in_quantity': 'inbound_buy:in_quantity', + 'stock_quantity': 'inbound_buy:stock_quantity', + 'available_quantity': 'inbound_buy:available_quantity', + 'inspection_status': 'inbound_buy:inspection_status', + 'warehouse_location': 'inbound_buy:warehouse_location', + 'unit_price': 'inbound_buy:unit_price', + 'tax_rate': 'inbound_buy:tax_rate', + 'total_price': 'inbound_buy:total_price', + 'currency': 'inbound_buy:currency', + 'exchange_rate': 'inbound_buy:exchange_rate', + 'supplier_name': 'inbound_buy:supplier_name', + 'buyer_name': 'inbound_buy:buyer_name', + 'buyer_email': 'inbound_buy:buyer_email', + 'original_link': 'inbound_buy:original_link', + 'detail_link': 'inbound_buy:detail_link', + 'arrival_photo': 'inbound_buy:arrival_photo', + 'inspection_report': 'inbound_buy:inspection_report', + 'material_name': 'inbound_buy:material_name', + 'spec_model': 'inbound_buy:spec_model', + 'category': 'inbound_buy:category', + 'unit': 'inbound_buy:unit', + 'material_type': 'inbound_buy:material_type', + 'company_name': 'inbound_buy:company_name', + } + # 如果用户是超级管理员且有 'inbound_buy:*'，则不过滤 + if 'inbound_buy:*' 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 + return item_dict + + # ------------------------------------------------------------------ # 0. 基础物料搜索 # ------------------------------------------------------------------ @inbound_buy_bp.route('/search-base', methods=['GET']) +@permission_required('inbound_buy') def search_base(): try: keyword = request.args.get('keyword', '') @@ -33,6 +108,7 @@ def search_base(): # 1. 获取列表 (修改：接收 category 和 material_type) # ------------------------------------------------------------------ @inbound_buy_bp.route('/list', methods=['GET']) +@permission_required('inbound_buy') def get_list(): try: page = request.args.get('page', 1, type=int) @@ -48,6 +124,10 @@ def get_list(): statuses = statuses_str.split(',') if statuses_str else [] result = BuyInboundService.get_list(page, limit, keyword, statuses, category, material_type) + # 字段级脱敏 + 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": "success", "data": result}) except Exception as e: traceback.print_exc() @@ -58,6 +138,7 @@ def get_list(): # 2. 新增入库 # ------------------------------------------------------------------ @inbound_buy_bp.route('/submit', methods=['POST']) +@permission_required('inbound_buy:operation') def submit(): try: data = request.get_json() @@ -80,6 +161,7 @@ def submit(): # 3. 更新入库 # ------------------------------------------------------------------ @inbound_buy_bp.route('/', methods=['PUT']) +@permission_required('inbound_buy:operation') def update_buy(id): try: data = request.get_json() @@ -93,6 +175,7 @@ def update_buy(id): # 4. 删除 # ------------------------------------------------------------------ @inbound_buy_bp.route('/', methods=['DELETE']) +@permission_required('inbound_buy:operation') def delete_buy(id): try: BuyInboundService.delete_inbound(id) @@ -105,6 +188,7 @@ def delete_buy(id): # 5. [新增] 获取筛选下拉选项 (修复404的关键) # ------------------------------------------------------------------ @inbound_buy_bp.route('/options', methods=['GET']) +@permission_required('inbound_buy') def get_options(): try: data = BuyInboundService.get_filter_options() @@ -117,6 +201,7 @@ def get_options(): # 6. 获取关联的出库历史 (如果有) # ------------------------------------------------------------------ @inbound_buy_bp.route('//history', methods=['GET']) +@permission_required('inbound_buy') def get_history(id): # 如果没有出库模块，这个接口可能为空，但为保持兼容性保留 return jsonify({"code": 200, "msg": "success", "data": []}) @@ -126,6 +211,7 @@ def get_history(id): # 7. 供应商建议 # ------------------------------------------------------------------ @inbound_buy_bp.route('/suggestions/suppliers', methods=['GET']) +@permission_required('inbound_buy') def get_supplier_suggestions(): base_id = request.args.get('base_id', type=int) if not base_id: @@ -138,6 +224,7 @@ def get_supplier_suggestions(): # 8. 采购人建议 (全局) # ------------------------------------------------------------------ @inbound_buy_bp.route('/suggestions/users', methods=['GET']) +@permission_required('inbound_buy') def get_user_suggestions(): keyword = request.args.get('keyword', '') data = BuyInboundService.get_history_purchasers(keyword) @@ -148,6 +235,7 @@ def get_user_suggestions(): # 9. 链接建议 # ------------------------------------------------------------------ @inbound_buy_bp.route('/suggestions/links', methods=['GET']) +@permission_required('inbound_buy') def get_link_suggestions(): base_id = request.args.get('base_id', type=int) link_type = request.args.get('type', 'original') # original or detail @@ -161,9 +249,10 @@ def get_link_suggestions(): # 10. 库位建议 # ------------------------------------------------------------------ @inbound_buy_bp.route('/suggestions/locations', methods=['GET']) +@permission_required('inbound_buy') def get_location_suggestions(): base_id = request.args.get('base_id', type=int) if not base_id: return jsonify({"code": 400, "msg": "base_id required"}), 400 data = BuyInboundService.get_history_locations(base_id) - return jsonify({"code": 200, "msg": "success", "data": data}) \ No newline at end of file + return jsonify({"code": 200, "msg": "success", "data": data}) diff --git a/inventory-web/src/views/stock/inbound/buy.vue b/inventory-web/src/views/stock/inbound/buy.vue index ed3a735..46bf8ac 100644 --- a/inventory-web/src/views/stock/inbound/buy.vue +++ b/inventory-web/src/views/stock/inbound/buy.vue @@ -56,7 +56,7 @@
- 新增 + 新增 @@ -67,13 +67,13 @@
基础信息
- {{ c.label }} + {{ c.label }}
库存与商务
- {{ c.label }} + {{ c.label }} @@ -167,7 +167,7 @@ - +