"""客服后台菜单可见性(侧栏展示,不影响各页面原有权限校验)。""" import logging from types import SimpleNamespace from backend.utils import _AllPermissions, verify_kefu_permission from jituan.models import KefuMenuPage from jituan.services.admin_context import ( build_admin_club_context, can_manage_admin_assignments, is_system_super_admin, ) from jituan.services.club_rbac import user_bound_role_names from jituan.services.kefu_menu_definitions import KEFU_MENU_ROW_DEFS logger = logging.getLogger(__name__) def _is_super_permissions(permissions): return isinstance(permissions, _AllPermissions) def _has_super_perm(permissions): """俱乐部超级管理权限 000001(角色/后台用户等),不含集团级配置。""" if _is_super_permissions(permissions): return True return '000001' in permissions def _page_visible(page, permissions, user): if page.require_group_manage: return can_manage_admin_assignments(user) if page.require_super: return _has_super_perm(permissions) or is_system_super_admin(user) if _is_super_permissions(permissions): return True codes = page.perm_codes or [] if not codes: return False return any(code in permissions for code in codes) def _permission_codes_list(permissions): if _is_super_permissions(permissions): return ['000001'] return list(permissions) if permissions else [] def _builtin_menu_pages(): """DB 无数据时的内置菜单(与 seed_kefu_menu 定义一致)。""" pages = [] for row in KEFU_MENU_ROW_DEFS: pages.append(SimpleNamespace( page_id=row['page_id'], name=row['name'], path=row.get('path', ''), parent_id=row.get('parent_id', ''), sort_order=row.get('sort_order', 0), perm_codes=row.get('perm_codes', []), require_super=row.get('require_super', False), require_group_manage=row.get('require_group_manage', False), )) return pages def _load_menu_pages(): """从 DB 加载菜单;失败或空表时尝试 seed,仍空则用内置定义。""" pages = [] menu_source = 'db' try: pages = KefuMenuPage.objects.all().order_by('sort_order', 'page_id') pages = list(pages) except Exception as e: logger.warning('KefuMenuPage.objects 读取失败: %s', e) pages = [] if not pages: try: pages = KefuMenuPage.query.order_by('sort_order', 'page_id').to_list() except Exception as e: logger.warning('KefuMenuPage.query 读取失败: %s', e) pages = [] need_seed = not pages or not any(p.page_id == 'miniapp.assets' for p in pages) if need_seed: try: from django.core.management import call_command call_command('seed_kefu_menu') pages = list(KefuMenuPage.objects.all().order_by('sort_order', 'page_id')) except Exception as e: logger.warning('seed_kefu_menu 自动执行失败: %s', e) if not pages: pages = _builtin_menu_pages() menu_source = 'builtin' logger.warning('kefu_menu_page 表无数据,使用内置菜单定义(请执行 migrate jituan + seed_kefu_menu)') return pages, menu_source def build_menu_access_payload(request, username=None): """ 根据用户 gvsdsdk 权限计算可见菜单 page_id 列表。 返回 (payload, error_response);error_response 非空时表示校验失败。 """ phone = (username or getattr(request.user, 'Phone', '') or '').strip() kefu, permissions = verify_kefu_permission(request, phone) if kefu is None: return None, permissions pages, menu_source = _load_menu_pages() visible_ids = [] for page in pages: if not page.path: continue if _page_visible(page, permissions, request.user): visible_ids.append(page.page_id) for page in pages: if page.path: continue children = [p for p in pages if p.parent_id == page.page_id and p.path] if any(child.page_id in visible_ids for child in children): visible_ids.append(page.page_id) club_ctx = build_admin_club_context(request.user) path_by_id = {p.page_id: p.path for p in pages if p.path} visible_paths = [path_by_id[pid] for pid in visible_ids if pid in path_by_id] return { 'visible_page_ids': sorted(set(visible_ids)), 'visible_paths': visible_paths, 'permission_codes': _permission_codes_list(permissions), 'role_names': user_bound_role_names(request.user), 'yonghuid': getattr(request.user, 'UserUID', '') or '', 'can_switch_club': bool(club_ctx.get('can_switch_club')), 'is_group_admin': bool(club_ctx.get('is_group_admin')), 'menu_ready': True, 'menu_source': menu_source, }, None