feat: 权限系统重构 — 操作权限双向展开 + 启动时自动补全默认权限

## permission_service.py
- init_all_menus: menu_defs 新增 inbound_purchase(采购申请) 菜单项
- init_all_menus: 创建 inbound_purchase:operation 操作元素
- 新增 ensure_default_permissions(): 启动时从 sys_user 提取活跃角色,
  对空权限角色自动补全默认菜单(按角色类型分级)
- 自动迁移: 已有 inbound_buy 权限的角色 → 自动获得 inbound_purchase
- SUPERVISOR 启动时自动获得 inbound_purchase 菜单权限

## decorators.py
- _expand_operation_perms 改为双向桥接:
  正向(inbound_buy:operation←inbound_buy:write)→通过
  反向(inbound_buy←inbound_buy:operation)→通过(可编辑隐含可读)
- 新增详细诊断日志,记录展开路径和失败原因

## __init__.py
- 启动时调用 PermissionService.ensure_default_permissions()
This commit is contained in:
yueli
2026-07-16 11:25:41 +08:00
parent 329820117f
commit 0b75740ae0
3 changed files with 217 additions and 3 deletions

View File

@ -197,6 +197,8 @@ def create_app():
PermissionService.init_stocktake_menus() PermissionService.init_stocktake_menus()
# 初始化所有菜单的层级结构 # 初始化所有菜单的层级结构
PermissionService.init_all_menus() PermissionService.init_all_menus()
# ★ 启动时自动为所有已知角色补充默认权限(仅补充空角色,不覆盖已有)
PermissionService.ensure_default_permissions()
except Exception as e: except Exception as e:
print(f"⚠️ 菜单初始化跳过: {e}") print(f"⚠️ 菜单初始化跳过: {e}")

View File

