From 42cb117a8a3e58eea24eae00ab71e55f54e9d535 Mon Sep 17 00:00:00 2001 From: XingQue Date: Sun, 28 Jun 2026 23:59:32 +0800 Subject: [PATCH] fix: kefu permissions use legacy permission table instead of gvsdsdk Co-authored-by: Cursor --- backend/utils.py | 76 ++++----- backend/view.py | 272 ++++++++++++++----------------- jituan/services/admin_context.py | 16 ++ jituan/services/club_rbac.py | 13 +- jituan/views.py | 6 +- users/views.py | 7 +- 6 files changed, 183 insertions(+), 207 deletions(-) diff --git a/backend/utils.py b/backend/utils.py index 32c9ea0..b69fc16 100644 --- a/backend/utils.py +++ b/backend/utils.py @@ -77,14 +77,19 @@ class _AllPermissions(list): return True -def _legacy_backend_role_permissions(phone): +def _legacy_backend_role_permissions(phone, yonghuid=None): """ 读取旧权限表（backend.models）： - user_role.account_id = 手机号 → role → role_permission → permission.perm_code - 与 gvsdsdk 表并行，未改 SDK 结构。 + user_role.account_id = 手机号或 UserUID → role → role_permission → permission.perm_code """ - phone = (phone or '').strip() - if not phone: + account_ids = [] + ph = (phone or '').strip() + uid = (yonghuid or '').strip() + if ph: + account_ids.append(ph) + if uid and uid not in account_ids: + account_ids.append(uid) + if not account_ids: return [], [] try: from backend.models import ( @@ -93,7 +98,7 @@ def _legacy_backend_role_permissions(phone): Permission as BackendPermission, ) role_ids = list( - LegacyUserRole.objects.filter(account_id=phone).values_list('role_id', flat=True) + LegacyUserRole.objects.filter(account_id__in=account_ids).values_list('role_id', flat=True) ) if not role_ids: return [], [] @@ -107,10 +112,22 @@ def _legacy_backend_role_permissions(phone): ) return perm_codes, role_names except Exception as e: - logger.warning('legacy_backend_role_permissions phone=%s err=%s', phone, e) + logger.warning('legacy_backend_role_permissions ids=%s err=%s', account_ids, e) return [], [] +def has_legacy_backend_binding(user): + """旧表 user_role 或 role 下已有 perm_code。""" + from backend.models import UserRole as LegacyUserRole + ph = getattr(user, 'Phone', '') or '' + uid = getattr(user, 'UserUID', '') or '' + ids = [x for x in [ph, uid] if x] + if ids and LegacyUserRole.objects.filter(account_id__in=ids).exists(): + return True + codes, _ = _legacy_backend_role_permissions(ph, uid) + return bool(codes) + + def verify_kefu_permission(request, username_from_frontend=None): """ 验证客服/管理员身份，并返回该用户的所有权限列表（基于新RBAC表） @@ -128,9 +145,10 @@ def verify_kefu_permission(request, username_from_frontend=None): # ---------- 1. 必须是客服或管理员 ---------- from jituan.constants import SUPER_ADMIN_PHONES + from jituan.services.admin_context import is_kefu_backend_account is_super_phone = getattr(current_user, 'Phone', '') in SUPER_ADMIN_PHONES is_admin = current_user.UserType == 'admin' or current_user.IsSuperuser or is_super_phone - if current_user.UserType != 'kefu' and not is_admin: + if not is_kefu_backend_account(current_user): return None, Response({'code': 403, 'msg': '身份错误，非客服账号'}, status=403) # ---------- 2. 客服扩展表存在且状态正常（管理员跳过） ---------- @@ -156,42 +174,12 @@ def verify_kefu_permission(request, username_from_frontend=None): 'msg': '身份错误，无权操作其他账号，相关操作已记录' }, status=403) - # ---------- 4. 获取用户的所有权限 ---------- - # 使用 gvsdsdk RBAC 表（UserRole → RolePermission → Permission） - try: - user_uuid = current_user.UserUUID - role_uuids = list(SdkUserRole.objects.filter( - UserUUID=user_uuid - ).values_list('RoleUUID', flat=True)) - - if role_uuids: - perm_uuids = list(SdkRolePermission.objects.filter( - RoleUUID__in=role_uuids - ).values_list('PermUUID', flat=True)) - - permissions = list(SdkPermission.objects.filter( - PermUUID__in=perm_uuids, - PermStatus=1 - ).values_list('PermCode', flat=True).distinct()) - else: - permissions = [] - except Exception as e: - logger.warning( - 'verify_kefu_permission failed: UserUUID=%s Phone=%s err=%s', - getattr(current_user, 'UserUUID', None), - getattr(current_user, 'Phone', None), - e, - exc_info=True, - ) - permissions = [] - - # 旧权限表（user_role 按手机号），与 gvsdsdk 合并 - legacy_codes, _ = _legacy_backend_role_permissions(current_user.Phone) - if legacy_codes: - if isinstance(permissions, _AllPermissions): - pass - else: - permissions = list(set(list(permissions) + legacy_codes)) + # ---------- 4. 权限码：只读旧表 permission（user_role → role → role_permission）---------- + legacy_codes, _ = _legacy_backend_role_permissions( + current_user.Phone, + getattr(current_user, 'UserUID', None), + ) + permissions = list(legacy_codes) if legacy_codes else [] # 管理员/超级用户自动拥有所有权限 if is_admin and '000001' not in permissions: diff --git a/backend/view.py b/backend/view.py index 1590d50..51fe91f 100644 --- a/backend/view.py +++ b/backend/view.py @@ -148,46 +148,62 @@ class GetRolePermissionView(APIView): if '000001' not in permissions: return Response({'code': 403, 'msg': '您无权访问此页面'}) - # 3. 获取所有权限 - all_perms = Permission.objects.filter(PermStatus=1).values('PermCode', 'PermName', 'PermDesc') - permissions_list = [_serialize_permission_row(p) for p in all_perms] - - from jituan.services.club_rbac import count_role_users_in_scope, filter_roles_for_request - - kefu_user_uuids = list( - User.objects.filter(KefuProfile__isnull=False).values_list('UserUUID', flat=True) + from backend.models import ( + Role as BackendRole, + Permission as BackendPermission, + UserRole as LegacyUserRole, ) - roles = filter_roles_for_request(request) + + all_perms = BackendPermission.objects.all().values('perm_code', 'perm_name', 'description') + permissions_list = [ + { + 'perm_code': p['perm_code'], + 'perm_name': p['perm_name'], + 'perm_desc': p.get('description') or '', + } + for p in all_perms + ] + roles_data = [] - for role in roles: - user_count = count_role_users_in_scope(role, request, kefu_user_uuids) - role_perm_uuids = RolePermission.objects.filter( - RoleUUID=role.RoleUUID - ).values_list('PermUUID', flat=True) - perms_of_role = [ - _serialize_permission_row(p) - for p in Permission.objects.filter( - PermUUID__in=list(role_perm_uuids) - ).values('PermCode', 'PermName', 'PermDesc') - ] + for role in BackendRole.objects.all().order_by('role_id'): + user_count = LegacyUserRole.objects.filter(role=role).count() + perms_of_role = BackendPermission.objects.filter(rolepermission__role=role).distinct() roles_data.append({ - 'role_code': role.RoleName, - 'role_name': role.RoleName, - 'description': role.RoleDesc or '', - 'club_id': getattr(role, 'ClubID', '') or '', + 'role_code': role.role_name, + 'role_name': role.role_name, + 'description': role.description or '', + 'club_id': '', 'user_count': user_count, - 'permissions': perms_of_role + 'permissions': [ + { + 'perm_code': p.perm_code, + 'perm_name': p.perm_name, + 'perm_desc': p.description or '', + } + for p in perms_of_role + ], }) return Response({ 'code': 0, 'data': { 'roles': roles_data, - 'permissions': permissions_list - } + 'permissions': permissions_list, + }, }) +def _resolve_backend_role(role_code): + from backend.models import Role as BackendRole + rc = (role_code or '').strip() + if not rc: + return None + return ( + BackendRole.objects.filter(role_name=rc).first() + or BackendRole.objects.filter(role_code=rc).first() + ) + + class ModifyRolePermissionView(APIView): """ 通用接口：支持删除角色、为角色添加权限、移除角色权限 @@ -213,22 +229,26 @@ class ModifyRolePermissionView(APIView): if not role_code: return Response({'code': 400, 'msg': '缺少角色编码'}) - try: - role = Role.objects.get(RoleName=role_code) - except Role.DoesNotExist: + from backend.models import ( + Permission as BackendPermission, + RolePermission as LegacyRolePermission, + UserRole as LegacyUserRole, + ) + + role = _resolve_backend_role(role_code) + if not role: return Response({'code': 404, 'msg': '角色不存在'}) # 超级管理员角色的特殊限制 - if role.RoleName == '管理员': + if role.role_name == '管理员': if action in ['delete_role']: return Response({'code': 403, 'msg': '超级管理员角色不能删除'}) # 根据 action 处理 if action == 'delete_role': - # 删除角色：同时删除角色权限关联、用户角色关联 with transaction.atomic(): - RolePermission.objects.filter(RoleUUID=role.RoleUUID).delete() - UserRole.objects.filter(RoleUUID=role.RoleUUID).delete() + LegacyRolePermission.objects.filter(role=role).delete() + LegacyUserRole.objects.filter(role=role).delete() role.delete() return Response({'code': 0, 'msg': '角色删除成功'}) @@ -236,34 +256,26 @@ class ModifyRolePermissionView(APIView): perm_code = request.data.get('perm_code') if not perm_code: return Response({'code': 400, 'msg': '缺少权限编码'}) - if role.RoleName == '管理员' and perm_code == '000001': + if role.role_name == '管理员' and perm_code == '000001': return Response({'code': 403, 'msg': '此权限已存在，不能重复添加'}) - try: - perm = Permission.objects.get(PermCode=perm_code) - except Permission.DoesNotExist: + perm = BackendPermission.objects.filter(perm_code=perm_code).first() + if not perm: return Response({'code': 404, 'msg': '权限不存在'}) - # 检查是否已存在 - if RolePermission.objects.filter(RoleUUID=role.RoleUUID, PermUUID=perm.PermUUID).exists(): + if LegacyRolePermission.objects.filter(role=role, permission=perm).exists(): return Response({'code': 400, 'msg': '该角色已拥有此权限'}) - RolePermission.objects.create( - RolePermUUID=uuid.uuid4().bytes, - RoleUUID=role.RoleUUID, - PermUUID=perm.PermUUID, - CreateTime=timezone.now(), - ) + LegacyRolePermission.objects.create(role=role, permission=perm) return Response({'code': 0, 'msg': '权限添加成功'}) elif action == 'remove_permission': perm_code = request.data.get('perm_code') if not perm_code: return Response({'code': 400, 'msg': '缺少权限编码'}) - if role.RoleName == '管理员' and perm_code == '000001': + if role.role_name == '管理员' and perm_code == '000001': return Response({'code': 403, 'msg': '不能移除超级管理员的此权限'}) - try: - perm = Permission.objects.get(PermCode=perm_code) - except Permission.DoesNotExist: + perm = BackendPermission.objects.filter(perm_code=perm_code).first() + if not perm: return Response({'code': 404, 'msg': '权限不存在'}) - deleted, _ = RolePermission.objects.filter(RoleUUID=role.RoleUUID, PermUUID=perm.PermUUID).delete() + deleted, _ = LegacyRolePermission.objects.filter(role=role, permission=perm).delete() if deleted == 0: return Response({'code': 400, 'msg': '该角色未拥有此权限'}) return Response({'code': 0, 'msg': '权限移除成功'}) @@ -272,45 +284,32 @@ class ModifyRolePermissionView(APIView): role_name = (request.data.get('role_name') or '').strip() description = (request.data.get('description') or '').strip() perm_codes = request.data.get('perm_codes') or [] - if role.RoleName == '管理员': - role_name = role.RoleName - elif role_name and role_name != role.RoleName: - if Role.objects.filter(RoleName=role_name).exclude(RoleUUID=role.RoleUUID).exists(): + from backend.models import Role as BackendRole + if role.role_name == '管理员': + role_name = role.role_name + elif role_name and role_name != role.role_name: + if BackendRole.objects.filter(role_name=role_name).exclude(role_id=role.role_id).exists(): return Response({'code': 400, 'msg': '角色名称已存在'}) with transaction.atomic(): - if role.RoleName != '管理员': + if role.role_name != '管理员': if role_name: - role.RoleName = role_name - role.RoleDesc = description - role.save(update_fields=['RoleName', 'RoleDesc']) - RolePermission.objects.filter(RoleUUID=role.RoleUUID).delete() + role.role_name = role_name + role.description = description + role.save(update_fields=['role_name', 'description']) + LegacyRolePermission.objects.filter(role=role).delete() for pc in perm_codes: - if pc == '000001' and role.RoleName != '管理员': + if pc == '000001' and role.role_name != '管理员': continue - try: - perm = Permission.objects.get(PermCode=pc, PermStatus=1) - except Permission.DoesNotExist: + perm = BackendPermission.objects.filter(perm_code=pc).first() + if not perm: continue - RolePermission.objects.create( - RolePermUUID=uuid.uuid4().bytes, - RoleUUID=role.RoleUUID, - PermUUID=perm.PermUUID, - CreateTime=timezone.now(), - ) - if role.RoleName == '管理员': - try: - super_perm = Permission.objects.get(PermCode='000001', PermStatus=1) - if not RolePermission.objects.filter( - RoleUUID=role.RoleUUID, PermUUID=super_perm.PermUUID, - ).exists(): - RolePermission.objects.create( - RolePermUUID=uuid.uuid4().bytes, - RoleUUID=role.RoleUUID, - PermUUID=super_perm.PermUUID, - CreateTime=timezone.now(), - ) - except Permission.DoesNotExist: - pass + LegacyRolePermission.objects.create(role=role, permission=perm) + if role.role_name == '管理员': + super_perm = BackendPermission.objects.filter(perm_code='000001').first() + if super_perm and not LegacyRolePermission.objects.filter( + role=role, permission=super_perm, + ).exists(): + LegacyRolePermission.objects.create(role=role, permission=super_perm) return Response({'code': 0, 'msg': '角色保存成功'}) else: @@ -339,47 +338,33 @@ class AddRoleView(APIView): if '000001' not in permissions: return Response({'code': 403, 'msg': '您无权进行此操作'}) - # 检查角色名称是否已存在（同俱乐部内唯一） - from jituan.services.club_rbac import resolve_role_club_id, role_name_exists - target_club_id = resolve_role_club_id( - request, (request.data.get('target_club_id') or '').strip() or None, + from backend.models import ( + Role as BackendRole, + Permission as BackendPermission, + RolePermission as LegacyRolePermission, ) - if role_name_exists(role_name, target_club_id): - return Response({'code': 400, 'msg': '该俱乐部下角色名称已存在'}) - # 生成唯一角色编码（格式：ROLE_ + 8位随机字母数字） + if BackendRole.objects.filter(role_name=role_name).exists(): + return Response({'code': 400, 'msg': '角色名称已存在'}) + role_code = f"ROLE_{''.join(random.choices(string.ascii_uppercase + string.digits, k=8))}" - # 创建角色 with transaction.atomic(): - role = Role.objects.create( - RoleUUID=uuid.uuid4().bytes, - TenantUUID=b'\x00' * 16, - RoleName=role_name, - RoleType=2, - RoleStatus=1, - AssignScope=0, - RoleDesc=description, - ClubID=target_club_id, - CreateTime=timezone.now(), + role = BackendRole.objects.create( + role_code=role_code, + role_name=role_name, + description=description, ) - # 绑定权限（过滤掉不存在的权限编码） for pc in perm_codes: - # 不允许添加 000001 权限（此权限仅供超级管理员） if pc == '000001': continue - try: - perm = Permission.objects.get(PermCode=pc) - RolePermission.objects.create( - RolePermUUID=uuid.uuid4().bytes, - RoleUUID=role.RoleUUID, - PermUUID=perm.PermUUID, - CreateTime=timezone.now(), - ) - except Permission.DoesNotExist: + perm = BackendPermission.objects.filter(perm_code=pc).first() + if not perm: logger.warning(f"添加角色时忽略不存在的权限编码: {pc}") + continue + LegacyRolePermission.objects.create(role=role, permission=perm) - return Response({'code': 0, 'msg': '角色添加成功', 'data': {'role_code': role_code}}) + return Response({'code': 0, 'msg': '角色添加成功', 'data': {'role_code': role_name}}) @@ -418,15 +403,14 @@ class GetAdminUserListView(APIView): if nicheng: users = users.filter(KefuProfile__nicheng__icontains=nicheng) - # 关键修正：先求值为列表，再传入 phone__in if role_code: - target_role = Role.objects.filter(RoleName=role_code).first() + from backend.models import Role as BackendRole, UserRole as LegacyUserRole + target_role = _resolve_backend_role(role_code) if target_role: - user_uuids = list( - UserRole.objects.filter(RoleUUID=target_role.RoleUUID) - .values_list('UserUUID', flat=True) + account_ids = list( + LegacyUserRole.objects.filter(role=target_role).values_list('account_id', flat=True) ) - users = users.filter(UserUUID__in=user_uuids) + users = users.filter(Phone__in=account_ids) else: users = users.none() @@ -435,16 +419,12 @@ class GetAdminUserListView(APIView): page_obj = paginator.get_page(page) data_list = [] + from backend.models import UserRole as LegacyUserRole for user in page_obj: - # 通过 gvsdsdk UserRole 获取用户角色 - user_role_uuids = list( - UserRole.objects.filter(UserUUID=user.UserUUID) - .values_list('RoleUUID', flat=True) - ) - user_roles = Role.objects.filter(RoleUUID__in=user_role_uuids) + user_roles = LegacyUserRole.objects.filter(account_id=user.Phone).select_related('role') roles_data = [ - {'role_code': r.RoleName, 'role_name': r.RoleName} - for r in user_roles + {'role_code': ur.role.role_name, 'role_name': ur.role.role_name} + for ur in user_roles ] data_list.append({ 'phone': user.Phone or '', @@ -490,16 +470,16 @@ class GetAdminRolesView(APIView): if '000001' not in permissions: return Response({'code': 403, 'msg': '您无权访问此页面'}) - from jituan.services.club_rbac import filter_roles_for_request - roles = filter_roles_for_request(request).values('RoleName', 'ClubID') + from backend.models import Role as BackendRole + roles = BackendRole.objects.all().order_by('role_id') return Response({ 'code': 0, 'data': { 'roles': [ { - 'role_code': r['RoleName'], - 'role_name': r['RoleName'], - 'club_id': r.get('ClubID') or '', + 'role_code': r.role_name, + 'role_name': r.role_name, + 'club_id': '', } for r in roles ] @@ -638,25 +618,22 @@ class ModifyAdminUserView(APIView): role_codes = request.data.get('role_codes', []) if not role_codes: return Response({'code': 400, 'msg': '请选择角色'}) - from jituan.services.club_rbac import resolve_role_by_name - club_id = getattr(target_user, 'ClubID', '') or '' + from backend.models import Role as BackendRole, UserRole as LegacyUserRole added, missing = [], [] for rc in role_codes: - role = resolve_role_by_name(rc, club_id) + role = _resolve_backend_role(rc) if not role: missing.append(rc) continue - UserRole.objects.get_or_create( - UserRoleUUID=uuid.uuid4().bytes, - UserUUID=target_user.UserUUID, - RoleUUID=role.RoleUUID, - CompanyUUID=uuid.UUID('b0000000-0000-0000-0000-000000000001').bytes, + LegacyUserRole.objects.get_or_create( + account_id=target_user.Phone, + role=role, ) - added.append(role.RoleName) + added.append(role.role_name) if not added: return Response({ 'code': 400, - 'msg': f'未找到可绑定的角色：{", ".join(missing or role_codes)}。请确认顶栏俱乐部与角色所属俱乐部一致', + 'msg': f'未找到可绑定的角色：{", ".join(missing or role_codes)}', }) if missing: return Response({ @@ -669,12 +646,11 @@ class ModifyAdminUserView(APIView): role_code = request.data.get('role_code') if not role_code: return Response({'code': 400, 'msg': '缺少角色编码'}) - from jituan.services.club_rbac import resolve_role_by_name - club_id = getattr(target_user, 'ClubID', '') or '' - role = resolve_role_by_name(role_code, club_id) + from backend.models import UserRole as LegacyUserRole + role = _resolve_backend_role(role_code) if not role: - return Response({'code': 404, 'msg': '角色不存在或与用户俱乐部不匹配'}) - UserRole.objects.filter(UserUUID=target_user.UserUUID, RoleUUID=role.RoleUUID).delete() + return Response({'code': 404, 'msg': '角色不存在'}) + LegacyUserRole.objects.filter(account_id=target_user.Phone, role=role).delete() return Response({'code': 0, 'msg': '角色移除成功'}) else: diff --git a/jituan/services/admin_context.py b/jituan/services/admin_context.py index 7da229d..359fa37 100644 --- a/jituan/services/admin_context.py +++ b/jituan/services/admin_context.py @@ -101,6 +101,22 @@ def is_system_super_admin(user): ) +def is_kefu_backend_account(user): + """可登录客服后台：超管 / UserType=kefu / KefuProfile / 旧表 user_role 有绑定。""" + if is_system_super_admin(user): + return True + if getattr(user, 'UserType', '') == 'kefu': + return True + try: + kefu = user.KefuProfile + if kefu.zhuangtai == 1: + return True + except Exception: + pass + from backend.utils import has_legacy_backend_binding + return has_legacy_backend_binding(user) + + def can_manage_admin_assignments(user, permissions=None): """是否可维护数据范围任职、俱乐部密钥等集团级配置（不含俱乐部 000001）。""" if is_system_super_admin(user): diff --git a/jituan/services/club_rbac.py b/jituan/services/club_rbac.py index c73a3f6..d7ddf9d 100644 --- a/jituan/services/club_rbac.py +++ b/jituan/services/club_rbac.py @@ -79,16 +79,11 @@ def resolve_role_by_name(role_name, club_id=None): def user_bound_role_names(user): from backend.utils import _legacy_backend_role_permissions - uuids = list( - UserRole.objects.filter(UserUUID=user.UserUUID).values_list('RoleUUID', flat=True) + _, legacy_names = _legacy_backend_role_permissions( + getattr(user, 'Phone', ''), + getattr(user, 'UserUID', None), ) - names = [] - if uuids: - names = list( - Role.objects.filter(RoleUUID__in=uuids, RoleStatus=1).values_list('RoleName', flat=True) - ) - _, legacy_names = _legacy_backend_role_permissions(getattr(user, 'Phone', '')) - return sorted(set(names + legacy_names)) + return sorted(set(legacy_names)) def resolve_kefu_admin_user(yonghuid=None, phone=None): diff --git a/jituan/views.py b/jituan/views.py index fb48db2..7f772f4 100644 --- a/jituan/views.py +++ b/jituan/views.py @@ -12,7 +12,7 @@ from rest_framework_simplejwt.tokens import RefreshToken from jituan.constants import SUPER_ADMIN_PHONES, ADMIN_ROLE_LABELS, CLUB_ID_DEFAULT, DATA_SCOPE_ALL, DATA_SCOPE_SINGLE from jituan.models import Club, AdminAssignment -from jituan.services.admin_context import build_admin_club_context, can_manage_admin_assignments +from jituan.services.admin_context import build_admin_club_context, is_kefu_backend_account, can_manage_admin_assignments from jituan.services.kefu_menu import build_menu_access_payload from jituan.services.wechat_login import login_or_register_by_club from jituan.services.caiwu_stats import build_caiwu_payload @@ -204,7 +204,7 @@ class ClubMeContextView(APIView): def get(self, request): user = request.user - if user.UserType not in ('kefu', 'admin') and not user.IsSuperuser: + if not is_kefu_backend_account(user): return Response({'code': 403, 'msg': '非后台账号', 'data': None}, status=403) return Response({ 'code': 0, @@ -219,7 +219,7 @@ class ClubKefuMenuAccessView(APIView): def get(self, request): user = request.user - if user.UserType not in ('kefu', 'admin') and not user.IsSuperuser: + if not is_kefu_backend_account(user): return Response({'code': 403, 'msg': '非后台账号', 'data': None}, status=403) payload, err = build_menu_access_payload(request) if err is not None: diff --git a/users/views.py b/users/views.py index eaa9828..64c6942 100644 --- a/users/views.py +++ b/users/views.py @@ -6926,13 +6926,14 @@ class KefuStatsView(APIView): logger.warning(f"统计接口 phone 不匹配: 请求phone={phone}, 用户phone={current_user.Phone}") return Response({'detail': '认证失败'}, status=status.HTTP_401_UNAUTHORIZED) - # 4. 验证用户类型是否为客服 - if current_user.UserType not in ('kefu', 'admin'): + # 4. 验证是否为后台客服账号 + from jituan.services.admin_context import is_kefu_backend_account, is_system_super_admin + if not is_kefu_backend_account(current_user): logger.warning(f"统计接口用户类型非客服: {current_user.UserUID}") return Response({'detail': '认证失败'}, status=status.HTTP_401_UNAUTHORIZED) # 5. 尝试获取客服扩展表数据（管理员跳过） - is_admin = current_user.UserType == 'admin' or current_user.IsSuperuser + is_admin = is_system_super_admin(current_user) if is_admin: return Response({ 'code': 0,