XingQue/Django
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: kefu permissions use legacy permission table instead of gvsdsdk

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-06-28 23:59:32 +08:00
parent c336a893a0
commit 42cb117a8a
6 changed files with 183 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
backend/utils.py
View File

@@ -77,14 +77,19 @@ class _AllPermissions(list):
return True return True
def _legacy_backend_role_permissions(phone): def _legacy_backend_role_permissions(phone, yonghuid=None):
""" """
读取旧权限表（backend.models）： 读取旧权限表（backend.models）：
user_role.account_id = 手机号 → role → role_permission → permission.perm_code user_role.account_id = 手机号或 UserUID → role → role_permission → permission.perm_code
与 gvsdsdk 表并行，未改 SDK 结构。
""" """
phone = (phone or '').strip() account_ids = []
if not phone: 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 [], [] return [], []
try: try:
from backend.models import ( from backend.models import (
@@ -93,7 +98,7 @@ def _legacy_backend_role_permissions(phone):
Permission as BackendPermission, Permission as BackendPermission,
) )
role_ids = list( 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: if not role_ids:
return [], [] return [], []
@@ -107,10 +112,22 @@ def _legacy_backend_role_permissions(phone):
) )
return perm_codes, role_names return perm_codes, role_names
except Exception as e: 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 [], [] 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): def verify_kefu_permission(request, username_from_frontend=None):
""" """
验证客服/管理员身份，并返回该用户的所有权限列表（基于新RBAC表） 验证客服/管理员身份，并返回该用户的所有权限列表（基于新RBAC表）
@@ -128,9 +145,10 @@ def verify_kefu_permission(request, username_from_frontend=None):
# ---------- 1. 必须是客服或管理员 ---------- # ---------- 1. 必须是客服或管理员 ----------
from jituan.constants import SUPER_ADMIN_PHONES 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_super_phone = getattr(current_user, 'Phone', '') in SUPER_ADMIN_PHONES
is_admin = current_user.UserType == 'admin' or current_user.IsSuperuser or is_super_phone 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) return None, Response({'code': 403, 'msg': '身份错误，非客服账号'}, status=403)
# ---------- 2. 客服扩展表存在且状态正常（管理员跳过） ---------- # ---------- 2. 客服扩展表存在且状态正常（管理员跳过） ----------
@@ -156,42 +174,12 @@ def verify_kefu_permission(request, username_from_frontend=None):
'msg': '身份错误，无权操作其他账号，相关操作已记录' 'msg': '身份错误，无权操作其他账号，相关操作已记录'
}, status=403) }, status=403)
# ---------- 4. 获取用户的所有权限 ---------- # ---------- 4. 权限码：只读旧表 permission（user_role → role → role_permission）----------
# 使用 gvsdsdk RBAC 表（UserRole → RolePermission → Permission） legacy_codes, _ = _legacy_backend_role_permissions(
try: current_user.Phone,
user_uuid = current_user.UserUUID getattr(current_user, 'UserUID', None),
role_uuids = list(SdkUserRole.objects.filter( )
UserUUID=user_uuid permissions = list(legacy_codes) if legacy_codes else []
).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))
# 管理员/超级用户自动拥有所有权限 # 管理员/超级用户自动拥有所有权限
if is_admin and '000001' not in permissions: if is_admin and '000001' not in permissions:

272
backend/view.py
View File

