duxingchen/KCGL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add MaterialBase permission control with field-level filtering

Browse Source 
Co-authored-by: aider (openai/DeepSeek-V3.2-Thinking) <aider@aider.chat>
This commit is contained in:
dxc
2026-02-27 10:16:43 +08:00
parent c86e67b793
commit 73ee163352
3 changed files with 131 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
inventory-backend/app/api/v1/inbound/base.py
View File

@ -1,22 +1,79 @@
# 文件路径: app/api/v1/inbound/base.py
from flask import Blueprint, request, jsonify, send_file
from flask import Blueprint, request, jsonify, send_file, g
from app.services.inbound.base_service import MaterialBaseService
from app.utils.decorators import login_required, permission_required
import traceback
import datetime
inbound_base_bp = Blueprint('stock_base', __name__)
# ==============================================================================
# 辅助函数：获取当前用户的字段级权限列表（基于角色查询）
# ==============================================================================
def get_current_field_permissions():
"""
返回当前用户拥有的字段权限码列表（例如 ['id','companyName',...]）
超级管理员返回所有权限。
此函数为示例实现，实际应根据项目权限模型完善。
"""
# TODO: 从 JWT 或数据库查询当前用户角色对应的权限码
# 这里假设角色为 'admin'/'manager' 拥有全部字段权限，其他角色只有部分
# 实际应替换为真实的权限查询逻辑
from flask_jwt_extended import get_jwt
claims = get_jwt()
user_role = claims.get('role')
if user_role == 'super_admin':
# 所有字段权限
return ['id', 'companyName', 'name', 'commonName', 'category', 'type',
'spec', 'unit', 'inventoryCount', 'availableCount', 'files', 'isEnabled']
if user_role in ['admin', 'manager']:
return ['id', 'companyName', 'name', 'commonName', 'category', 'type',
'spec', 'unit', 'inventoryCount', 'availableCount', 'files', 'isEnabled']
# 普通用户只有部分权限
return ['name', 'spec', 'unit', 'inventoryCount', 'availableCount']
def filter_item_by_permissions(item_dict, field_permissions):
"""
根据字段权限过滤 item 字典，无权限的字段值置为 None
"""
# 字段名到权限码的映射（与前端 permissionMap 保持一致）
field_to_perm = {
'id': 'id',
'companyName': 'companyName',
'name': 'name',
'commonName': 'commonName',
'category': 'category',
'type': 'type',
'spec': 'spec',
'unit': 'unit',
'inventoryCount': 'inventoryCount',
'availableCount': 'availableCount',
'generalManual': 'files',
'generalImage': 'files',
'isEnabled': 'isEnabled'
}
for field, perm_code in field_to_perm.items():
if field in item_dict and perm_code not in field_permissions:
item_dict[field] = None
return item_dict
# ==============================================================================
# 1. 搜索接口 (GET /api/v1/inbound/base/search)
# ==============================================================================
@inbound_base_bp.route('/search', methods=['GET'])
@permission_required('material:base:read')
def search_base():
try:
keyword = request.args.get('keyword', '')
data = MaterialBaseService.search_material(keyword)
return jsonify({"code": 200, "msg": "success", "data": data})
# 字段级脱敏
field_perms = get_current_field_permissions()
filtered_data = [filter_item_by_permissions(item, field_perms) for item in data]
return jsonify({"code": 200, "msg": "success", "data": filtered_data})
except Exception as e:
traceback.print_exc()
return jsonify({"code": 500, "msg": str(e)}), 500
@ -26,6 +83,7 @@ def search_base():
# 2. 列表接口 (GET /api/v1/inbound/base/list)
# ==============================================================================
@inbound_base_bp.route('/list', methods=['GET'])
@permission_required('material:base:read')
def get_list():
try:
page = request.args.get('pageNum', 1, type=int)
@ -41,6 +99,10 @@ def get_list():
}
result = MaterialBaseService.get_list(page, limit, filters)
# 字段级脱敏
field_perms = get_current_field_permissions()
if result.get('items'):
result['items'] = [filter_item_by_permissions(item, field_perms) for item in result['items']]
return jsonify({"code": 200, "msg": "success", "data": result})
except Exception as e:
traceback.print_exc()
@ -51,6 +113,7 @@ def get_list():
# 2.1 选项接口 (GET /api/v1/inbound/base/options)
# ==============================================================================
@inbound_base_bp.route('/options', methods=['GET'])
@permission_required('material:base:read')
def get_options():
try:
data = MaterialBaseService.get_distinct_options()
@ -64,6 +127,7 @@ def get_options():
# 2.2 导出接口 (GET /api/v1/inbound/base/export)
# ==============================================================================
@inbound_base_bp.route('/export', methods=['GET'])
@permission_required('material:base:read')
def export_data():
try:
# 获取筛选条件
@ -101,6 +165,7 @@ def export_data():
# 3. 新增接口 (POST /api/v1/inbound/base/)
# ==============================================================================
@inbound_base_bp.route('/', methods=['POST'])
@permission_required('material:base:write')
def create():
try:
data = request.get_json()
@ -122,6 +187,7 @@ def create():
# 4. 修改接口 (PUT /api/v1/inbound/base/<id>)
# ==============================================================================
@inbound_base_bp.route('/<int:id>', methods=['PUT'])
@permission_required('material:base:write')
def update(id):
try:
data = request.get_json()
@ -136,10 +202,11 @@ def update(id):
# 5. 删除接口 (DELETE /api/v1/inbound/base/<id>)
# ==============================================================================
@inbound_base_bp.route('/<int:id>', methods=['DELETE'])
@permission_required('material:base:delete')
def delete(id):
try:
MaterialBaseService.delete_material(id)
return jsonify({"code": 200, "msg": "删除成功"})
except Exception as e:
traceback.print_exc()
return jsonify({"code": 500, "msg": str(e)}), 500
return jsonify({"code": 500, "msg": str(e)}), 500

