421 lines
16 KiB
Python
421 lines
16 KiB
Python
import threading
|
||
import requests
|
||
import time
|
||
import logging
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
from django.conf import settings
|
||
from django.db import transaction
|
||
from django.db.models import F
|
||
|
||
from config.models import Club, DailyDispatchStat, DailyIncomeStat, DailyPayoutStat
|
||
from orders.models import Dingdan, DingdanPingtai, DingdanShangjia, Lilubiao
|
||
from products.models import ShangpinLeixing, Gsfenhong
|
||
from users.models import UserDashou, UserGuanshi
|
||
|
||
from backend.utils import update_guanshi_daily_by_action
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
|
||
# 可选：线程池限制（避免大量线程同时创建，建议使用 Celery，这里简单用线程）
|
||
THREAD_POOL_SIZE = 10
|
||
_semaphore = threading.Semaphore(THREAD_POOL_SIZE)
|
||
|
||
def sync_order_to_partners(dingdan_id):
|
||
"""
|
||
异步向所有开启互通的合作平台同步订单。
|
||
使用信号量限制并发线程数。
|
||
"""
|
||
def _task():
|
||
_semaphore.acquire()
|
||
try:
|
||
_sync_order_task(dingdan_id)
|
||
finally:
|
||
_semaphore.release()
|
||
|
||
thread = threading.Thread(target=_task)
|
||
thread.daemon = True
|
||
thread.start()
|
||
|
||
def _sync_order_task(dingdan_id):
|
||
"""实际同步任务（在线程中执行）"""
|
||
try:
|
||
# 获取订单对象（只取必要字段，提升性能）
|
||
dingdan = Dingdan.objects.only(
|
||
'dingdan_id', 'zhuangtai', 'jine', 'dashou_fencheng', 'jieshao',
|
||
'beizhu', 'nicheng', 'leixing_id', 'yaoqiuleixing', 'huiyuan_id',
|
||
'yongjin', 'zhiding_id', 'create_time'
|
||
).get(dingdan_id=dingdan_id)
|
||
|
||
# 基本条件判断
|
||
if dingdan.zhuangtai != 1:
|
||
logger.info(f"订单 {dingdan_id} 状态 {dingdan.zhuangtai} 不是1，不同步")
|
||
return
|
||
if dingdan.zhiding_id:
|
||
logger.info(f"订单 {dingdan_id} 是指定单，不同步")
|
||
return
|
||
if dingdan.yaoqiuleixing != 1:
|
||
logger.info(f"订单 {dingdan_id} 抢单要求类型不是会员抢单，不同步")
|
||
return
|
||
if not dingdan.leixing_id:
|
||
logger.info(f"订单 {dingdan_id} 无商品类型，不同步")
|
||
return
|
||
try:
|
||
leixing = ShangpinLeixing.objects.get(id=dingdan.leixing_id)
|
||
if not leixing.is_cross_enabled:
|
||
logger.info(f"订单 {dingdan_id} 的商品类型未开启互通，不同步")
|
||
return
|
||
except ShangpinLeixing.DoesNotExist:
|
||
logger.info(f"订单 {dingdan_id} 的商品类型不存在，不同步")
|
||
return
|
||
|
||
# 查询所有开启互通的俱乐部
|
||
clubs = Club.objects.filter(is_interop_enabled=True).select_related()
|
||
if not clubs.exists():
|
||
return
|
||
|
||
# 为每个俱乐部发送订单
|
||
for club in clubs:
|
||
_send_to_single_partner(dingdan, club)
|
||
|
||
except Exception as e:
|
||
logger.error(f"同步订单任务异常: {e}", exc_info=True)
|
||
|
||
def _send_to_single_partner(dingdan, club):
|
||
"""向单个俱乐部发送订单（带重试，幂等处理）"""
|
||
# 计算实际金额：订单打手分成 × 对方费率
|
||
actual_amount = dingdan.dashou_fencheng * club.dashou_rate
|
||
|
||
# 构建请求数据
|
||
payload = {
|
||
'token': settings.CROSS_PLATFORM_TOKEN,
|
||
'our_club_id': club.club_id, # 我方在对方系统中的标识（存储于对方表的 partner_club_id）
|
||
'partner_club_id': club.partner_club_id, # 对方自己的俱乐部ID
|
||
'partner_order_id': dingdan.dingdan_id, # 我方订单ID作为对方的外部订单号
|
||
'jine': str(actual_amount), # 对方平台应展示的订单金额
|
||
'dashou_fencheng': str(actual_amount), # 对方平台打手分成（与金额一致）
|
||
'jieshao': dingdan.jieshao,
|
||
'beizhu': dingdan.beizhu,
|
||
'create_time': dingdan.create_time.isoformat(),
|
||
'nicheng': dingdan.nicheng,
|
||
'leixing_id': dingdan.leixing_id, # 我方商品类型ID，对方需要映射
|
||
'user_id': get_order_user_id(dingdan), # 对方平台下单用户ID（不带前缀）
|
||
}
|
||
|
||
# 重试机制（最多3次，指数退避）
|
||
for attempt in range(1, 4):
|
||
try:
|
||
response = requests.post(
|
||
club.partner_domain + '/dingdan/partner_sync_order',
|
||
json=payload,
|
||
timeout=5
|
||
)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
# 对方返回 code=0 表示成功（包括新建或已存在）
|
||
if data.get('code') == 0:
|
||
# 判断本次响应是新创建还是已存在
|
||
status = data.get('data', {}).get('status', '')
|
||
if status == 'existed':
|
||
# 订单已经存在于对方系统，不重复统计
|
||
logger.info(f"订单 {dingdan.dingdan_id} 在俱乐部 {club.club_id} 已存在，跳过统计")
|
||
else:
|
||
# 新创建成功，更新统计
|
||
try:
|
||
update_daily_dispatch_stat(
|
||
direction=1, # 我方派单
|
||
partner_club_id=club.partner_club_id,
|
||
dispatch_amount=actual_amount,
|
||
|
||
)
|
||
logger.info(f"成功同步订单 {dingdan.dingdan_id} 到俱乐部 {club.club_id}")
|
||
except Exception as e:
|
||
logger.error(f"更新派单统计失败: {e}", exc_info=True)
|
||
|
||
# 标记订单为跨平台（仅当第一次成功时标记，避免重复标记）
|
||
if not dingdan.is_cross:
|
||
Dingdan.objects.filter(dingdan_id=dingdan.dingdan_id).update(
|
||
is_cross=1,
|
||
dispatch_type=1 # 我方派单
|
||
)
|
||
logger.info(f"订单 {dingdan.dingdan_id} 已标记为跨平台订单")
|
||
break
|
||
else:
|
||
# 对方返回业务错误（如映射不存在等），不重试
|
||
logger.error(f"同步订单失败（业务）: {data}")
|
||
break
|
||
else:
|
||
logger.error(f"同步订单HTTP失败: {response.status_code}")
|
||
except Exception as e:
|
||
logger.error(f"同步订单异常（尝试 {attempt}/3）: {e}")
|
||
|
||
if attempt < 3:
|
||
time.sleep(2 ** attempt) # 指数退避
|
||
else:
|
||
logger.error(f"同步订单 {dingdan.dingdan_id} 到俱乐部 {club.club_id} 最终失败")
|
||
|
||
|
||
|
||
# 获取下单方ID（根据订单类型）
|
||
def get_order_user_id(order):
|
||
if order.fadan_pingtai == 1:
|
||
try:
|
||
ext = DingdanPingtai.objects.get(dingdan=order)
|
||
return ext.laoban_id # 老板ID
|
||
except DingdanPingtai.DoesNotExist:
|
||
return None
|
||
elif order.fadan_pingtai == 2:
|
||
try:
|
||
ext = DingdanShangjia.objects.get(dingdan=order)
|
||
return ext.shangjia_id # 商家ID
|
||
except DingdanShangjia.DoesNotExist:
|
||
return None
|
||
return None
|
||
|
||
|
||
|
||
|
||
|
||
def update_daily_dispatch_stat(direction, partner_club_id,
|
||
dispatch_amount=None, dispatch_count=1,
|
||
claimed_amount=None, claimed_count=1,
|
||
success_amount=None, success_count=1):
|
||
"""
|
||
原子更新每日派单统计（支持派单、抢单、成交三种类型）。
|
||
只更新传递了非None参数的字段，未传递的字段保持不变。
|
||
|
||
参数:
|
||
direction: 1=我方派单，2=对方派单
|
||
partner_club_id: 对方俱乐部ID
|
||
dispatch_amount: 派单金额（Decimal），如果提供则累加到 dispatch_amount 和 dispatch_count
|
||
dispatch_count: 派单数量，默认为1，与 dispatch_amount 同时提供时生效
|
||
claimed_amount: 抢单金额（Decimal），如果提供则累加到 claimed_amount 和 claimed_count
|
||
claimed_count: 抢单数量，默认为1，与 claimed_amount 同时提供时生效
|
||
success_amount: 成交金额（Decimal），如果提供则累加到 success_amount 和 success_count
|
||
success_count: 成交数量，默认为1，与 success_amount 同时提供时生效
|
||
"""
|
||
today = date.today()
|
||
year, month, day = today.year, today.month, today.day
|
||
|
||
with transaction.atomic():
|
||
# 使用 select_for_update 锁定行，避免并发累加冲突
|
||
stat, created = DailyDispatchStat.objects.select_for_update().get_or_create(
|
||
date=today,
|
||
year=year,
|
||
month=month,
|
||
day=day,
|
||
direction=direction,
|
||
partner_club_id=partner_club_id,
|
||
defaults={
|
||
'dispatch_count': dispatch_count if dispatch_amount is not None else 0,
|
||
'dispatch_amount': dispatch_amount or Decimal('0.00'),
|
||
'claimed_count': claimed_count if claimed_amount is not None else 0,
|
||
'claimed_amount': claimed_amount or Decimal('0.00'),
|
||
'success_count': success_count if success_amount is not None else 0,
|
||
'success_amount': success_amount or Decimal('0.00'),
|
||
}
|
||
)
|
||
|
||
if not created:
|
||
update_fields = []
|
||
if dispatch_amount is not None:
|
||
stat.dispatch_count = F('dispatch_count') + dispatch_count
|
||
stat.dispatch_amount = F('dispatch_amount') + dispatch_amount
|
||
update_fields.extend(['dispatch_count', 'dispatch_amount'])
|
||
if claimed_amount is not None:
|
||
stat.claimed_count = F('claimed_count') + claimed_count
|
||
stat.claimed_amount = F('claimed_amount') + claimed_amount
|
||
update_fields.extend(['claimed_count', 'claimed_amount'])
|
||
if success_amount is not None:
|
||
stat.success_count = F('success_count') + success_count
|
||
stat.success_amount = F('success_amount') + success_amount
|
||
update_fields.extend(['success_count', 'success_amount'])
|
||
|
||
if update_fields:
|
||
stat.save(update_fields=update_fields)
|
||
|
||
logger.info(f"更新派单统计: 日期={today}, 方向={direction}, 俱乐部={partner_club_id}, "
|
||
f"派单+{dispatch_count if dispatch_amount else 0}单/{dispatch_amount or 0}, "
|
||
f"抢单+{claimed_count if claimed_amount else 0}单/{claimed_amount or 0}, "
|
||
f"成交+{success_count if success_amount else 0}单/{success_amount or 0}")
|
||
|
||
|
||
|
||
|
||
|
||
|
||
def update_daily_income(amount):
|
||
"""
|
||
原子更新当日收入统计（金额累加，笔数加1）。
|
||
用于微信支付成功回调。
|
||
|
||
参数:
|
||
amount: Decimal 本次收入金额
|
||
"""
|
||
today = date.today()
|
||
year, month, day = today.year, today.month, today.day
|
||
|
||
with transaction.atomic():
|
||
stat, created = DailyIncomeStat.objects.select_for_update().get_or_create(
|
||
date=today,
|
||
defaults={
|
||
'year': year,
|
||
'month': month,
|
||
'day': day,
|
||
'total_amount': amount,
|
||
'total_count': 1
|
||
}
|
||
)
|
||
if not created:
|
||
# 原子累加
|
||
stat.total_amount = F('total_amount') + amount
|
||
stat.total_count = F('total_count') + 1
|
||
stat.save(update_fields=['total_amount', 'total_count'])
|
||
|
||
logger.info(
|
||
f"每日收入更新: {today}, +{amount}元, 总金额={stat.total_amount if created else stat.total_amount + amount}, 总笔数={stat.total_count if created else stat.total_count + 1}")
|
||
|
||
|
||
|
||
|
||
|
||
|
||
def update_daily_payout(amount):
|
||
"""
|
||
原子更新当日出款统计（金额累加，笔数加1）。
|
||
用于提现成功、结算打款等场景。
|
||
|
||
参数:
|
||
amount: Decimal 本次出款金额
|
||
"""
|
||
today = date.today()
|
||
year, month, day = today.year, today.month, today.day
|
||
|
||
with transaction.atomic():
|
||
stat, created = DailyPayoutStat.objects.select_for_update().get_or_create(
|
||
date=today,
|
||
defaults={
|
||
'year': year,
|
||
'month': month,
|
||
'day': day,
|
||
'total_amount': amount,
|
||
'total_count': 1
|
||
}
|
||
)
|
||
if not created:
|
||
# 原子累加
|
||
stat.total_amount = F('total_amount') + amount
|
||
stat.total_count = F('total_count') + 1
|
||
stat.save(update_fields=['total_amount', 'total_count'])
|
||
|
||
logger.info(f"每日出款更新: {today}, +{amount}元, 总金额={stat.total_amount if created else stat.total_amount + amount}, 总笔数={stat.total_count if created else stat.total_count + 1}")
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
def calc_shangjia_order_fencheng(jine):
|
||
"""
|
||
商家发单时计算打手/管事分成。
|
||
打手优先：打手+管事 > 订单金额时，管事分成为 0。
|
||
"""
|
||
jine = Decimal(str(jine))
|
||
try:
|
||
lilu_obj = Lilubiao.objects.get(fadanpingtai='3')
|
||
rate_dashou = Decimal(str(lilu_obj.lilu)) if lilu_obj.lilu and lilu_obj.lilu > 0 else Decimal('1')
|
||
except Lilubiao.DoesNotExist:
|
||
rate_dashou = Decimal('1')
|
||
|
||
lilu_guanshi_obj = Lilubiao.objects.filter(fadanpingtai='13').first()
|
||
rate_guanshi = (
|
||
Decimal(str(lilu_guanshi_obj.lilu))
|
||
if lilu_guanshi_obj and lilu_guanshi_obj.lilu is not None
|
||
else Decimal('0')
|
||
)
|
||
|
||
dashou_fencheng = (jine * rate_dashou).quantize(Decimal('0.01'))
|
||
guanshi_raw = (jine * rate_guanshi).quantize(Decimal('0.01'))
|
||
|
||
if dashou_fencheng + guanshi_raw > jine:
|
||
guanshi_fencheng = Decimal('0.00')
|
||
else:
|
||
guanshi_fencheng = guanshi_raw
|
||
|
||
return dashou_fencheng, guanshi_fencheng
|
||
|
||
|
||
def settle_shangjia_order_guanshi_fenhong(order, dashou_id):
|
||
"""
|
||
商家订单结单时结算管事分红（幂等，失败不抛异常阻断主流程）。
|
||
"""
|
||
|
||
if getattr(order, 'fadan_pingtai', None) != 2:
|
||
return
|
||
|
||
guanshi_fencheng = order.guanshi_fencheng or Decimal('0')
|
||
if guanshi_fencheng <= 0:
|
||
return
|
||
|
||
if Gsfenhong.objects.filter(dingdan_id=order.dingdan_id).exists():
|
||
logger.info(f"商家订单管事分红已处理: {order.dingdan_id}")
|
||
return
|
||
|
||
if not dashou_id:
|
||
return
|
||
|
||
try:
|
||
dashou = UserDashou.objects.select_related('user').get(user__yonghuid=dashou_id)
|
||
except UserDashou.DoesNotExist:
|
||
logger.info(f"商家订单管事分红跳过: 打手{dashou_id}不存在")
|
||
return
|
||
|
||
guanshi_id = dashou.yaoqingren
|
||
if not guanshi_id:
|
||
logger.info(f"商家订单管事分红跳过: 打手{dashou_id}无邀请管事")
|
||
return
|
||
|
||
try:
|
||
with transaction.atomic():
|
||
guanshi = UserGuanshi.objects.select_for_update().get(user__yonghuid=guanshi_id)
|
||
UserGuanshi.objects.filter(id=guanshi.id).update(
|
||
yue=F('yue') + guanshi_fencheng,
|
||
chongzhifenrun=F('chongzhifenrun') + guanshi_fencheng,
|
||
)
|
||
|
||
user_main = dashou.user
|
||
Gsfenhong.objects.create(
|
||
dingdan_id=order.dingdan_id,
|
||
guanshi=guanshi_id,
|
||
dashouid=dashou_id,
|
||
shuoming='商家订单分红',
|
||
fenhong=guanshi_fencheng,
|
||
avatar=user_main.avatar if user_main else None,
|
||
nicheng=dashou.nicheng or '未知打手',
|
||
fenhong_leixing=3,
|
||
)
|
||
logger.info(
|
||
f"商家订单管事分红成功: 订单{order.dingdan_id}, 管事{guanshi_id}, "
|
||
f"打手{dashou_id}, 金额{guanshi_fencheng}元"
|
||
)
|
||
except UserGuanshi.DoesNotExist:
|
||
logger.warning(f"商家订单管事分红跳过: 管事{guanshi_id}不存在")
|
||
return
|
||
except Exception as e:
|
||
logger.error(f"商家订单管事分红失败: {e}", exc_info=True)
|
||
return
|
||
|
||
try:
|
||
update_guanshi_daily_by_action(
|
||
yonghuid=guanshi_id,
|
||
action=3,
|
||
amount=guanshi_fencheng,
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"商家订单管事日统计更新失败: {e}")
|