140 lines
4.7 KiB
Python
140 lines
4.7 KiB
Python
"""客服后台菜单可见性（侧栏展示，不影响各页面原有权限校验）。"""
|
||
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.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),
|
||
'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
|