import threading import requests from django.conf import settings from peizhi.models import Club from shangpin.models import ShangpinLeixing from dingdan.models import Dingdan # 根据实际路径导入 import logging from datetime import date from django.db import transaction from django.db.models import F from peizhi.models import DailyDispatchStat # 假设模型在 peizhi 应用中 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 actual_amount -= 1 # 构建请求数据 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: import time time.sleep(2 ** attempt) # 指数退避 else: logger.error(f"同步订单 {dingdan.dingdan_id} 到俱乐部 {club.club_id} 最终失败") # 获取下单方ID（根据订单类型） def get_order_user_id(order): from dingdan.models import DingdanPingtai, DingdanShangjia 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 # dingdan/utils/dispatch_stat.py import logging import logging from datetime import date from decimal import Decimal from django.db import transaction from django.db.models import F from peizhi.models import DailyDispatchStat # 请根据实际模型位置修改导入路径 logger = logging.getLogger(__name__) 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}") from peizhi.models import DailyIncomeStat logger = logging.getLogger(__name__) 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}") from peizhi.models import DailyPayoutStat logger = logging.getLogger(__name__) 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}") from decimal import Decimal from dingdan.models import Lilubiao 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): """ 商家订单结单时结算管事分红（幂等，失败不抛异常阻断主流程）。 """ from shangpin.models import Gsfenhong from yonghu.models import UserDashou, UserGuanshi 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: from houtai.utils import update_guanshi_daily_by_action update_guanshi_daily_by_action( yonghuid=guanshi_id, action=3, amount=guanshi_fencheng, ) except Exception as e: logger.error(f"商家订单管事日统计更新失败: {e}")