From 1d2e8feced2134e23444d6da5060bceda8d46327 Mon Sep 17 00:00:00 2001 From: dxc Date: Fri, 27 Feb 2026 13:03:27 +0800 Subject: [PATCH] feat: apply RBAC permission control to product module Co-authored-by: aider (openai/DeepSeek-V3.2-Thinking) --- .../app/api/v1/inbound/product.py | 94 ++++++++++++++++++- .../src/views/stock/inbound/product.vue | 75 ++++++++++++++- 2 files changed, 163 insertions(+), 6 deletions(-) diff --git a/inventory-backend/app/api/v1/inbound/product.py b/inventory-backend/app/api/v1/inbound/product.py index 9fca920..6b03ac8 100644 --- a/inventory-backend/app/api/v1/inbound/product.py +++ b/inventory-backend/app/api/v1/inbound/product.py @@ -1,15 +1,90 @@ # inventory-backend/app/api/v1/inbound/product.py from flask import Blueprint, request, jsonify from app.services.inbound.product_service import ProductInboundService +from app.utils.decorators import permission_required import traceback inbound_product_bp = Blueprint('stock_product', __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_product: 开头的权限码（这里我们返回一个特殊标记，表示全部） + # 为了简单，我们返回 ['inbound_product:*']，在过滤函数中特殊处理 + return ['inbound_product:*'] + 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_product:id', + 'base_id': 'inbound_product:base_id', + 'company_name': 'inbound_product:company_name', + 'material_name': 'inbound_product:material_name', + 'category': 'inbound_product:category', + 'material_type': 'inbound_product:material_type', + 'spec_model': 'inbound_product:spec_model', + 'unit': 'inbound_product:unit', + 'sku': 'inbound_product:sku', + 'inbound_date': 'inbound_product:inbound_date', + 'barcode': 'inbound_product:barcode', + 'serial_number': 'inbound_product:serial_number', + 'status': 'inbound_product:status', + 'quality_status': 'inbound_product:quality_status', + 'in_quantity': 'inbound_product:in_quantity', + 'stock_quantity': 'inbound_product:stock_quantity', + 'available_quantity': 'inbound_product:available_quantity', + 'warehouse_location': 'inbound_product:warehouse_location', + 'bom_code': 'inbound_product:bom_code', + 'bom_version': 'inbound_product:bom_version', + 'work_order_code': 'inbound_product:work_order_code', + 'order_id': 'inbound_product:order_id', + 'production_manager': 'inbound_product:production_manager', + 'production_start_time': 'inbound_product:production_start_time', + 'production_end_time': 'inbound_product:production_end_time', + 'raw_material_cost': 'inbound_product:raw_material_cost', + 'manual_cost': 'inbound_product:manual_cost', + 'sale_price': 'inbound_product:sale_price', + 'product_photo': 'inbound_product:product_photo', + 'quality_report_link': 'inbound_product:quality_report_link', + 'inspection_report_link': 'inbound_product:inspection_report_link', + 'detail_link': 'inbound_product:detail_link', + } + # 如果用户是超级管理员且有 'inbound_product:*'，则不过滤 + if 'inbound_product:*' 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. 基础物料搜索 (关键接口：配合 Service 实现自动回填) # ------------------------------------------------------------------ @inbound_product_bp.route('/search-base', methods=['GET']) +@permission_required('inbound_product') def search_base(): """ 对应前端 API: /inbound/product/search-base @@ -19,7 +94,10 @@ def search_base(): keyword = request.args.get('keyword', '') # 调用 Service 层已修复的 search_base_material 方法 data = ProductInboundService.search_base_material(keyword) - return jsonify({"code": 200, "msg": "success", "data": data}) + # 字段级脱敏 + user_permissions = get_current_user_permissions() + filtered_data = [filter_item_by_permissions(item, user_permissions) for item in data] + return jsonify({"code": 200, "msg": "success", "data": filtered_data}) except Exception as e: # 捕获异常并打印堆栈，方便调试 traceback.print_exc() @@ -29,6 +107,7 @@ def search_base(): # 0.5 [新增] BOM 搜索接口 # ------------------------------------------------------------------ @inbound_product_bp.route('/search-bom', methods=['GET']) +@permission_required('inbound_product') def search_bom(): """ 供前端下拉框远程搜索使用 (搜索BOM) @@ -50,6 +129,7 @@ def search_bom(): # 1. 获取列表 (支持 status 多选筛选) # ------------------------------------------------------------------ @inbound_product_bp.route('/list', methods=['GET']) +@permission_required('inbound_product') def get_list(): try: page = request.args.get('page', 1, type=int) @@ -61,6 +141,10 @@ def get_list(): statuses = statuses_str.split(',') if statuses_str else [] result = ProductInboundService.get_list(page, limit, keyword, statuses) + # 字段级脱敏 + 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() @@ -71,6 +155,7 @@ def get_list(): # 2. 新增入库 # ------------------------------------------------------------------ @inbound_product_bp.route('/submit', methods=['POST']) +@permission_required('inbound_product:operation') def submit(): try: # 调用 Service 处理入库，获取新创建的对象 @@ -91,6 +176,7 @@ def submit(): # 3. 更新入库 # ------------------------------------------------------------------ @inbound_product_bp.route('/', methods=['PUT']) +@permission_required('inbound_product:operation') def update(id): try: ProductInboundService.update_inbound(id, request.get_json()) @@ -104,6 +190,7 @@ def update(id): # 4. 删除 # ------------------------------------------------------------------ @inbound_product_bp.route('/', methods=['DELETE']) +@permission_required('inbound_product:operation') def delete(id): try: ProductInboundService.delete_inbound(id) @@ -117,6 +204,7 @@ def delete(id): # 5. 获取出库历史 # ------------------------------------------------------------------ @inbound_product_bp.route('//history', methods=['GET']) +@permission_required('inbound_product') def get_history(id): try: data = ProductInboundService.get_outbound_history(id) @@ -130,6 +218,7 @@ def get_history(id): # 6. 系统用户建议 # ------------------------------------------------------------------ @inbound_product_bp.route('/suggestions/users', methods=['GET']) +@permission_required('inbound_product') def get_user_suggestions(): keyword = request.args.get('keyword', '') data = ProductInboundService.search_system_users(keyword) @@ -140,9 +229,10 @@ def get_user_suggestions(): # 7. 获取筛选选项 # ------------------------------------------------------------------ @inbound_product_bp.route('/options', methods=['GET']) +@permission_required('inbound_product') def get_options(): try: data = ProductInboundService.get_filter_options() return jsonify({"code": 200, "msg": "success", "data": data}) except Exception as e: - return jsonify({"code": 500, "msg": str(e)}), 500 \ No newline at end of file + return jsonify({"code": 500, "msg": str(e)}), 500 diff --git a/inventory-web/src/views/stock/inbound/product.vue b/inventory-web/src/views/stock/inbound/product.vue index 50aa6ad..b901e60 100644 --- a/inventory-web/src/views/stock/inbound/product.vue +++ b/inventory-web/src/views/stock/inbound/product.vue @@ -41,13 +41,13 @@
- 成品入库登记 + 成品入库登记 刷新 - {{ c.label }} + {{ c.label }} @@ -137,7 +137,7 @@ - +