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

feat: JWT多租户数据权限隔离 & 主管系统管理权限 & 含税单价补齐

Browse Source 
## 多租户公司数据隔离
- 新增 get_current_company_filter() 工具函数 (decorators.py)
  SUPER_ADMIN: 可传company_name参数过滤或传ALL看全量
  其他角色: 强制隔离到JWT中的company_name
- 重构 base_service.py / buy_service.py: 用集中式函数替换内联公司过滤
- SysRolePermission 表新增 company_name 字段，支持同角色不同公司权限
- get_user_permissions() 新增 company_name 参数，查公司定制+全局模板权限
- permission.py API 新增 @permission_required 拦截 + 公司过滤
- 19个API/service文件传递 company_name 到权限查询

## 主管系统管理权限
- delete_user() 允许SUPERVISOR删除同公司用户 (原仅SUPER_ADMIN)
- get_all_users() 新增 company_name 参数过滤
- 用户列表/权限分配 API 应用 get_current_company_filter()
- 前端 UserCreate.vue: 超管可见公司下拉框，主管隐藏部门字段

## 前端多租户适配
- material/list.vue / buy.vue: 公司下拉框仅超管可见，默认ALL
- UserCreate.vue: 新增搜索栏公司筛选，部门字段按角色显隐
- auth.ts: getUserList() 支持 params 参数

## Bug修复: 含税单价字段补齐
- buy.vue: 表格列/高级筛选/排序/权限映射新增 post_tax_unit_price
- buy_service.py: allowed_fields/sort_field_map 新增 post_tax_unit_price
This commit is contained in:
yueli
2026-07-13 15:12:22 +08:00
parent 4ce42309db
commit 4f5965db02
22 changed files with 289 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
inventory-backend/app/api/v1/auth.py
View File

