From 0b75740ae01e0ca94f5c8f981095f033520a612b Mon Sep 17 00:00:00 2001 From: yueli Date: Thu, 16 Jul 2026 11:25:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=9D=83=E9=99=90=E7=B3=BB=E7=BB=9F?= =?UTF-8?q?=E9=87=8D=E6=9E=84=20=E2=80=94=20=E6=93=8D=E4=BD=9C=E6=9D=83?= =?UTF-8?q?=E9=99=90=E5=8F=8C=E5=90=91=E5=B1=95=E5=BC=80=20+=20=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=E6=97=B6=E8=87=AA=E5=8A=A8=E8=A1=A5=E5=85=A8=E9=BB=98?= =?UTF-8?q?=E8=AE=A4=E6=9D=83=E9=99=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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() --- inventory-backend/app/__init__.py | 2 + .../app/services/permission_service.py | 171 +++++++++++++++++- inventory-backend/app/utils/decorators.py | 47 ++++- 3 files changed, 217 insertions(+), 3 deletions(-) diff --git a/inventory-backend/app/__init__.py b/inventory-backend/app/__init__.py index eb6b230..d2c41de 100644 --- a/inventory-backend/app/__init__.py +++ b/inventory-backend/app/__init__.py @@ -197,6 +197,8 @@ def create_app(): PermissionService.init_stocktake_menus() # 初始化所有菜单的层级结构 PermissionService.init_all_menus() + # ★ 启动时自动为所有已知角色补充默认权限（仅补充空角色，不覆盖已有） + PermissionService.ensure_default_permissions() except Exception as e: print(f"⚠️ 菜单初始化跳过: {e}") diff --git a/inventory-backend/app/services/permission_service.py b/inventory-backend/app/services/permission_service.py index f0f25b9..55b42a0 100644 --- a/inventory-backend/app/services/permission_service.py +++ b/inventory-backend/app/services/permission_service.py @@ -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 sqlalchemy.exc import SQLAlchemyError from sqlalchemy import func, or_ @@ -496,6 +496,7 @@ class PermissionService: ('material_base', '基础信息', '/material/index', 'material_mgmt', 1), # 入库管理子菜单 + ('inbound_purchase', '采购申请', '/purchase', 'inventory_mgmt', 0), ('inbound_buy', '采购入库', '/inventory/buy', 'inventory_mgmt', 1), ('inbound_semi', '半成品入库', '/inventory/semi', 'inventory_mgmt', 2), ('inbound_product', '成品入库', '/inventory/product', 'inventory_mgmt', 3), @@ -637,6 +638,20 @@ class PermissionService: db.session.add(new_perm) 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"✅ 所有菜单初始化完成") return True @@ -644,3 +659,157 @@ class PermissionService: db.session.rollback() print(f"❌ 初始化菜单失败: {str(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}") diff --git a/inventory-backend/app/utils/decorators.py b/inventory-backend/app/utils/decorators.py index fb9fa55..733c3c6 100644 --- a/inventory-backend/app/utils/decorators.py +++ b/inventory-backend/app/utils/decorators.py @@ -126,6 +126,44 @@ def login_required(fn): return fn(*args, **kwargs) 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 wrapper(fn): @@ -160,8 +198,13 @@ def permission_required(permission_code): return jsonify(msg='权限查询失败'), 403 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 fn(*args, **kwargs) return decorator