57
inventory-backend/app/utils/decorators.py
View File

@ -1,7 +1,8 @@
# app/utils/decorators.py
from functools import wraps
from flask_jwt_extended import get_jwt
from flask import jsonify
from flask_jwt_extended import get_jwt, verify_jwt_in_request
from flask import jsonify, g
import logging
def role_required(*roles):
@ -27,4 +28,54 @@ def role_required(*roles):
return decorator
return wrapper
return wrapper
def login_required(fn):
"""
验证 JWT 令牌是否存在且有效
"""
@wraps(fn)
def decorator(*args, **kwargs):
try:
verify_jwt_in_request()
except Exception as e:
logging.warning(f"JWT verification failed: {e}")
return jsonify(msg='登录已过期，请重新登录'), 401
return fn(*args, **kwargs)
return decorator
def permission_required(permission_code):
"""
检查当前用户是否拥有指定权限码
使用方法: @permission_required('material:base:read')
"""
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
# 首先验证 JWT
try:
verify_jwt_in_request()
except Exception as e:
logging.warning(f"JWT verification failed: {e}")
return jsonify(msg='登录已过期，请重新登录'), 401
claims = get_jwt()
user_role = claims.get('role')
# 超级管理员放行
if user_role == 'super_admin':
return fn(*args, **kwargs)
# TODO: 根据角色和 permission_code 查询数据库验证权限
# 此处为示例逻辑：假设角色 'admin' 和 'manager' 拥有所有权限
# 实际项目中应替换为真实的权限查询
if user_role in ['admin', 'manager']:
return fn(*args, **kwargs)
# 其他角色暂时拒绝，并记录日志
logging.warning(
f'Permission check not implemented for {permission_code}, user role {user_role}. Access denied.')
return jsonify(msg='权限不足：您没有访问此资源的权限'), 403
return decorator
return wrapper