127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
"""权限校验与成员解析。"""
|
|
from rest_framework.exceptions import PermissionDenied, NotFound
|
|
|
|
from merchant_ops.constants import (
|
|
MEMBER_STATUS_ACTIVE, MEMBER_STATUS_REMOVED, OWNER_ALL_PERMISSIONS,
|
|
)
|
|
from merchant_ops.models import MerchantStaffMember, MerchantStaffMemberPermission
|
|
from merchant_ops.services.bootstrap import ensure_merchant_roles
|
|
from users.models import UserShangjia
|
|
|
|
|
|
class StaffAuthError(Exception):
|
|
def __init__(self, code, msg):
|
|
self.code = code
|
|
self.msg = msg
|
|
super().__init__(msg)
|
|
|
|
|
|
def is_merchant_owner(user):
|
|
try:
|
|
profile = user.ShopProfile
|
|
return profile.zhuangtai == 1
|
|
except UserShangjia.DoesNotExist:
|
|
return False
|
|
|
|
|
|
def get_active_staff_member(user):
|
|
return MerchantStaffMember.query.filter(
|
|
staff_user_id=user.UserUID,
|
|
status=MEMBER_STATUS_ACTIVE,
|
|
).select_related('role').first()
|
|
|
|
|
|
def get_member_permissions(member):
|
|
"""角色权限 + 成员覆盖。"""
|
|
perms = set(
|
|
member.role.role_permissions.values_list('perm_code', flat=True)
|
|
)
|
|
for mp in member.member_permissions.all():
|
|
if mp.granted:
|
|
perms.add(mp.perm_code)
|
|
else:
|
|
perms.discard(mp.perm_code)
|
|
return perms
|
|
|
|
|
|
def member_has_perm(member, perm_code):
|
|
return perm_code in get_member_permissions(member)
|
|
|
|
|
|
def assert_staff_perm(member, perm_code):
|
|
if not member_has_perm(member, perm_code):
|
|
raise StaffAuthError(403, f'无权限: {perm_code}')
|
|
|
|
|
|
def resolve_staff_context(user):
|
|
"""子客服上下文；非客服返回 None。"""
|
|
member = get_active_staff_member(user)
|
|
if not member:
|
|
return None
|
|
wallet = member.wallets.first()
|
|
perms = list(get_member_permissions(member))
|
|
return {
|
|
'active': True,
|
|
'merchant_id': member.merchant_id,
|
|
'member_id': member.id,
|
|
'role_id': member.role_id,
|
|
'role_code': member.role.role_code,
|
|
'role_name': member.role.role_name,
|
|
'display_name': member.display_name or '',
|
|
'permissions': perms,
|
|
'wallet': {
|
|
'quota_total': str(wallet.quota_total if wallet else 0),
|
|
'quota_used': str(wallet.quota_used if wallet else 0),
|
|
'quota_frozen': str(wallet.quota_frozen if wallet else 0),
|
|
'quota_available': str(wallet.quota_available if wallet else 0),
|
|
},
|
|
}
|
|
|
|
|
|
def resolve_owner_or_staff(request, perm_code=None):
|
|
"""
|
|
返回 (mode, merchant_id, member, operator_id)
|
|
mode: 'owner' | 'staff'
|
|
"""
|
|
user = request.user
|
|
if is_merchant_owner(user):
|
|
return 'owner', user.UserUID, None, user.UserUID
|
|
|
|
member = get_active_staff_member(user)
|
|
if not member:
|
|
raise StaffAuthError(403, '非商家老板或有效子客服')
|
|
|
|
if perm_code:
|
|
assert_staff_perm(member, perm_code)
|
|
return 'staff', member.merchant_id, member, user.UserUID
|
|
|
|
|
|
def assert_not_merchant_when_bind(user):
|
|
if is_merchant_owner(user):
|
|
raise StaffAuthError(403, '您已是商家，无法成为子客服')
|
|
|
|
|
|
def assert_no_active_staff_when_merchant_register(user):
|
|
if MerchantStaffMember.query.filter(
|
|
staff_user_id=user.UserUID, status=MEMBER_STATUS_ACTIVE
|
|
).exists():
|
|
raise StaffAuthError(403, '您当前是商家客服，请先联系商家解除后再认证商家')
|
|
|
|
|
|
def assert_can_bind_staff(user):
|
|
assert_not_merchant_when_bind(user)
|
|
if MerchantStaffMember.query.filter(
|
|
staff_user_id=user.UserUID, status=MEMBER_STATUS_ACTIVE
|
|
).exists():
|
|
raise StaffAuthError(403, '您已绑定商家客服，无法重复绑定')
|
|
|
|
|
|
def get_owner_permissions():
|
|
return list(OWNER_ALL_PERMISSIONS)
|
|
|
|
|
|
def ensure_owner_merchant(merchant_id, user):
|
|
if user.UserUID != merchant_id or not is_merchant_owner(user):
|
|
raise StaffAuthError(403, '仅商家老板可操作')
|
|
ensure_merchant_roles(merchant_id)
|