@@ -148,46 +148,62 @@ class GetRolePermissionView(APIView):
if '000001' not in permissions: if '000001' not in permissions:
return Response({'code': 403, 'msg': '您无权访问此页面'}) return Response({'code': 403, 'msg': '您无权访问此页面'})
# 3. 获取所有权限 from backend.models import (
all_perms = Permission.objects.filter(PermStatus=1).values('PermCode', 'PermName', 'PermDesc') Role as BackendRole,
permissions_list = [_serialize_permission_row(p) for p in all_perms] Permission as BackendPermission,
UserRole as LegacyUserRole,
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)
) )
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 = [] roles_data = []
for role in roles: for role in BackendRole.objects.all().order_by('role_id'):
user_count = count_role_users_in_scope(role, request, kefu_user_uuids) user_count = LegacyUserRole.objects.filter(role=role).count()
role_perm_uuids = RolePermission.objects.filter( perms_of_role = BackendPermission.objects.filter(rolepermission__role=role).distinct()
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')
]
roles_data.append({ roles_data.append({
'role_code': role.RoleName, 'role_code': role.role_name,
'role_name': role.RoleName, 'role_name': role.role_name,
'description': role.RoleDesc or '', 'description': role.description or '',
'club_id': getattr(role, 'ClubID', '') or '', 'club_id': '',
'user_count': user_count, '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({ return Response({
'code': 0, 'code': 0,
'data': { 'data': {
'roles': roles_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): class ModifyRolePermissionView(APIView):
""" """
通用接口：支持删除角色、为角色添加权限、移除角色权限 通用接口：支持删除角色、为角色添加权限、移除角色权限
@@ -213,22 +229,26 @@ class ModifyRolePermissionView(APIView):
if not role_code: if not role_code:
return Response({'code': 400, 'msg': '缺少角色编码'}) return Response({'code': 400, 'msg': '缺少角色编码'})
try: from backend.models import (
role = Role.objects.get(RoleName=role_code) Permission as BackendPermission,
except Role.DoesNotExist: RolePermission as LegacyRolePermission,
UserRole as LegacyUserRole,
)
role = _resolve_backend_role(role_code)
if not role:
return Response({'code': 404, 'msg': '角色不存在'}) return Response({'code': 404, 'msg': '角色不存在'})
# 超级管理员角色的特殊限制 # 超级管理员角色的特殊限制
if role.RoleName == '管理员': if role.role_name == '管理员':
if action in ['delete_role']: if action in ['delete_role']:
return Response({'code': 403, 'msg': '超级管理员角色不能删除'}) return Response({'code': 403, 'msg': '超级管理员角色不能删除'})
# 根据 action 处理 # 根据 action 处理
if action == 'delete_role': if action == 'delete_role':
# 删除角色：同时删除角色权限关联、用户角色关联
with transaction.atomic(): with transaction.atomic():
RolePermission.objects.filter(RoleUUID=role.RoleUUID).delete() LegacyRolePermission.objects.filter(role=role).delete()
UserRole.objects.filter(RoleUUID=role.RoleUUID).delete() LegacyUserRole.objects.filter(role=role).delete()
role.delete() role.delete()
return Response({'code': 0, 'msg': '角色删除成功'}) return Response({'code': 0, 'msg': '角色删除成功'})
@@ -236,34 +256,26 @@ class ModifyRolePermissionView(APIView):
perm_code = request.data.get('perm_code') perm_code = request.data.get('perm_code')
if not perm_code: if not perm_code:
return Response({'code': 400, 'msg': '缺少权限编码'}) 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': '此权限已存在，不能重复添加'}) return Response({'code': 403, 'msg': '此权限已存在，不能重复添加'})
try: perm = BackendPermission.objects.filter(perm_code=perm_code).first()
perm = Permission.objects.get(PermCode=perm_code) if not perm:
except Permission.DoesNotExist:
return Response({'code': 404, 'msg': '权限不存在'}) return Response({'code': 404, 'msg': '权限不存在'})
# 检查是否已存在 if LegacyRolePermission.objects.filter(role=role, permission=perm).exists():
if RolePermission.objects.filter(RoleUUID=role.RoleUUID, PermUUID=perm.PermUUID).exists():
return Response({'code': 400, 'msg': '该角色已拥有此权限'}) return Response({'code': 400, 'msg': '该角色已拥有此权限'})
RolePermission.objects.create( LegacyRolePermission.objects.create(role=role, permission=perm)
RolePermUUID=uuid.uuid4().bytes,
RoleUUID=role.RoleUUID,
PermUUID=perm.PermUUID,
CreateTime=timezone.now(),
)
return Response({'code': 0, 'msg': '权限添加成功'}) return Response({'code': 0, 'msg': '权限添加成功'})
elif action == 'remove_permission': elif action == 'remove_permission':
perm_code = request.data.get('perm_code') perm_code = request.data.get('perm_code')
if not perm_code: if not perm_code:
return Response({'code': 400, 'msg': '缺少权限编码'}) 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': '不能移除超级管理员的此权限'}) return Response({'code': 403, 'msg': '不能移除超级管理员的此权限'})
try: perm = BackendPermission.objects.filter(perm_code=perm_code).first()
perm = Permission.objects.get(PermCode=perm_code) if not perm:
except Permission.DoesNotExist:
return Response({'code': 404, 'msg': '权限不存在'}) 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: if deleted == 0:
return Response({'code': 400, 'msg': '该角色未拥有此权限'}) return Response({'code': 400, 'msg': '该角色未拥有此权限'})
return Response({'code': 0, 'msg': '权限移除成功'}) return Response({'code': 0, 'msg': '权限移除成功'})
@@ -272,45 +284,32 @@ class ModifyRolePermissionView(APIView):
role_name = (request.data.get('role_name') or '').strip() role_name = (request.data.get('role_name') or '').strip()
description = (request.data.get('description') or '').strip() description = (request.data.get('description') or '').strip()
perm_codes = request.data.get('perm_codes') or [] perm_codes = request.data.get('perm_codes') or []
if role.RoleName == '管理员': from backend.models import Role as BackendRole
role_name = role.RoleName if role.role_name == '管理员':
elif role_name and role_name != role.RoleName: role_name = role.role_name
if Role.objects.filter(RoleName=role_name).exclude(RoleUUID=role.RoleUUID).exists(): 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': '角色名称已存在'}) return Response({'code': 400, 'msg': '角色名称已存在'})
with transaction.atomic(): with transaction.atomic():
if role.RoleName != '管理员': if role.role_name != '管理员':
if role_name: if role_name:
role.RoleName = role_name role.role_name = role_name
role.RoleDesc = description role.description = description
role.save(update_fields=['RoleName', 'RoleDesc']) role.save(update_fields=['role_name', 'description'])
RolePermission.objects.filter(RoleUUID=role.RoleUUID).delete() LegacyRolePermission.objects.filter(role=role).delete()
for pc in perm_codes: for pc in perm_codes:
if pc == '000001' and role.RoleName != '管理员': if pc == '000001' and role.role_name != '管理员':
continue continue
try: perm = BackendPermission.objects.filter(perm_code=pc).first()
perm = Permission.objects.get(PermCode=pc, PermStatus=1) if not perm:
except Permission.DoesNotExist:
continue continue
RolePermission.objects.create( LegacyRolePermission.objects.create(role=role, permission=perm)
RolePermUUID=uuid.uuid4().bytes, if role.role_name == '管理员':
RoleUUID=role.RoleUUID, super_perm = BackendPermission.objects.filter(perm_code='000001').first()
PermUUID=perm.PermUUID, if super_perm and not LegacyRolePermission.objects.filter(
CreateTime=timezone.now(), role=role, permission=super_perm,
) ).exists():
if role.RoleName == '管理员': LegacyRolePermission.objects.create(role=role, permission=super_perm)
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
return Response({'code': 0, 'msg': '角色保存成功'}) return Response({'code': 0, 'msg': '角色保存成功'})
else: else:
@@ -339,47 +338,33 @@ class AddRoleView(APIView):
if '000001' not in permissions: if '000001' not in permissions:
return Response({'code': 403, 'msg': '您无权进行此操作'}) return Response({'code': 403, 'msg': '您无权进行此操作'})
# 检查角色名称是否已存在（同俱乐部内唯一） from backend.models import (
from jituan.services.club_rbac import resolve_role_club_id, role_name_exists Role as BackendRole,
target_club_id = resolve_role_club_id( Permission as BackendPermission,
request, (request.data.get('target_club_id') or '').strip() or None, 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))}" role_code = f"ROLE_{''.join(random.choices(string.ascii_uppercase + string.digits, k=8))}"
# 创建角色
with transaction.atomic(): with transaction.atomic():
role = Role.objects.create( role = BackendRole.objects.create(
RoleUUID=uuid.uuid4().bytes, role_code=role_code,
TenantUUID=b'\x00' * 16, role_name=role_name,
RoleName=role_name, description=description,
RoleType=2,
RoleStatus=1,
AssignScope=0,
RoleDesc=description,
ClubID=target_club_id,
CreateTime=timezone.now(),
) )
# 绑定权限（过滤掉不存在的权限编码）
for pc in perm_codes: for pc in perm_codes:
# 不允许添加 000001 权限（此权限仅供超级管理员）
if pc == '000001': if pc == '000001':
continue continue
try: perm = BackendPermission.objects.filter(perm_code=pc).first()
perm = Permission.objects.get(PermCode=pc) if not perm:
RolePermission.objects.create(
RolePermUUID=uuid.uuid4().bytes,
RoleUUID=role.RoleUUID,
PermUUID=perm.PermUUID,
CreateTime=timezone.now(),
)
except Permission.DoesNotExist:
logger.warning(f"添加角色时忽略不存在的权限编码: {pc}") 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: if nicheng:
users = users.filter(KefuProfile__nicheng__icontains=nicheng) users = users.filter(KefuProfile__nicheng__icontains=nicheng)
# 关键修正：先求值为列表，再传入 phone__in
if role_code: 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: if target_role:
user_uuids = list( account_ids = list(
UserRole.objects.filter(RoleUUID=target_role.RoleUUID) LegacyUserRole.objects.filter(role=target_role).values_list('account_id', flat=True)
.values_list('UserUUID', flat=True)
) )
users = users.filter(UserUUID__in=user_uuids) users = users.filter(Phone__in=account_ids)
else: else:
users = users.none() users = users.none()
@@ -435,16 +419,12 @@ class GetAdminUserListView(APIView):
page_obj = paginator.get_page(page) page_obj = paginator.get_page(page)
data_list = [] data_list = []
from backend.models import UserRole as LegacyUserRole
for user in page_obj: for user in page_obj:
# 通过 gvsdsdk UserRole 获取用户角色 user_roles = LegacyUserRole.objects.filter(account_id=user.Phone).select_related('role')
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)
roles_data = [ roles_data = [
{'role_code': r.RoleName, 'role_name': r.RoleName} {'role_code': ur.role.role_name, 'role_name': ur.role.role_name}
for r in user_roles for ur in user_roles
] ]
data_list.append({ data_list.append({
'phone': user.Phone or '', 'phone': user.Phone or '',
@@ -490,16 +470,16 @@ class GetAdminRolesView(APIView):
if '000001' not in permissions: if '000001' not in permissions:
return Response({'code': 403, 'msg': '您无权访问此页面'}) return Response({'code': 403, 'msg': '您无权访问此页面'})
from jituan.services.club_rbac import filter_roles_for_request from backend.models import Role as BackendRole
roles = filter_roles_for_request(request).values('RoleName', 'ClubID') roles = BackendRole.objects.all().order_by('role_id')
return Response({ return Response({
'code': 0, 'code': 0,
'data': { 'data': {
'roles': [ 'roles': [
{ {
'role_code': r['RoleName'], 'role_code': r.role_name,
'role_name': r['RoleName'], 'role_name': r.role_name,
'club_id': r.get('ClubID') or '', 'club_id': '',
} }
for r in roles for r in roles
] ]
@@ -638,25 +618,22 @@ class ModifyAdminUserView(APIView):
role_codes = request.data.get('role_codes', []) role_codes = request.data.get('role_codes', [])
if not role_codes: if not role_codes:
return Response({'code': 400, 'msg': '请选择角色'}) return Response({'code': 400, 'msg': '请选择角色'})
from jituan.services.club_rbac import resolve_role_by_name from backend.models import Role as BackendRole, UserRole as LegacyUserRole
club_id = getattr(target_user, 'ClubID', '') or ''
added, missing = [], [] added, missing = [], []
for rc in role_codes: for rc in role_codes:
role = resolve_role_by_name(rc, club_id) role = _resolve_backend_role(rc)
if not role: if not role:
missing.append(rc) missing.append(rc)
continue continue
UserRole.objects.get_or_create( LegacyUserRole.objects.get_or_create(
UserRoleUUID=uuid.uuid4().bytes, account_id=target_user.Phone,
UserUUID=target_user.UserUUID, role=role,
RoleUUID=role.RoleUUID,
CompanyUUID=uuid.UUID('b0000000-0000-0000-0000-000000000001').bytes,
) )
added.append(role.RoleName) added.append(role.role_name)
if not added: if not added:
return Response({ return Response({
'code': 400, 'code': 400,
'msg': f'未找到可绑定的角色：{", ".join(missing or role_codes)}。请确认顶栏俱乐部与角色所属俱乐部一致', 'msg': f'未找到可绑定的角色：{", ".join(missing or role_codes)}',
}) })
if missing: if missing:
return Response({ return Response({
@@ -669,12 +646,11 @@ class ModifyAdminUserView(APIView):
role_code = request.data.get('role_code') role_code = request.data.get('role_code')
if not role_code: if not role_code:
return Response({'code': 400, 'msg': '缺少角色编码'}) return Response({'code': 400, 'msg': '缺少角色编码'})
from jituan.services.club_rbac import resolve_role_by_name from backend.models import UserRole as LegacyUserRole
club_id = getattr(target_user, 'ClubID', '') or '' role = _resolve_backend_role(role_code)
role = resolve_role_by_name(role_code, club_id)
if not role: if not role:
return Response({'code': 404, 'msg': '角色不存在或与用户俱乐部不匹配'}) return Response({'code': 404, 'msg': '角色不存在'})
UserRole.objects.filter(UserUUID=target_user.UserUUID, RoleUUID=role.RoleUUID).delete() LegacyUserRole.objects.filter(account_id=target_user.Phone, role=role).delete()
return Response({'code': 0, 'msg': '角色移除成功'}) return Response({'code': 0, 'msg': '角色移除成功'})
else: else:

16
jituan/services/admin_context.py
View File

@@ -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): def can_manage_admin_assignments(user, permissions=None):
"""是否可维护数据范围任职、俱乐部密钥等集团级配置（不含俱乐部 000001）。""" """是否可维护数据范围任职、俱乐部密钥等集团级配置（不含俱乐部 000001）。"""
if is_system_super_admin(user): if is_system_super_admin(user):

13
jituan/services/club_rbac.py
View File

@@ -79,16 +79,11 @@ def resolve_role_by_name(role_name, club_id=None):
def user_bound_role_names(user): def user_bound_role_names(user):
from backend.utils import _legacy_backend_role_permissions from backend.utils import _legacy_backend_role_permissions
uuids = list( _, legacy_names = _legacy_backend_role_permissions(
UserRole.objects.filter(UserUUID=user.UserUUID).values_list('RoleUUID', flat=True) getattr(user, 'Phone', ''),
getattr(user, 'UserUID', None),
) )
names = [] return sorted(set(legacy_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))
def resolve_kefu_admin_user(yonghuid=None, phone=None): def resolve_kefu_admin_user(yonghuid=None, phone=None):

6
jituan/views.py
View File

@@ -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.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.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.kefu_menu import build_menu_access_payload
from jituan.services.wechat_login import login_or_register_by_club from jituan.services.wechat_login import login_or_register_by_club
from jituan.services.caiwu_stats import build_caiwu_payload from jituan.services.caiwu_stats import build_caiwu_payload
@@ -204,7 +204,7 @@ class ClubMeContextView(APIView):
def get(self, request): def get(self, request):
user = request.user 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': 403, 'msg': '非后台账号', 'data': None}, status=403)
return Response({ return Response({
'code': 0, 'code': 0,
@@ -219,7 +219,7 @@ class ClubKefuMenuAccessView(APIView):
def get(self, request): def get(self, request):
user = request.user 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': 403, 'msg': '非后台账号', 'data': None}, status=403)
payload, err = build_menu_access_payload(request) payload, err = build_menu_access_payload(request)
if err is not None: if err is not None:

7
users/views.py
View File

@@ -6926,13 +6926,14 @@ class KefuStatsView(APIView):
logger.warning(f"统计接口 phone 不匹配: 请求phone={phone}, 用户phone={current_user.Phone}") logger.warning(f"统计接口 phone 不匹配: 请求phone={phone}, 用户phone={current_user.Phone}")
return Response({'detail': '认证失败'}, status=status.HTTP_401_UNAUTHORIZED) return Response({'detail': '认证失败'}, status=status.HTTP_401_UNAUTHORIZED)
# 4. 验证用户类型是否为客服 # 4. 验证是否为后台客服账号
if current_user.UserType not in ('kefu', 'admin'): 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}") logger.warning(f"统计接口用户类型非客服: {current_user.UserUID}")
return Response({'detail': '认证失败'}, status=status.HTTP_401_UNAUTHORIZED) return Response({'detail': '认证失败'}, status=status.HTTP_401_UNAUTHORIZED)
# 5. 尝试获取客服扩展表数据（管理员跳过） # 5. 尝试获取客服扩展表数据（管理员跳过）
is_admin = current_user.UserType == 'admin' or current_user.IsSuperuser is_admin = is_system_super_admin(current_user)
if is_admin: if is_admin:
return Response({ return Response({
'code': 0, 'code': 0,