292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""
|
|
阿龙电竞 - 订单定时任务
|
|
处理订单超时自动确认等逻辑
|
|
完整版本，包含所有必要的异常处理
|
|
"""
|
|
|
|
from celery import shared_task
|
|
from django.db import transaction
|
|
from django.db.models import F, Q
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
from decimal import Decimal
|
|
import logging
|
|
|
|
# 导入模型 - 确保路径正确
|
|
from dingdan.models import Dingdan, DingdanShangjia
|
|
from yonghu.models import UserMain, UserDashou, UserShangjia
|
|
from utils.celery_utils import safe_decimal_operation, log_task_execution, rollback_on_failure
|
|
from utils.celery_guard import celery_non_broadcast_blocked
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@shared_task(bind=True, max_retries=3, default_retry_delay=60)
|
|
def process_expired_order(self, dingdan_id):
|
|
"""
|
|
处理单个超时订单（订单状态=8，超过48小时未处理）
|
|
"""
|
|
blocked = celery_non_broadcast_blocked()
|
|
if blocked:
|
|
logger.info('跳过 process_expired_order(%s): %s', dingdan_id, blocked)
|
|
return blocked
|
|
|
|
if not getattr(settings, 'ORDER_AUTO_SETTLE_ENABLED', False):
|
|
logger.info('订单自动结算已禁用，跳过 process_expired_order(%s)', dingdan_id)
|
|
return f'自动结算已禁用，跳过订单{dingdan_id}'
|
|
|
|
try:
|
|
guanshi_settle_ctx = None
|
|
# 使用select_for_update锁定记录，防止并发修改
|
|
with transaction.atomic():
|
|
|
|
# 获取订单，同时锁定记录
|
|
order = Dingdan.objects.select_for_update().get(
|
|
dingdan_id=dingdan_id
|
|
)
|
|
|
|
|
|
|
|
# 其余代码不变...
|
|
# ... 保持原有代码
|
|
|
|
|
|
|
|
# 再次检查订单状态是否为8
|
|
if order.zhuangtai != 8:
|
|
logger.info(f"订单{dingdan_id}状态已不是8（当前:{order.zhuangtai}），跳过处理")
|
|
return f"订单{dingdan_id}状态已变更，跳过处理"
|
|
|
|
# 获取发单平台类型
|
|
fadan_pingtai = order.fadan_pingtai
|
|
|
|
# 获取打手ID和分成金额
|
|
dashou_id = order.jiedan_dashou_id
|
|
dashou_fencheng = safe_decimal_operation(order.dashou_fencheng)
|
|
|
|
# 1. 更新订单状态为3（已完成）
|
|
order.zhuangtai = 3
|
|
order.save(update_fields=['zhuangtai'])
|
|
|
|
try:
|
|
from utils.chat_utils import schedule_order_status_chat_push
|
|
schedule_order_status_chat_push(dingdan_id)
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. 更新打手信息（如果存在）
|
|
updated_dashou = False
|
|
if dashou_id:
|
|
try:
|
|
# 查询打手用户主表
|
|
dashou_user = UserMain.objects.filter(yonghuid=dashou_id).first()
|
|
if dashou_user:
|
|
# 获取打手扩展表
|
|
dashou_profile = getattr(dashou_user, 'dashou_profile', None)
|
|
if dashou_profile:
|
|
# 使用F表达式原子更新
|
|
UserDashou.objects.filter(pk=dashou_profile.pk).update(
|
|
chengjiaozongliang=F('chengjiaozongliang') + 1,
|
|
yue=F('yue') + dashou_fencheng,
|
|
zonge=F('zonge') + dashou_fencheng,
|
|
jinrishouyi=F('jinrishouyi') + dashou_fencheng,
|
|
jinyueshouyi=F('jinyueshouyi') + dashou_fencheng
|
|
)
|
|
updated_dashou = True
|
|
logger.info(f"更新打手{dashou_id}信息成功")
|
|
except Exception as e:
|
|
logger.error(f"更新打手信息失败，打手ID: {dashou_id}, 错误: {str(e)}")
|
|
# 继续执行，不中断流程
|
|
|
|
# 3. 如果是商家发单，更新商家信息
|
|
updated_shangjia = False
|
|
if fadan_pingtai == 2: # 商家发单
|
|
try:
|
|
# 获取商家扩展信息
|
|
shangjia_kuozhan = getattr(order, 'shangjia_kuozhan', None)
|
|
if shangjia_kuozhan:
|
|
shangjia_id = shangjia_kuozhan.shangjia_id
|
|
if shangjia_id:
|
|
# 查询商家用户
|
|
shangjia_user = UserMain.objects.filter(yonghuid=shangjia_id).first()
|
|
if shangjia_user:
|
|
shangjia_profile = getattr(shangjia_user, 'shop_profile', None)
|
|
if shangjia_profile:
|
|
# 更新商家成交订单数量
|
|
UserShangjia.objects.filter(pk=shangjia_profile.pk).update(
|
|
chengjiao=F('chengjiao') + 1
|
|
)
|
|
updated_shangjia = True
|
|
logger.info(f"更新商家{shangjia_id}信息成功")
|
|
except Exception as e:
|
|
logger.error(f"更新商家信息失败，订单ID: {dingdan_id}, 错误: {str(e)}")
|
|
# 继续执行，不中断流程
|
|
|
|
# 记录任务执行结果
|
|
result_msg = f"订单{dingdan_id}自动结算完成"
|
|
if updated_dashou:
|
|
result_msg += "，打手信息已更新"
|
|
if updated_shangjia:
|
|
result_msg += "，商家信息已更新"
|
|
|
|
log_task_execution(f"自动结算订单 {dingdan_id}", True, result_msg)
|
|
|
|
if fadan_pingtai == 2 and dashou_id:
|
|
guanshi_settle_ctx = (dingdan_id, dashou_id)
|
|
|
|
if guanshi_settle_ctx:
|
|
try:
|
|
from dingdan.utils import settle_shangjia_order_guanshi_fenhong
|
|
settle_order = Dingdan.objects.get(dingdan_id=guanshi_settle_ctx[0])
|
|
settle_shangjia_order_guanshi_fenhong(settle_order, guanshi_settle_ctx[1])
|
|
except Exception as e:
|
|
logger.error(f"自动结算管事分红失败 订单{dingdan_id}: {e}", exc_info=True)
|
|
|
|
return result_msg
|
|
|
|
except Dingdan.DoesNotExist:
|
|
error_msg = f"订单{dingdan_id}不存在"
|
|
logger.warning(error_msg)
|
|
return error_msg
|
|
except Exception as e:
|
|
error_msg = f"处理订单{dingdan_id}时发生错误: {str(e)}"
|
|
logger.error(error_msg)
|
|
|
|
# 重试机制
|
|
if self.request.retries < self.max_retries:
|
|
logger.info(f"订单{dingdan_id}处理失败，准备第{self.request.retries + 1}次重试")
|
|
raise self.retry(exc=e)
|
|
|
|
log_task_execution(f"自动结算订单 {dingdan_id}", False, error_msg)
|
|
return error_msg
|
|
|
|
|
|
@shared_task(bind=True, max_retries=3, default_retry_delay=15)
|
|
def retry_establish_order_chat_task(self, dingdan_id):
|
|
"""抢单后建聊失败时的延迟补偿任务"""
|
|
blocked = celery_non_broadcast_blocked()
|
|
if blocked:
|
|
return blocked
|
|
|
|
try:
|
|
from utils.chat_utils import establish_order_chat_with_retry
|
|
ok = establish_order_chat_with_retry(dingdan_id, max_rounds=5)
|
|
if ok:
|
|
logger.info(f"建聊补偿成功: 订单{dingdan_id}")
|
|
return f'chat ok {dingdan_id}'
|
|
logger.warning(f"建聊补偿仍失败: 订单{dingdan_id}")
|
|
if self.request.retries < self.max_retries:
|
|
raise self.retry()
|
|
return f'chat failed {dingdan_id}'
|
|
except Exception as e:
|
|
logger.error(f"建聊补偿任务异常 订单{dingdan_id}: {e}", exc_info=True)
|
|
if self.request.retries < self.max_retries:
|
|
raise self.retry(exc=e)
|
|
return str(e)
|
|
|
|
|
|
@shared_task(bind=True, max_retries=2, default_retry_delay=10)
|
|
def push_order_status_chat_task(self, dingdan_id):
|
|
"""订单状态变更后推送群聊状态卡片"""
|
|
blocked = celery_non_broadcast_blocked()
|
|
if blocked:
|
|
return blocked
|
|
|
|
try:
|
|
from utils.chat_utils import push_order_status_chat_update
|
|
ok = push_order_status_chat_update(dingdan_id)
|
|
if ok:
|
|
return f'status chat ok {dingdan_id}'
|
|
if self.request.retries < self.max_retries:
|
|
raise self.retry()
|
|
return f'status chat failed {dingdan_id}'
|
|
except Exception as e:
|
|
logger.error(f"订单状态聊天推送异常 {dingdan_id}: {e}", exc_info=True)
|
|
if self.request.retries < self.max_retries:
|
|
raise self.retry(exc=e)
|
|
return str(e)
|
|
|
|
|
|
@shared_task
|
|
@rollback_on_failure("批量检查超时订单")
|
|
def check_order_expire_task():
|
|
"""
|
|
批量检查超时订单（补偿机制）
|
|
每5分钟执行一次，作为信号处理的备用方案
|
|
查找状态为8且创建时间超过72小时的订单（宽松时间，防止误处理）
|
|
"""
|
|
blocked = celery_non_broadcast_blocked()
|
|
if blocked:
|
|
logger.info('跳过 check_order_expire_task: %s', blocked)
|
|
return {"success": True, "message": blocked, "count": 0}
|
|
|
|
if not getattr(settings, 'ORDER_AUTO_SETTLE_ENABLED', False):
|
|
logger.info('订单自动结算已禁用，跳过 check_order_expire_task')
|
|
return {"success": True, "message": "自动结算已禁用", "count": 0}
|
|
|
|
try:
|
|
# 计算更宽松的时间点：当前时间减去72小时（48+24小时）
|
|
# 这样避免误处理刚变为状态8的订单
|
|
check_time = timezone.now() - timedelta(seconds=settings.ORDER_EXPIRE_SECONDS + 86400) # 48+24=72小时
|
|
|
|
# 查询状态为8且创建时间早于检查时间的订单
|
|
expired_orders = Dingdan.objects.filter(
|
|
zhuangtai=8,
|
|
create_time__lt=check_time
|
|
).values_list('dingdan_id', flat=True)
|
|
|
|
expired_count = expired_orders.count()
|
|
|
|
if expired_count == 0:
|
|
logger.info("补偿检查：无超时订单")
|
|
return {"success": True, "message": "无超时订单", "count": 0}
|
|
|
|
# 分批处理超时订单（避免一次处理太多）
|
|
batch_size = 10
|
|
processed_count = 0
|
|
|
|
for i in range(0, expired_count, batch_size):
|
|
batch = expired_orders[i:i+batch_size]
|
|
|
|
# 异步处理每个超时订单
|
|
for dingdan_id in batch:
|
|
process_expired_order.apply_async(
|
|
args=[dingdan_id],
|
|
queue='order_tasks',
|
|
priority=8 # 补偿任务的优先级低一些
|
|
)
|
|
processed_count += 1
|
|
|
|
result_msg = f"补偿检查：发现{expired_count}个可能超时的订单，已提交处理{processed_count}个"
|
|
log_task_execution("批量检查超时订单", True, result_msg)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": result_msg,
|
|
"total_count": expired_count,
|
|
"processed_count": processed_count
|
|
}
|
|
|
|
except Exception as e:
|
|
error_msg = f"批量检查超时订单失败: {str(e)}"
|
|
logger.error(error_msg)
|
|
log_task_execution("批量检查超时订单", False, error_msg)
|
|
|
|
return {
|
|
"success": False,
|
|
"message": error_msg,
|
|
"error": str(e)
|
|
}
|
|
|
|
|
|
|
|
|