Django/orders/tasks.py

"""
阿龙电竞 - 订单定时任务
处理订单超时自动确认等逻辑
完整版本，包含所有必要的异常处理
"""

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 orders.models import Dingdan
from users.models import UserMain, UserDashou, UserShangjia
from utils.celery_utils import safe_decimal_operation, log_task_execution, rollback_on_failure

# from users.utils import update_platform_profit  # 导入函数

logger = logging.getLogger(__name__)



@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def process_expired_order(self, dingdan_id):
    """
    处理单个超时订单（订单状态=8，超过48小时未处理）
    """
    try:
        # 使用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'])

            # 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 += "，商家信息已更新"

            if fadan_pingtai == 1:
                try:
                    update_platform_profit(dingdan_id)
                except Exception as e:
                    # 平台收益更新失败不影响主流程，但记录日志
                    logger.error(f"强制结单后更新平台收益失败: {str(e)}")    

            log_task_execution(f"自动结算订单 {dingdan_id}", True, result_msg)
            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
@rollback_on_failure("批量检查超时订单")
def check_order_expire_task():
    """
    批量检查超时订单（补偿机制）
    每5分钟执行一次，作为信号处理的备用方案
    查找状态为8且创建时间超过72小时的订单（宽松时间，防止误处理）
    """
    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)
        }