@ -1,4 +1,4 @@
from app.models.system import SysMenu, SysElement, SysRolePermission from app.models.system import SysMenu, SysElement, SysRolePermission, SysUser
from app.extensions import db from app.extensions import db
from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import func, or_ from sqlalchemy import func, or_
@ -496,6 +496,7 @@ class PermissionService:
('material_base', '基础信息', '/material/index', 'material_mgmt', 1), ('material_base', '基础信息', '/material/index', 'material_mgmt', 1),
# 入库管理子菜单 # 入库管理子菜单
('inbound_purchase', '采购申请', '/purchase', 'inventory_mgmt', 0),
('inbound_buy', '采购入库', '/inventory/buy', 'inventory_mgmt', 1), ('inbound_buy', '采购入库', '/inventory/buy', 'inventory_mgmt', 1),
('inbound_semi', '半成品入库', '/inventory/semi', 'inventory_mgmt', 2), ('inbound_semi', '半成品入库', '/inventory/semi', 'inventory_mgmt', 2),
('inbound_product', '成品入库', '/inventory/product', 'inventory_mgmt', 3), ('inbound_product', '成品入库', '/inventory/product', 'inventory_mgmt', 3),
@ -637,6 +638,20 @@ class PermissionService:
db.session.add(new_perm) db.session.add(new_perm)
db.session.commit() db.session.commit()
# ★ 采购申请操作权限元素
purchase_op = SysElement.query.filter_by(
menu_code='inbound_purchase',
code='inbound_purchase:operation'
).first()
if not purchase_op:
db.session.add(SysElement(
menu_code='inbound_purchase',
name='可编辑',
code='inbound_purchase:operation',
element_type='operation'
))
print(f"✅ 采购申请操作权限元素已创建")
print(f"✅ 所有菜单初始化完成") print(f"✅ 所有菜单初始化完成")
return True return True
@ -644,3 +659,157 @@ class PermissionService:
db.session.rollback() db.session.rollback()
print(f"❌ 初始化菜单失败: {str(e)}") print(f"❌ 初始化菜单失败: {str(e)}")
raise e raise e
@staticmethod
def ensure_default_permissions():
"""
启动时自动为所有已知角色补充默认权限(仅补充,不覆盖已有权限)。
解决的问题:
- 重新部署/重启后,非超级管理员角色的 sys_role_permission 表为空
- 管理员不必每次都手动重新分配权限
默认分配策略(保守,仅给菜单访问权,不给操作/编辑权):
- 遍历 sys_user 表中所有非空 role 值
- 对每个角色,赋予其对应模块的顶级菜单权限
- 已有权限的角色不会被覆盖
"""
try:
# 1. 收集所有已知角色(从 sys_user.distinct role)
roles = db.session.query(SysUser.role).filter(
SysUser.role.isnot(None),
SysUser.role != '',
SysUser.status == 'active'
).distinct().all()
known_roles = [r[0] for r in roles if r[0]]
if not known_roles:
print("[默认权限] 未找到活跃角色,跳过")
return
# 2. 角色 → 应拥有的菜单列表
ROLE_DEFAULT_MENUS = {
'WAREHOUSE_MGR': [
'material_mgmt', 'inventory_mgmt', 'stocktake_mgmt',
'outbound_mgmt', 'bom_mgmt', 'operation_mgmt', 'scrap_mgmt',
'system_mgmt',
],
'WAREHOUSE_ADMIN': [
'material_mgmt', 'inventory_mgmt', 'stocktake_mgmt',
'outbound_mgmt', 'bom_mgmt', 'operation_mgmt', 'scrap_mgmt',
'system_mgmt',
],
'SUPERVISOR': [
'material_mgmt', 'inventory_mgmt', 'stocktake_mgmt',
'outbound_mgmt', 'bom_mgmt', 'operation_mgmt', 'scrap_mgmt',
'system_mgmt',
],
'PURCHASER': ['material_mgmt', 'inventory_mgmt'],
'INBOUND': ['inventory_mgmt'],
'SALES': ['outbound_mgmt'],
'OUTBOUND': ['outbound_mgmt'],
'WAREHOUSE_OP': ['inventory_mgmt', 'outbound_mgmt'],
}
# ★ SUPERVISOR 主管默认拥有采购申请的子菜单权限(确保能看到所有人的申请)
SUPERVISOR_PURCHASE_MENUS = ['inbound_purchase']
total_inserted = 0
for role_code in known_roles:
# 跳过超级管理员(已有全权限)
if role_code.upper() == 'SUPER_ADMIN':
continue
# 检查该角色是否已有权限
existing_count = SysRolePermission.query.filter_by(
role_code=role_code
).count()
if existing_count > 0:
print(f"[默认权限] {role_code} 已有 {existing_count} 条权限,跳过")
continue
# 获取该角色应获得的默认菜单
default_menus = ROLE_DEFAULT_MENUS.get(
role_code,
['material_mgmt', 'inventory_mgmt', 'outbound_mgmt'] # 未知角色:只给基础三模块
)
for menu_code in default_menus:
# 验证菜单存在
menu = SysMenu.query.filter_by(code=menu_code).first()
if not menu:
continue
existing = SysRolePermission.query.filter_by(
role_code=role_code,
target_code=menu_code,
type='menu'
).first()
if existing:
continue
db.session.add(SysRolePermission(
role_code=role_code,
target_code=menu_code,
type='menu'
))
total_inserted += 1
print(f"[默认权限] {role_code} ← 已补充 {total_inserted} 条默认菜单权限")
# ★ 自动迁移:已有 inbound_buy 权限的角色 → 自动获得 inbound_purchase
for role_code in known_roles:
if role_code.upper() == 'SUPER_ADMIN':
continue
has_inbound_buy = SysRolePermission.query.filter_by(
role_code=role_code, target_code='inbound_buy', type='menu'
).first()
if has_inbound_buy:
has_purchase = SysRolePermission.query.filter_by(
role_code=role_code, target_code='inbound_purchase', type='menu'
).first()
if not has_purchase:
db.session.add(SysRolePermission(
role_code=role_code, target_code='inbound_purchase', type='menu'
))
# 同时迁移操作权限
has_buy_op = SysRolePermission.query.filter_by(
role_code=role_code, target_code='inbound_buy:operation', type='element'
).first()
if has_buy_op:
has_purchase_op = SysRolePermission.query.filter_by(
role_code=role_code, target_code='inbound_purchase:operation', type='element'
).first()
if not has_purchase_op:
db.session.add(SysRolePermission(
role_code=role_code, target_code='inbound_purchase:operation', type='element'
))
print(f"[权限迁移] {role_code}: inbound_buy → inbound_purchase 已自动迁移")
total_inserted += 1
# ★ SUPERVISOR 主管默认获得采购申请子菜单权限
for role_code in known_roles:
if role_code.upper() == 'SUPERVISOR':
for menu_code in SUPERVISOR_PURCHASE_MENUS:
menu = SysMenu.query.filter_by(code=menu_code).first()
if not menu:
continue
existing = SysRolePermission.query.filter_by(
role_code=role_code, target_code=menu_code, type='menu'
).first()
if not existing:
db.session.add(SysRolePermission(
role_code=role_code, target_code=menu_code, type='menu'
))
total_inserted += 1
print(f"[默认权限] SUPERVISOR ← 已补充采购申请菜单权限")
if total_inserted > 0:
db.session.commit()
print(f"[默认权限] 总计为 {sum(1 for r in known_roles if r.upper() != 'SUPER_ADMIN')} 个角色补充了默认权限")
except Exception as e:
db.session.rollback()
print(f"[默认权限] 补充失败(不阻断启动): {e}")