@ -31,16 +31,17 @@ auth_bp = Blueprint('auth', __name__)
def get_current_user_permissions():
"""
返回当前用户拥有的所有权限码列表（包括菜单和元素）
此函数根据角色查询数据库得到权限。
此函数根据角色+公司查询数据库得到权限。
"""
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
# 超级管理员返回所有字段权限
if user_role.upper() == 'SUPER_ADMIN':
return ['system_user:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms
@ -269,7 +270,9 @@ def update_user(user_id):
@permission_required('system_user')
def get_users():
try:
users = AuthService.get_all_users()
from app.utils.decorators import get_current_company_filter
company_filter = get_current_company_filter()
users = AuthService.get_all_users(company_name=company_filter)
user_permissions = get_current_user_permissions()
filtered_users = [filter_item_by_permissions(user, user_permissions) for user in users]
return jsonify({'msg': '获取成功', 'data': filtered_users}), 200
@ -293,8 +296,9 @@ def delete_user(user_id):
try:
claims = get_jwt()
operator_role = claims.get('role')
operator_company = claims.get('company_name', '')
username = AuthService.delete_user(user_id, operator_role)
username = AuthService.delete_user(user_id, operator_role, operator_company)
return jsonify({'msg': '删除成功', 'username': username}), 200
except Exception as e:
current_app.logger.error(f"Delete User Failed: {str(e)}")
@ -311,7 +315,8 @@ def get_my_permissions():
try:
claims = get_jwt()
role = claims.get('role')
permissions = AuthService.get_user_permissions(role)
company_name = claims.get('company_name', '')
permissions = AuthService.get_user_permissions(role, company_name=company_name)
return jsonify({'msg': '获取成功', 'data': permissions}), 200
except Exception as e:
current_app.logger.error(f"Get Permissions Failed: {str(e)}")

3
inventory-backend/app/api/v1/bom.py
View File

@ -22,12 +22,13 @@ def get_current_user_permissions():
"""
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
if user_role.upper() == 'SUPER_ADMIN':
return ['bom_manage:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms

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

@ -18,17 +18,17 @@ inbound_base_bp = Blueprint('stock_base', __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')
user_company = claims.get('company_name', '')
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
# 超级管理员返回所有字段权限
if user_role.upper() == 'SUPER_ADMIN':
# 返回通配符权限（供列表脱敏使用）以及所有具体权限（供导出脱敏使用）
return [
'material_list:*',
'material_list:id',
@ -45,7 +45,7 @@ def get_current_user_permissions():
'material_list:isEnabled',
'material_list:operation'
]
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms

3
inventory-backend/app/api/v1/inbound/buy.py
View File

@ -18,6 +18,7 @@ def get_current_user_permissions():
from app.services.auth_service import AuthService
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
@ -25,7 +26,7 @@ def get_current_user_permissions():
# 返回所有以 inbound_buy: 开头的权限码（这里我们返回一个特殊标记，表示全部）
# 为了简单，我们返回 ['inbound_buy:*']，在过滤函数中特殊处理
return ['inbound_buy:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms

3
inventory-backend/app/api/v1/inbound/product.py
View File

@ -12,9 +12,10 @@ def get_current_user_permissions():
from app.services.auth_service import AuthService
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role: return []
if user_role.upper() == 'SUPER_ADMIN': return ['inbound_product:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
return perm_dict.get('menus', []) + perm_dict.get('elements', [])
def filter_item_by_permissions(item_dict, user_permissions):

3
inventory-backend/app/api/v1/inbound/semi.py
View File

@ -12,9 +12,10 @@ def get_current_user_permissions():
from app.services.auth_service import AuthService
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role: return []
if user_role.upper() == 'SUPER_ADMIN': return ['inbound_semi:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
return perm_dict.get('menus', []) + perm_dict.get('elements', [])
def filter_item_by_permissions(item_dict, user_permissions):

3
inventory-backend/app/api/v1/inbound/service.py
View File

@ -19,12 +19,13 @@ def get_current_user_permissions():
from app.services.auth_service import AuthService
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
if user_role.upper() == 'SUPER_ADMIN':
return ['inbound_service:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms

6
inventory-backend/app/api/v1/outbound.py
View File

@ -20,12 +20,13 @@ def get_current_user_permissions():
from app.services.auth_service import AuthService
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
if user_role.upper() == 'SUPER_ADMIN':
return ['outbound_list:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms
@ -116,12 +117,13 @@ def create_outbound():
# 权限检查：需要 outbound_create:operation 或 outbound_selection:operation 之一
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return jsonify({'code': 403, 'msg': '未授权'}), 403
# 超级管理员直接放行
if user_role.upper() != 'SUPER_ADMIN':
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
if ('outbound_create:operation' not in perms) and ('outbound_selection:operation' not in perms):
return jsonify({'code': 403, 'msg': '权限不足'}), 403

30
inventory-backend/app/api/v1/permission.py
View File

@ -1,32 +1,42 @@
# inventory-backend/app/api/v1/permission.py
from flask import Blueprint, request, jsonify, current_app
from flask_jwt_extended import jwt_required
from flask_jwt_extended import jwt_required, get_jwt
from app.services.permission_service import PermissionService
from app.utils.decorators import audit_log
from app.utils.decorators import permission_required, audit_log
permission_bp = Blueprint('permission', __name__)
def _get_operator_company():
"""从 JWT 获取当前操作者公司（None=超管）"""
claims = get_jwt()
role = claims.get('role', '')
if role and role.upper() == 'SUPER_ADMIN':
return None # 超管不限制公司
return claims.get('company_name', '')
@permission_bp.route('/tree', methods=['GET'])
@jwt_required()
@permission_required('system_permission')
def get_tree():
"""获取权限树"""
try:
data = PermissionService.get_permission_tree()
return jsonify({'code': 200, 'msg': '获取成功', 'data': data}), 200
except Exception as e:
# 打印详细错误到控制台，方便调试
current_app.logger.error(f"Get Tree Failed: {str(e)}")
# 返回 500 时带上错误信息
return jsonify({'code': 500, 'msg': f'服务器内部错误: {str(e)}'}), 500
@permission_bp.route('/role/<string:role_code>', methods=['GET'])
@jwt_required()
@permission_required('system_permission')
def get_role_perms(role_code):
"""获取某个角色的权限列表"""
"""获取某个角色的权限列表（已选中的）"""
try:
data = PermissionService.get_role_permissions(role_code)
company_name = _get_operator_company()
data = PermissionService.get_role_permissions(role_code, company_name=company_name)
return jsonify({'code': 200, 'msg': '获取成功', 'data': data}), 200
except Exception as e:
current_app.logger.error(f"Get Role Perms Failed: {str(e)}")
@ -35,19 +45,21 @@ def get_role_perms(role_code):
@permission_bp.route('/assign', methods=['POST'])
@jwt_required()
@permission_required('system_permission:operation')
@audit_log(
module='权限管理',
action='分配',
get_target_name_fn=lambda: request.get_json().get('role_code') if request.get_json() else None
)
def assign_perms():
"""保存权限分配"""
"""保存权限分配（自动带上当前操作者的公司标识）"""
try:
data = request.get_json()
role_code = data.get('role_code')
permissions = data.get('permissions', []) # list of codes
permissions = data.get('permissions', [])
PermissionService.assign_permissions(role_code, permissions)
company_name = _get_operator_company()
PermissionService.assign_permissions(role_code, permissions, company_name=company_name)
return jsonify({'code': 200, 'msg': '保存成功'}), 200
except Exception as e:
current_app.logger.error(f"Assign Perms Failed: {str(e)}")

6
inventory-backend/app/api/v1/scrap.py
View File

@ -24,11 +24,12 @@ def get_current_user_permissions():
from app.services.auth_service import AuthService
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return []
if user_role.upper() == 'SUPER_ADMIN':
return ['scrap_list:*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms
@ -71,12 +72,13 @@ def scan_barcode():
def create_scrap():
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return jsonify({'code': 403, 'msg': '未授权'}), 403
# 超级管理员直接放行
if user_role.upper() != 'SUPER_ADMIN':
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
if 'scrap_create:operation' not in perms:
return jsonify({'code': 403, 'msg': '权限不足'}), 403

3
inventory-backend/app/api/v1/transactions.py
View File

@ -19,12 +19,13 @@ def get_current_user_permissions():
"""
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
if user_role.upper() == 'SUPER_ADMIN':
return ['*']
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
# 合并菜单和元素权限
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
return perms

3
inventory-backend/app/models/system.py
View File

@ -147,12 +147,13 @@ class SysElement(db.Model):
class SysRolePermission(db.Model):
"""角色权限关联表"""
"""角色权限关联表（支持公司级别隔离）"""
__tablename__ = 'sys_role_permission'
id = db.Column(db.Integer, primary_key=True)
role_code = db.Column(db.String(50), nullable=False)
target_code = db.Column(db.String(100), nullable=False) # menu_code 或 element_code
type = db.Column(db.String(20), nullable=False) # 'menu' 或 'element'
company_name = db.Column(db.String(255), comment='所属公司（NULL/空/System=全局模板，具体值=公司定制）')
# ==========================================

78
inventory-backend/app/services/auth_service.py
View File

@ -1,7 +1,7 @@
# app/services/auth_service.py
from app.models.system import SysUser, SysRolePermission # <== 引入 SysRolePermission
from app.extensions import db, redis_client, revoke_all_tokens_for_user
from sqlalchemy import func
from sqlalchemy import func, or_
from flask_jwt_extended import create_access_token, create_refresh_token, get_jwt_identity
from flask import current_app
from app.utils.constants import UserRole
@ -350,23 +350,43 @@ class AuthService:
return user.to_dict()
@staticmethod
def get_all_users():
"""获取所有系统用户"""
users = SysUser.query.order_by(SysUser.id.desc()).all()
def get_all_users(company_name=None):
"""
获取系统用户列表（支持公司级别隔离）
参数:
company_name — None=不过滤(超管), 具体值=仅返回该公司的用户
"""
query = SysUser.query.order_by(SysUser.id.desc())
if company_name:
query = query.filter(SysUser.department == company_name)
users = query.all()
return [user.to_dict() for user in users]
@staticmethod
def delete_user(user_id, operator_role):
"""删除用户：删除前自动吊销该用户所有 JWT Token"""
# 标准化操作者角色为全大写
def delete_user(user_id, operator_role, operator_company=None):
"""
删除用户：删除前自动吊销该用户所有 JWT Token
权限检查：
- SUPER_ADMIN → 可删除任何用户
- SUPERVISOR → 仅可删除同公司（department）的用户
"""
operator_role_upper = operator_role.upper() if operator_role else None
if operator_role_upper != UserRole.SUPER_ADMIN:
raise Exception("权限不足：只有超级管理员可以删除用户")
user = SysUser.query.get(user_id)
if not user:
raise Exception("用户不存在")
# 权限校验
if operator_role_upper == UserRole.SUPER_ADMIN:
pass # 超管可删除任何人
elif operator_role_upper == UserRole.SUPERVISOR:
if operator_company and user.department != operator_company:
raise Exception("权限不足：主管只能删除本公司的用户")
else:
raise Exception("权限不足：只有超级管理员或主管可以删除用户")
# 提前获取用户名用于审计日志
username = user.username
@ -386,47 +406,67 @@ class AuthService:
return username
@staticmethod
def get_user_permissions(role_code):
def get_user_permissions(role_code, company_name=None):
"""
获取指定角色的所有权限代码列表
获取指定角色的所有权限代码列表（支持公司级别隔离）
参数:
role_code — 角色代码
company_name — 所属公司（None=不过滤/超管，具体值=只查该公司权限+全局模板权限）
返回格式: {
'menus': ['inbound_buy', 'system_user'],
'elements': ['inbound_buy:unit_price', ...]
}
公司过滤逻辑：
- 若无 company_name → 不过滤（兼容旧逻辑 / 超级管理员）
- 若有 company_name → 查「公司定制权限」+「全局模板权限（company_name 为空/NULL/System）」
"""
# 防御性编程：role_code 为空时直接返回空权限，避免后续 SQL 崩溃
# 防御性编程：role_code 为空时直接返回空权限
if not role_code:
return {'menus': [], 'elements': []}
# 超级管理员返回所有权限（通配符）
from app.utils.constants import UserRole
if role_code.upper() == UserRole.SUPER_ADMIN:
# 返回通配符，表示拥有所有菜单和元素权限
return {
'menus': ['*'],
'elements': ['*']
}
# 构建公司过滤条件：(公司定制 OR 全局模板)
if company_name:
company_filter = or_(
SysRolePermission.company_name == company_name,
SysRolePermission.company_name.is_(None),
SysRolePermission.company_name == '',
SysRolePermission.company_name == 'System'
)
else:
company_filter = True # 不过滤
# 1. 查菜单权限
# 使用 func.upper() 处理数据库字段的大小写
menu_perms = SysRolePermission.query.filter(
func.upper(SysRolePermission.role_code) == role_code.upper(),
SysRolePermission.type == 'menu'
SysRolePermission.type == 'menu',
company_filter
).all()
menu_codes = [p.target_code for p in menu_perms]
# 2. 查元素(列)权限
# 注意：这里我们只返回用户拥有的。前端逻辑是："如果列配置了Key且用户没这个Key，则隐藏"
element_perms = SysRolePermission.query.filter(
func.upper(SysRolePermission.role_code) == role_code.upper(),
SysRolePermission.type == 'element'
SysRolePermission.type == 'element',
company_filter
).all()
element_codes = [p.target_code for p in element_perms]
# 调试日志：输出查询结果便于排查字段权限问题
# 调试日志
from flask import current_app
current_app.logger.info(
f"[权限查询] role={role_code}, 查询到菜单权限={menu_codes}, 元素权限={element_codes}"
f"[权限查询] role={role_code}, company={company_name or '(all)'}, "
f"菜单权限={menu_codes}, 元素权限={element_codes}"
)
return {

3
inventory-backend/app/services/dify_permission_service.py
View File

@ -133,7 +133,8 @@ class DifyPermissionService:
"""
if not role:
role = DifyPermissionService.get_user_role()
return AuthService.get_user_permissions(role)
# Dify AI 场景不传 company_name，使用全局模板权限
return AuthService.get_user_permissions(role, company_name=None)
@staticmethod
def get_target_codes(role: str = None) -> list:

65
inventory-backend/app/services/inbound/base_service.py
View File

@ -216,39 +216,13 @@ class MaterialBaseService:
))
# ============================================================
# 【行级数据隔离】基于 JWT 中的 company_name 进行过滤
# 【行级数据隔离】基于 JWT 多租户公司过滤
# ============================================================
from flask_jwt_extended import get_jwt
claims = get_jwt()
user_role = claims.get('role', '').upper() if claims.get('role') else ''
user_company = claims.get('company_name', '')
from app.utils.decorators import get_current_company_filter
# 获取用户权限列表（用于检查 global:cross_company_op 特权）
from app.api.v1.inbound.base import get_current_user_permissions
user_perms = get_current_user_permissions() or []
normalized_perms = set(p.lower().replace('_', '').replace(':', '') for p in user_perms)
# 检查是否拥有全局特权或超管角色
has_cross_company = 'globalcrosscompanyop' in normalized_perms
# 获取前端传的查询参数
req_company = filters.get('company') if filters else None
if user_role != 'SUPER_ADMIN' and not has_cross_company:
# 【显式拒绝越权】如果前端传了公司参数，且不是当前用户的公司，返回403
if req_company and req_company != user_company:
from flask import abort
abort(403, description=f'越权访问：您无权查询 {req_company} 的数据')
# 正常查询本公司数据
if user_company:
query = query.filter(MaterialBase.company_name == user_company)
# 如果用户没有所属公司字段，则只显示公司为空的记录（或不允许查看）
elif user_role == 'SUPER_ADMIN' or has_cross_company:
# 超级管理员或有跨域特权：允许跨公司视角
if req_company:
query = query.filter(MaterialBase.company_name == req_company)
# 没选公司则不加过滤，看到全量
company_limit = get_current_company_filter()
if company_limit is not None:
query = query.filter(MaterialBase.company_name == company_limit)
category = filters.get('category')
if category is not None and category != '':
@ -750,32 +724,13 @@ class MaterialBaseService:
MaterialBase.company_name.ilike(kw)
))
# ============================================================
# 【行级数据隔离】基于 JWT 中的 company_name 进行过滤（高级筛选）
# 【行级数据隔离】基于 JWT 多租户公司过滤
# ============================================================
from flask_jwt_extended import get_jwt
claims = get_jwt()
user_role = claims.get('role', '').upper() if claims.get('role') else ''
user_company = claims.get('company_name', '')
# 获取用户权限列表（用于检查 global:cross_company_op 特权）
from app.api.v1.inbound.base import get_current_user_permissions
user_perms = get_current_user_permissions() or []
normalized_perms = set(p.lower().replace('_', '').replace(':', '') for p in user_perms)
# 检查是否拥有全局特权或超管角色
has_cross_company = 'globalcrosscompanyop' in normalized_perms
from app.utils.decorators import get_current_company_filter
req_company = filters.get('company') if filters else None
if user_role != 'SUPER_ADMIN' and not has_cross_company:
# 普通用户：强制隔离
if user_company:
filter_conditions.append(MaterialBase.company_name == user_company)
elif user_role == 'SUPER_ADMIN' or has_cross_company:
# 超级管理员或有跨域特权：允许跨公司视角
if req_company:
filter_conditions.append(MaterialBase.company_name == req_company)
company_limit = get_current_company_filter()
if company_limit is not None:
filter_conditions.append(MaterialBase.company_name == company_limit)
category = filters.get('category')
if category is not None and category != '':

32
inventory-backend/app/services/inbound/buy_service.py
View File

@ -396,33 +396,13 @@ class BuyInboundService:
query = query.filter(MaterialBase.material_type == material_type.strip())
# ============================================================
# 【行级数据隔离】基于 JWT 中的 company_name 进行过滤
# 【行级数据隔离】基于 JWT 多租户公司过滤
# ============================================================
from flask_jwt_extended import get_jwt
claims = get_jwt()
user_role = claims.get('role', '').upper() if claims.get('role') else ''
user_company = claims.get('company_name', '')
from app.utils.decorators import get_current_company_filter
# 获取用户权限列表（用于检查 global:cross_company_op 特权）
from app.api.v1.inbound.base import get_current_user_permissions
user_perms = get_current_user_permissions() or []
normalized_perms = set(p.lower().replace('_', '').replace(':', '') for p in user_perms)
# 检查是否拥有全局特权或超管角色
has_cross_company = 'globalcrosscompanyop' in normalized_perms
if user_role != 'SUPER_ADMIN' and not has_cross_company:
# 无特权：严禁查其他公司，强制绑定本公司
if company and company.strip() and company.strip() != user_company:
from flask import abort
abort(403, description=f'越权访问：您无权查询 {company} 的数据')
if user_company:
query = query.filter(MaterialBase.company_name == user_company)
elif user_role == 'SUPER_ADMIN' or has_cross_company:
# 有特权：允许下拉框传过来的 company 参数生效
if company and company.strip():
query = query.filter(MaterialBase.company_name == company.strip())
company_limit = get_current_company_filter()
if company_limit is not None:
query = query.filter(MaterialBase.company_name == company_limit)
# 4. 状态筛选
if not statuses: statuses = ['在库', '借库']
@ -451,6 +431,7 @@ class BuyInboundService:
'qty_stock': StockBuy.stock_quantity,
'qty_available': StockBuy.available_quantity,
'unit_price': StockBuy.pre_tax_unit_price,
'post_tax_unit_price': StockBuy.post_tax_unit_price,
'total_price': StockBuy.total_price,
'tax_rate': StockBuy.tax_rate,
'currency': StockBuy.currency,
@ -515,6 +496,7 @@ class BuyInboundService:
'qty_available': StockBuy.available_quantity,
'warehouse_loc': StockBuy.warehouse_location,
'unit_price': StockBuy.pre_tax_unit_price,
'post_tax_unit_price': StockBuy.post_tax_unit_price,
'total_price': StockBuy.total_price,
'tax_rate': StockBuy.tax_rate,
'currency': StockBuy.currency,

73
inventory-backend/app/services/permission_service.py
View File

@ -1,7 +1,7 @@
from app.models.system import SysMenu, SysElement, SysRolePermission
from app.extensions import db
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import func
from sqlalchemy import func, or_
class PermissionService:
@ -60,83 +60,106 @@ class PermissionService:
return tree_data
@staticmethod
def get_role_permissions(role_code):
"""获取指定角色拥有的所有权限Code"""
def get_role_permissions(role_code, company_name=None):
"""
获取指定角色拥有的所有权限Code（支持公司级别隔离）
参数:
company_name — None=不过滤, 具体值=查公司定制+全局模板权限
"""
try:
# === 新增逻辑：超级管理员上帝模式 ===
# === 超级管理员上帝模式 ===
if role_code == 'SUPER_ADMIN':
# 直接获取所有可见菜单和元素，无视配置表
all_menus = [m.code for m in SysMenu.query.filter(SysMenu.is_visible == True).all()]
all_elements = [e.code for e in SysElement.query.all()]
return {
'menus': all_menus,
'elements': all_elements
}
# =================================
perms = SysRolePermission.query.filter_by(role_code=role_code).all()
# 构建公司过滤条件
if company_name:
company_filter = or_(
SysRolePermission.company_name == company_name,
SysRolePermission.company_name.is_(None),
SysRolePermission.company_name == '',
SysRolePermission.company_name == 'System'
)
else:
company_filter = True
perms = SysRolePermission.query.filter(
SysRolePermission.role_code == role_code,
company_filter
).all()
menu_codes = []
element_codes = []
for p in perms:
# 这里假设你的数据库存的是 target_code
if p.type == 'menu':
menu_codes.append(p.target_code)
else:
element_codes.append(p.target_code)
# 前端 handleRoleSelect 会合并这两个数组，所以分开返回没问题
return {
'menus': menu_codes,
'elements': element_codes
}
except Exception as e:
# 记录日志或处理错误
print(f"Error fetching role permissions: {e}")
return {'menus': [], 'elements': []}
@staticmethod
def assign_permissions(role_code, permissions):
def assign_permissions(role_code, permissions, company_name=None):
"""
保存角色的权限
permissions: 前端传来的 list，混合了 menu_code 和 element_code
保存角色的权限（支持公司级别隔离）
参数:
role_code — 角色代码
permissions — 前端传来的 list，混合了 menu_code 和 element_code
company_name — 所属公司（None=全局模板，具体值=公司定制权限）
"""
if not role_code:
raise ValueError("角色代码不能为空")
session = db.session
try:
# 1. 开启事务 (Flask-SQLAlchemy 自动管理，但明确逻辑更好)
# 1. 删除该角色+该公司的旧权限（只删同公司的，不影响全局模板和其他公司）
if company_name:
old_perms = SysRolePermission.query.filter_by(
role_code=role_code, company_name=company_name
).all()
else:
# 公司为空时兼容旧逻辑：删除 company_name 为 NULL 的旧权限
old_perms = SysRolePermission.query.filter(
SysRolePermission.role_code == role_code,
or_(
SysRolePermission.company_name.is_(None),
SysRolePermission.company_name == ''
)
).all()
# 2. 删除该角色旧的所有权限（改为对象级删除以触发审计事件）
old_perms = SysRolePermission.query.filter_by(role_code=role_code).all()
for p in old_perms:
db.session.delete(p)
# 3. 准备新数据
# 2. 准备新数据
if permissions:
# 3.1 去重
unique_codes = set(permissions)
# 3.2 预加载所有 Menu Code，用于区分是 Menu 还是 Element
# 这一步很重要，因为 SysRolePermission 表需要 type 字段
all_menu_codes = {res[0] for res in session.query(SysMenu.code).all()}
new_records = []
for code in unique_codes:
if not code: continue
# 判断类型：如果 code 存在于菜单表中，就是 menu，否则就是 element
p_type = 'menu' if code in all_menu_codes else 'element'
new_records.append(SysRolePermission(
role_code=role_code,
target_code=code,
type=p_type
type=p_type,
company_name=company_name # ★ 存储公司标识
))
# 3.3 批量插入
if new_records:
session.add_all(new_records)

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

@ -149,9 +149,12 @@ def permission_required(permission_code):
if user_role and user_role.upper() == 'SUPER_ADMIN':
return fn(*args, **kwargs)
# 获取当前用户公司，用于权限隔离
user_company = claims.get('company_name', '')
try:
from app.services.auth_service import AuthService
perm_dict = AuthService.get_user_permissions(user_role)
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
except Exception as e:
logging.warning(f"Failed to fetch permissions for role {user_role}: {e}")
return jsonify(msg='权限查询失败'), 403
@ -164,6 +167,46 @@ def permission_required(permission_code):
return decorator
return wrapper
def get_current_company_filter():
"""
多租户数据权限隔离工具函数。
根据当前 JWT 中的 role 和 company_name 决定 SQL 查询应过滤到哪个公司。
返回值:
None → 不限制公司（超级管理员全局视角，或显式传了 'ALL'）
str → 仅查询该公司数据
使用示例:
company_limit = get_current_company_filter()
if company_limit is not None:
query = query.filter(MaterialBase.company_name == company_limit)
"""
from flask import request
claims = get_jwt()
user_role = claims.get('role', '')
# 规范化角色为大写
if user_role:
user_role = user_role.upper()
# 从请求参数获取前端指定的公司（兼容 company_name 和 company 两个参数名）
req_company = request.args.get('company_name', '') or request.args.get('company', '')
if user_role == 'SUPER_ADMIN':
# 超级管理员：允许通过参数指定公司过滤
if req_company and req_company.strip().upper() != 'ALL':
return req_company.strip()
# 未指定或指定了 ALL → 返回 None（全量，不限制）
return None
else:
# 其他角色（SUPERVISOR / FINANCE / WAREHOUSE_MGR 等）：
# 强制隔离到 JWT 中记录的公司，无视前端参数
user_company = claims.get('company_name', '')
return user_company if user_company else None
def audit_log(module: str = None, action: str = None, get_target_id_fn=None, get_target_name_fn=None, get_details_fn=None):
"""
已废弃！