165 lines
5.6 KiB
Python
165 lines
5.6 KiB
Python
"""权限校验与成员解析。"""
|
|
from rest_framework.exceptions import PermissionDenied, NotFound
|
|
|
|
from merchant_ops.constants import (
|
|
MEMBER_STATUS_ACTIVE, MEMBER_STATUS_DISABLED, 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 require_active_staff_member(user):
|
|
"""子客服接口鉴权：区分未绑定、禁用、移除等场景（已是商家老板也可担任其他店子客服）。"""
|
|
member = MerchantStaffMember.query.filter(
|
|
staff_user_id=user.UserUID,
|
|
).select_related('role').order_by('-id').first()
|
|
if not member:
|
|
raise StaffAuthError(403, '未绑定子客服，请使用商家邀请码绑定')
|
|
if member.status == MEMBER_STATUS_DISABLED:
|
|
raise StaffAuthError(403, '子客服账号已被禁用，请联系商家老板')
|
|
if member.status == MEMBER_STATUS_REMOVED:
|
|
raise StaffAuthError(403, '子客服已被移除，请联系商家老板重新邀请')
|
|
if member.status != MEMBER_STATUS_ACTIVE:
|
|
raise StaffAuthError(403, '非有效子客服')
|
|
return member
|
|
|
|
|
|
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 staff_order_read_perm(member):
|
|
"""子客服读订单/商品类型/模板时，任一看单或派单权限即可。"""
|
|
for code in ('order_dispatch', 'order_view_all', 'order_view_self', 'order_detail'):
|
|
if member_has_perm(member, code):
|
|
return code
|
|
return None
|
|
|
|
|
|
def assert_staff_perm(member, perm_code):
|
|
if not member_has_perm(member, perm_code):
|
|
raise StaffAuthError(403, f'无权限: {perm_code}')
|
|
|
|
|
|
def staff_fields_for_login(user):
|
|
"""登录/注册响应附带子客服身份（与 shangjiastatus 并列，互斥）。"""
|
|
ctx = resolve_staff_context(user)
|
|
return {
|
|
'staffstatus': 1 if ctx else 0,
|
|
'staff_context': ctx,
|
|
}
|
|
|
|
|
|
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_self_merchant_when_bind(user, invite_merchant_id):
|
|
"""不能绑定为自己店铺的客服；可以是其他商家的子客服。"""
|
|
if invite_merchant_id is not None and str(user.UserUID) == str(invite_merchant_id):
|
|
raise StaffAuthError(403, '不能绑定为自己商家的客服')
|
|
|
|
|
|
def assert_can_bind_staff(user, invite_merchant_id=None):
|
|
assert_not_self_merchant_when_bind(user, invite_merchant_id)
|
|
existing = MerchantStaffMember.query.filter(
|
|
staff_user_id=user.UserUID, status=MEMBER_STATUS_ACTIVE
|
|
).first()
|
|
if existing:
|
|
if invite_merchant_id and str(existing.merchant_id) == str(invite_merchant_id):
|
|
raise StaffAuthError(400, '您已是该商家客服')
|
|
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 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)
|