View File

@ -126,6 +126,44 @@ def login_required(fn):
return fn(*args, **kwargs) return fn(*args, **kwargs)
return decorator return decorator
def _expand_operation_perms(permission_code, all_perms):
"""
操作权限自动展开映射器(双向粒度桥接)。
"""
if permission_code in all_perms:
logging.info(f"[权限展开] 精确匹配: {permission_code}")
return True
KNOWN_OPERATION_SUFFIXES = (
':operation', ':add', ':edit', ':delete', ':submit',
':approve', ':dispatch', ':write'
)
if ':' in permission_code:
prefix = permission_code.rsplit(':', 1)[0]
else:
prefix = permission_code
# 收集该模块下的所有用户权限,用于诊断日志
module_perms = [p for p in all_perms if p.startswith(prefix)]
for perm in all_perms:
if perm.startswith(prefix + ':') or perm == prefix:
for suffix in KNOWN_OPERATION_SUFFIXES:
if perm.endswith(suffix):
logging.info(f"[权限展开] {permission_code} ← 用户有 {perm} → 通过")
return True
if ':' not in permission_code and ':' in perm:
logging.info(f"[权限展开] {permission_code} ← 用户有 {perm}(下级权限)→ 通过")
return True
logging.warning(
f"[权限展开] 失败: 要求={permission_code}, "
f"该模块用户权限={module_perms}, 全部权限数={len(all_perms)}"
)
return False
def permission_required(permission_code): def permission_required(permission_code):
"""检查当前用户是否拥有指定权限码,同时检查用户是否仍然有效""" """检查当前用户是否拥有指定权限码,同时检查用户是否仍然有效"""
def wrapper(fn): def wrapper(fn):
@ -160,8 +198,13 @@ def permission_required(permission_code):
return jsonify(msg='权限查询失败'), 403 return jsonify(msg='权限查询失败'), 403
all_perms = perm_dict.get('menus', []) + perm_dict.get('elements', []) all_perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
if permission_code not in all_perms:
logging.warning(f"权限检查失败: 角色={user_role}, 所需权限={permission_code}") # ★ 操作权限展开检查(粒度桥接)
if not _expand_operation_perms(permission_code, all_perms):
logging.warning(
f"权限检查失败: 角色={user_role}, 所需={permission_code}, "
f"拥有={[p for p in all_perms if permission_code.split(':')[0] in p]}"
)
return jsonify(msg='权限不足:您没有访问此资源的权限'), 403 return jsonify(msg='权限不足:您没有访问此资源的权限'), 403
return fn(*args, **kwargs) return fn(*args, **kwargs)
return decorator return decorator