e6e8e718d3
COS 上传接口 fakuan_evidence / fakuan_evidence_draft 现允许具 penalty_apply 或 penalty_manage 且能查看该订单的子客服。 Co-authored-by: Cursor <cursoragent@cursor.com>
108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
"""视图辅助：以商家身份代理调用现有订单 View（不修改原接口）。"""
|
||
import logging
|
||
|
||
from backend.utils import pick_order_id
|
||
from merchant_ops.services.authz import StaffAuthError, get_active_staff_member, member_has_perm
|
||
from merchant_ops.models import MerchantOrderDispatch
|
||
from orders.models import MerchantOrderExt, Penalty
|
||
from users.business_models import User
|
||
from users.models import UserShangjia
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class PenaltyEvidenceAuthError(Exception):
|
||
def __init__(self, code, msg):
|
||
self.code = code
|
||
self.msg = msg
|
||
super().__init__(msg)
|
||
|
||
|
||
def resolve_penalty_evidence_upload_auth(request, dingdan_id, fadan_id=None):
|
||
"""
|
||
商家老板或具 penalty_apply / penalty_manage 且能查看该订单的子客服。
|
||
返回所属 merchant_id（ApplicantID）。
|
||
"""
|
||
user = request.user
|
||
try:
|
||
shangjia_profile = user.ShopProfile
|
||
if shangjia_profile.zhuangtai != 1:
|
||
raise PenaltyEvidenceAuthError(403, '商家状态异常')
|
||
try:
|
||
MerchantOrderExt.query.get(Order__OrderID=dingdan_id, MerchantID=user.UserUID)
|
||
except MerchantOrderExt.DoesNotExist:
|
||
raise PenaltyEvidenceAuthError(403, '订单不存在或无权操作')
|
||
if fadan_id and not Penalty.query.filter(
|
||
id=fadan_id,
|
||
RelatedOrderID=dingdan_id,
|
||
ApplicantID=user.UserUID,
|
||
).exists():
|
||
raise PenaltyEvidenceAuthError(403, '罚单不存在或无权操作')
|
||
return user.UserUID
|
||
except UserShangjia.DoesNotExist:
|
||
pass
|
||
|
||
member = get_active_staff_member(user)
|
||
if not member:
|
||
raise PenaltyEvidenceAuthError(403, '非商家或子客服无权上传')
|
||
if not (member_has_perm(member, 'penalty_apply') or member_has_perm(member, 'penalty_manage')):
|
||
raise PenaltyEvidenceAuthError(403, '无罚单权限')
|
||
ok, _ = staff_can_access_order(member, dingdan_id)
|
||
if not ok:
|
||
raise PenaltyEvidenceAuthError(403, '无权操作该订单')
|
||
if fadan_id and not Penalty.query.filter(
|
||
id=fadan_id,
|
||
RelatedOrderID=dingdan_id,
|
||
ApplicantID=member.merchant_id,
|
||
).exists():
|
||
raise PenaltyEvidenceAuthError(403, '罚单不存在或无权操作')
|
||
return member.merchant_id
|
||
|
||
|
||
def get_merchant_user(merchant_id):
|
||
return User.query.get(UserUID=merchant_id)
|
||
|
||
|
||
def staff_can_access_order(member, order_id, perm_all='order_view_all', perm_self='order_view_self'):
|
||
ext = MerchantOrderExt.query.filter(
|
||
MerchantID=member.merchant_id,
|
||
Order__OrderID=order_id,
|
||
).first()
|
||
if not ext:
|
||
return False, None
|
||
if member_has_perm(member, perm_all):
|
||
return True, ext
|
||
if member_has_perm(member, perm_self):
|
||
disp = MerchantOrderDispatch.query.filter(order_id=order_id).first()
|
||
if disp and disp.staff_member_id == member.id:
|
||
return True, ext
|
||
return False, ext
|
||
|
||
|
||
def proxy_merchant_order_view(request, member, view_class, perm_code):
|
||
"""子客服通过商家 User 代理调用现有 View.post。"""
|
||
return proxy_as_merchant_user(request, member, view_class, perm_code)
|
||
|
||
|
||
def proxy_as_merchant_user(request, member, view_class, perm_code):
|
||
"""子客服以商家身份代理调用任意商家 View（订单/配置等）。"""
|
||
if not member_has_perm(member, perm_code):
|
||
raise StaffAuthError(403, f'无权限: {perm_code}')
|
||
|
||
order_id = pick_order_id(request.data)
|
||
if order_id:
|
||
ok, _ = staff_can_access_order(member, order_id)
|
||
if not ok:
|
||
raise StaffAuthError(403, '无权操作该订单')
|
||
|
||
merchant_user = get_merchant_user(member.merchant_id)
|
||
old_user = request.user
|
||
request.user = merchant_user
|
||
request._staff_member = member
|
||
try:
|
||
return view_class().post(request)
|
||
finally:
|
||
request.user = old_user
|
||
if hasattr(request, '_staff_member'):
|
||
delattr(request, '_staff_member')
|