diff --git a/a_long_dianjing/settings.py b/a_long_dianjing/settings.py index 07a7973..82a48db 100644 --- a/a_long_dianjing/settings.py +++ b/a_long_dianjing/settings.py @@ -272,6 +272,8 @@ WEIXIN_TEMPLATE_ID = 'Ngs1sT9TPyFaN4cA108YyMV_0nnutTynPQFXoOdNZZE' # 模板ID # 🆕【新增】模板消息配置 WEIXIN_TEMPLATE_MAX_PER_MINUTE = 950 # 每分钟最大发送量（留50缓冲） WEIXIN_TEMPLATE_BATCH_SIZE = 100 # 每批发送数量 +WEIXIN_BROADCAST_ENABLED = True # 服务号订单广播总开关 +WEIXIN_BROADCAST_WORKERS = 30 # Celery 任务内并发推送线程数（勿过大，避免触发微信频控） WEIXIN_MCHID = '1746784529' # 商户号，支付功能需要紧急 @@ -453,6 +455,15 @@ CELERY_TASK_QUEUES = { 'exchange': 'periodic_tasks', 'routing_key': 'periodic_tasks', }, + 'broadcast': { + 'exchange': 'broadcast', + 'routing_key': 'broadcast', + }, +} + +CELERY_TASK_ROUTES = { + 'dingdan.tongzhi_tasks.dingdan_guangbo': {'queue': 'broadcast'}, + 'dingdan.tasks.push_order_status_chat_task': {'queue': 'default'}, } # ==================== 定时任务时间配置 ==================== diff --git a/dingdan/tasks.py b/dingdan/tasks.py index a55b91e..525dd72 100644 --- a/dingdan/tasks.py +++ b/dingdan/tasks.py @@ -70,6 +70,12 @@ def process_expired_order(self, dingdan_id): 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: @@ -177,6 +183,24 @@ def retry_establish_order_chat_task(self, dingdan_id): return str(e) +@shared_task(bind=True, max_retries=2, default_retry_delay=10) +def push_order_status_chat_task(self, dingdan_id): + """订单状态变更后推送群聊状态卡片""" + 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(): diff --git a/dingdan/tongzhi_tasks.py b/dingdan/tongzhi_tasks.py index 874c4b7..490f955 100644 --- a/dingdan/tongzhi_tasks.py +++ b/dingdan/tongzhi_tasks.py @@ -1,71 +1,39 @@ # dingdan/tongzhi_tasks.py import logging -from concurrent.futures import ThreadPoolExecutor, as_completed + +from django.conf import settings + from a_long_dianjing.celery_app import app -from yonghu.models import OfficialAccountUser from utils.weixin_broadcast import WeixinBroadcastSender logger = logging.getLogger('weixin_broadcast') -@app.task(bind=True, max_retries=3, default_retry_delay=60) +@app.task( + bind=True, + name='dingdan.tongzhi_tasks.dingdan_guangbo', + max_retries=2, + default_retry_delay=30, + queue='broadcast', + soft_time_limit=240, + time_limit=300, +) def dingdan_guangbo(self, order_info): """ - 异步并发广播订单模板消息给所有关注服务号的用户 - order_info 字典字段示例： - { - 'dingdan_id': 'PT123456', - 'game_type': '王者荣耀', - 'amount': '88.00', - 'order_desc': '星耀上王者' - } + Celery：并发向所有关注服务号的用户推送订单模板消息 """ - sender = WeixinBroadcastSender() - template_data = sender.build_order_message_data(order_info) + if not getattr(settings, 'WEIXIN_BROADCAST_ENABLED', True): + return {'success': True, 'msg': '广播已关闭'} + order_id = order_info.get('dingdan_id') - - # 查询所有已关注用户 try: - openids = list( - OfficialAccountUser.objects - .filter(is_subscribed=True) - .values_list('official_openid', flat=True) - ) + sender = WeixinBroadcastSender() + result = sender.broadcast_parallel(order_info) + if not result.get('success') and self.request.retries < self.max_retries: + raise self.retry(countdown=30) + return result except Exception as e: - logger.error(f'查询订阅用户失败: {e}') - raise self.retry(exc=e) - - if not openids: - return {'success': True, 'msg': '无订阅用户'} - - total = len(openids) - logger.info(f'📊 Celery 广播订单 {order_id} 开始，目标人数: {total}') - - success = 0 - fail = 0 - - # 20 个线程并发发送，微信接口耗时 200-500ms，1000 人约 15 秒完成 - with ThreadPoolExecutor(max_workers=20) as executor: - future_map = { - executor.submit(sender.send_to_user, openid, template_data, order_id): openid - for openid in openids - } - for future in as_completed(future_map): - openid = future_map[future] - try: - res = future.result(timeout=30) - if res.get('success'): - success += 1 - else: - fail += 1 - except Exception as e: - fail += 1 - logger.error(f'发送异常 {openid[:10]}...: {e}') - - logger.info(f'🏁 广播完成：成功 {success}，失败 {fail}，总计 {total}') - return { - 'success': True, - 'total': total, - 'success_count': success, - 'fail_count': fail - } \ No newline at end of file + logger.error(f'广播任务异常 订单={order_id}: {e}', exc_info=True) + if self.request.retries < self.max_retries: + raise self.retry(exc=e) + return {'success': False, 'msg': str(e), 'order_id': order_id} diff --git a/dingdan/views.py b/dingdan/views.py index fed8cd0..6899a5a 100644 --- a/dingdan/views.py +++ b/dingdan/views.py @@ -666,34 +666,12 @@ class WechatPayNotifyView(APIView): logger.error(f"更新收支记录失败: {e}") def _broadcast_async(self, dingdan): - """异步广播通知""" + """服务号模板消息广播（Celery 优先队列，接口不阻塞）""" try: - def task(): - try: - from . import WeixinBroadcastSender # 请根据实际路径修改 - order_info = { - 'dingdan_id': dingdan.dingdan_id, - 'game_type': self._get_game_type_name(dingdan.leixing_id), - 'amount': str(dingdan.jine), - 'order_desc': (dingdan.jieshao or '')[:50], - } - logger.info(f"准备广播订单: {order_info}") - sender = WeixinBroadcastSender() - result = sender.broadcast_order(order_info) - if result.get('success'): - logger.info(f"订单广播完成: {result.get('msg')}") - else: - logger.warning(f"订单广播异常: {result.get('msg')}") - except Exception as e: - logger.error(f"广播任务异常: {e}") - logger.error(traceback.format_exc()) - - thread = threading.Thread(target=task) - thread.daemon = True - thread.start() - logger.info("广播线程已启动") + from utils.order_broadcast import schedule_dingdan_guangbo_from_order + schedule_dingdan_guangbo_from_order(dingdan) except Exception as e: - logger.error(f"启动广播线程异常: {e}") + logger.error(f'提交广播任务失败: {e}', exc_info=True) def _get_game_type_name(self, leixing_id): try: @@ -1297,27 +1275,11 @@ class VirtualPayNotifyView(APIView): logger.error(f"更新收支记录失败: {e}") def _broadcast_async(self, dingdan): - """异步广播通知（同旧）""" try: - def task(): - try: - from . import WeixinBroadcastSender - order_info = { - 'dingdan_id': dingdan.dingdan_id, - 'game_type': self._get_game_type_name(dingdan.leixing_id), - 'amount': str(dingdan.jine), - 'order_desc': (dingdan.jieshao or '')[:50], - } - sender = WeixinBroadcastSender() - sender.broadcast_order(order_info) - except Exception as e: - logger.error(f"广播任务异常: {e}") - - thread = threading.Thread(target=task) - thread.daemon = True - thread.start() + from utils.order_broadcast import schedule_dingdan_guangbo_from_order + schedule_dingdan_guangbo_from_order(dingdan) except Exception as e: - logger.error(f"启动广播线程异常: {e}") + logger.error(f'提交广播任务失败: {e}', exc_info=True) def _get_game_type_name(self, leixing_id): try: @@ -2292,6 +2254,9 @@ class JiedanView(APIView): dingdan.clkf = current_user.phone if hasattr(current_user, 'phone') else yonghuid dingdan.save() + from utils.chat_utils import schedule_order_status_chat_push + schedule_order_status_chat_push(dingdan.dingdan_id) + # 事务提交后异步通知对方平台 transaction.on_commit( lambda: self._notify_partner_jiedan( @@ -2341,6 +2306,9 @@ class JiedanView(APIView): dingdan.zhuangtai = 3 dingdan.save() + from utils.chat_utils import schedule_order_status_chat_push + schedule_order_status_chat_push(dingdan.dingdan_id) + # 9. 返回成功响应 return Response({ 'code': 0, @@ -2738,16 +2706,10 @@ class ShangjiaPaifaView(APIView): except Exception as e: logger.error(f"跨平台同步失败: {e}", exc_info=True) - # Celery 广播任务（异步） + # Celery 广播任务（broadcast 优先队列） try: - order_info = { - 'dingdan_id': dingdan.dingdan_id, - 'game_type': self._get_game_type_name(dingdan.leixing_id), - 'amount': str(dingdan.jine), - 'order_desc': (dingdan.jieshao or '')[:50], - } - dingdan_guangbo.delay(order_info) - logger.info(f"已提交广播任务: {dingdan.dingdan_id}") + from utils.order_broadcast import schedule_dingdan_guangbo_from_order + schedule_dingdan_guangbo_from_order(dingdan, order_type='商家派单') except Exception as e: logger.error(f"广播任务提交失败: {e}", exc_info=True) @@ -3045,6 +3007,8 @@ class ShangjiaDingdanXiangqingView(APIView): 'beizhu': order.beizhu or '', 'nicheng': order.nicheng or '', 'create_time': order.create_time.strftime('%Y-%m-%d %H:%M:%S') if order.create_time else '', + 'fadanpingtai': order.fadan_pingtai or 0, + 'is_cross': order.is_cross or 0, # 打手信息 'dashou_yonghuid': dashou_yonghuid, @@ -3313,6 +3277,9 @@ class ShangjiaJiesuanView(APIView): action=2 # 2 = 结算 ) + from utils.chat_utils import schedule_order_status_chat_push + schedule_order_status_chat_push(dingdan_id) + settle_ctx = (dingdan_id, jiedan_dashou_id) success_response = Response({ 'code': 0, @@ -3468,6 +3435,9 @@ class ShangjiaChexiaoView(APIView): dingdan.save() + from utils.chat_utils import schedule_order_status_chat_push + schedule_order_status_chat_push(dingdan_id) + return Response({ 'code': 0, 'msg': '撤销成功', @@ -3592,6 +3562,9 @@ class ShangjiaTuikuanShenqingView(APIView): dingdan.tkly = tuikuan_liyou dingdan.save() + from utils.chat_utils import schedule_order_status_chat_push + schedule_order_status_chat_push(dingdan_id) + # ====== 原有功能：更新打手状态 ====== if dingdan.jiedan_dashou_id: try: @@ -4417,7 +4390,7 @@ import logging import requests from rest_framework.views import APIView from rest_framework.response import Response -from rest_framework import status, permissions +from rest_framework import status, permissions, parsers from .models import Dingdan, DingdanPingtai, DingdanShangjia from yonghu.models import UserMain, UserDashou, UserBoss, UserShangjia from peizhi.models import ClubConfig @@ -4497,12 +4470,13 @@ def subscribe_users_to_group(user_ids, group_ids, appkey=None, secret=None): def send_group_message( appkey, secret, group_id, sender_id, sender_name, sender_avatar, message_text='', custom_payload=None, group_name='', order_id='', is_cross=0, - to_data_extra=None, + to_data_extra=None, message_type='text', image_url=None, ): """ 通过 GoEasy REST API 发送群消息。 message_text: 普通文本内容 custom_payload: 自定义消息体（如订单卡片），此时 type 设为 'order' + image_url: 图片完整 URL，message_type='image' 时使用 """ url = 'https://rest-hangzhou.goeasy.io/v2/im/message' sender_avatar = sender_avatar or '' @@ -4519,8 +4493,16 @@ def send_group_message( "orderId": order_id, "isCross": is_cross } - msg_type = 'order' if custom_payload else 'text' - payload = custom_payload if custom_payload else message_text + + if message_type == 'image' and image_url: + msg_type = 'image' + payload = {"url": image_url} + elif custom_payload: + msg_type = 'order' + payload = custom_payload + else: + msg_type = 'text' + payload = message_text body = { "appkey": appkey, @@ -4653,6 +4635,7 @@ class DaiLiQunLiaoXiaoXiView(APIView): 跨平台判断条件：订单同时存在 partner_order_id 和 partner_club_id 即视为跨平台 """ permission_classes = [permissions.IsAuthenticated] + parser_classes = [parsers.MultiPartParser, parsers.FormParser, parsers.JSONParser] def post(self, request): # ===== 1. 参数解析 ===== @@ -4663,6 +4646,7 @@ class DaiLiQunLiaoXiaoXiView(APIView): message_type = request.data.get('messageType', 'text') message_text = request.data.get('text', '') order_payload = request.data.get('orderPayload', None) + image_file = request.FILES.get('file') logger.info(f"[群聊代理] 收到请求: orderId={order_id}, identityType={identity_type}, " f"messageType={message_type}, text={message_text[:20]}...") @@ -4716,24 +4700,56 @@ class DaiLiQunLiaoXiaoXiView(APIView): custom_payload = order_payload if message_type == 'order' else None text_content = message_text + image_url = None + + if message_type == 'image': + if not image_file: + return Response({'code': 400, 'msg': '缺少图片文件'}) + try: + from utils.oss_utils import upload_to_oss, validate_image + import uuid as _uuid + validate_image(image_file) + oss_path = f"chat/images/{order_id}/{_uuid.uuid4().hex}.jpg" + image_url = upload_to_oss(image_file, oss_path) + if not image_url: + return Response({'code': 500, 'msg': '图片上传失败'}) + except Exception as e: + logger.error(f"[群聊代理] 图片上传失败: {e}", exc_info=True) + return Response({'code': 500, 'msg': '图片上传失败'}) pair_to_data = build_pair_to_data_for_order(order, viewer_goeasy_id=sender_id) # ---------- 先向我方群发送消息（永远执行） ---------- - local_ok = send_group_message( - appkey=self_appkey, - secret=self_secret, - group_id=real_group_id, - sender_id=sender_id, - sender_name=sender_name, - sender_avatar=sender_avatar, - message_text=text_content, - custom_payload=custom_payload, - group_name=group_name, - order_id=order.dingdan_id, - is_cross=is_cross, - to_data_extra=pair_to_data, - ) + if message_type == 'image' and image_url: + local_ok = send_group_message( + appkey=self_appkey, + secret=self_secret, + group_id=real_group_id, + sender_id=sender_id, + sender_name=sender_name, + sender_avatar=sender_avatar, + group_name=group_name, + order_id=order.dingdan_id, + is_cross=is_cross, + to_data_extra=pair_to_data, + message_type='image', + image_url=image_url, + ) + else: + local_ok = send_group_message( + appkey=self_appkey, + secret=self_secret, + group_id=real_group_id, + sender_id=sender_id, + sender_name=sender_name, + sender_avatar=sender_avatar, + message_text=text_content, + custom_payload=custom_payload, + group_name=group_name, + order_id=order.dingdan_id, + is_cross=is_cross, + to_data_extra=pair_to_data, + ) if not local_ok: return Response({'code': 500, 'msg': '消息发送失败'}) @@ -5626,6 +5642,9 @@ class DashouTijiaoView(APIView): dingdan.update_time = timezone.now() dingdan.save() + from utils.chat_utils import schedule_order_status_chat_push + schedule_order_status_chat_push(dingdan_id) + # 7. 更新打手扩展表状态 dashou_profile.zhuangtai = 1 # 改为数字1（根据你之前说的） dashou_profile.save() @@ -5720,7 +5739,8 @@ class DashouDingdanXiangqingView(APIView): 'dashou_liuyan': dingdan.dashou_liuyan or '', 'fuwudashou_id': dingdan.jiedan_dashou_id or '', 'fadanpingtai': dingdan.fadan_pingtai or '', - 'tuikuan_liyou': dingdan.tkly or '' + 'tuikuan_liyou': dingdan.tkly or '', + 'is_cross': dingdan.is_cross or 0, } # 6. 查询打手提交的图片 @@ -6348,6 +6368,9 @@ class AdQiangZhiJieDan(APIView): dingdan_obj.zhuangtai = 3 dingdan_obj.save() + from utils.chat_utils import schedule_order_status_chat_push + schedule_order_status_chat_push(dingdan_id) + # 10.2 处理打手结算 if jiedan_dashou_id: try: @@ -9012,13 +9035,14 @@ class ZxsjghdsView(APIView): # 7.4 异步通知打手（广播新订单） try: - order_info = { + from utils.order_broadcast import submit_dingdan_guangbo + submit_dingdan_guangbo({ 'dingdan_id': new_dingdan_id, 'game_type': self._get_game_type_name(leixing_id), 'amount': str(jine), 'order_desc': (jieshao or '')[:50], - } - dingdan_guangbo.delay(order_info) + 'order_type': '商家派单', + }) except Exception as e: logger.error(f"广播通知失败: {e}") diff --git a/houtai/view.py b/houtai/view.py index 49e67c8..7b1a258 100644 --- a/houtai/view.py +++ b/houtai/view.py @@ -10840,13 +10840,14 @@ class ZxkfghdsView(APIView): # ---- 6.5 异步通知（广播新订单、跨平台同步） ---- try: - order_info = { + from utils.order_broadcast import submit_dingdan_guangbo + submit_dingdan_guangbo({ 'dingdan_id': new_dingdan_id, 'game_type': self._get_game_type_name(new_order.leixing_id), 'amount': str(new_order.jine), 'order_desc': (new_order.jieshao or '')[:50], - } - dingdan_guangbo.delay(order_info) + 'order_type': '商家派单', + }) except Exception as e: logger.error(f"广播通知失败: {e}") diff --git a/peizhi/views.py b/peizhi/views.py index 3589212..cdc36c4 100644 --- a/peizhi/views.py +++ b/peizhi/views.py @@ -3272,48 +3272,25 @@ class KehuTianxieDingdanView(APIView): }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) def send_order_notification_async(self, dingdan, youxi_nicheng, zhiding_dashou, zhuangtai): - """异步发送订单通知（客户填写订单后）""" + """客户填写订单后 — Celery 广播服务号模板消息""" try: - # 避免阻塞主线程，使用线程异步执行 - def notification_task(): - try: - # 获取游戏类型名称 - try: - leixing_obj = ShangpinLeixing.objects.get(id=dingdan.leixing_id) - game_type = leixing_obj.jieshao - except: - game_type = '未知游戏' - - # 构建订单信息 - order_info = { - 'dingdan_id': dingdan.dingdan_id, - 'game_type': game_type, - 'amount': str(dingdan.jine), - 'order_desc': dingdan.jieshao[:50] if dingdan.jieshao else '新订单', - 'youxi_nicheng': youxi_nicheng, - 'order_type': '指定订单' if zhuangtai == 7 else '普通订单', - 'zhiding_dashou': zhiding_dashou or '' - } - - # 调用微信广播发送器 - sender = WeixinBroadcastSender() - result = sender.broadcast_order(order_info) - - if result['success']: - logger.info(f"【客户填写订单】微信通知发送成功: {result['msg']}") - else: - logger.warning(f"【客户填写订单】微信通知发送失败: {result['msg']}") - - except Exception as e: - logger.error(f"【客户填写订单】发送通知任务异常: {str(e)}") - - # 启动异步线程 - thread = threading.Thread(target=notification_task) - thread.daemon = True - thread.start() + from utils.order_broadcast import submit_dingdan_guangbo + try: + leixing_obj = ShangpinLeixing.objects.get(id=dingdan.leixing_id) + game_type = leixing_obj.jieshao + except Exception: + game_type = '未知游戏' + order_info = { + 'dingdan_id': dingdan.dingdan_id, + 'game_type': game_type, + 'amount': str(dingdan.jine), + 'order_desc': dingdan.jieshao[:50] if dingdan.jieshao else '新订单', + 'order_type': '指定订单' if zhuangtai == 7 else '普通订单', + } + submit_dingdan_guangbo(order_info) except Exception as e: - logger.error(f"【客户填写订单】启动通知线程异常: {str(e)}") + logger.error(f'【客户填写订单】提交广播失败: {e}', exc_info=True) diff --git a/utils/chat_utils.py b/utils/chat_utils.py index 6ad1348..960cb63 100644 --- a/utils/chat_utils.py +++ b/utils/chat_utils.py @@ -166,7 +166,7 @@ def _ensure_avatar_url(avatar_url): def _build_pair_group_to_data(dashou_goeasy_id, dashou_name, dashou_avatar, partner_goeasy_id, partner_name, partner_avatar, order_id=None, order_desc=None, is_cross=0, - viewer_goeasy_id=None): + viewer_goeasy_id=None, order_zuangtai=None): """群会话 to.data：同时写入打手/对方双方 ID、昵称、头像，前端按当前身份取对方展示""" dashou_avatar = _ensure_avatar_url(dashou_avatar) partner_avatar = _ensure_avatar_url(partner_avatar) @@ -190,6 +190,8 @@ def _build_pair_group_to_data(dashou_goeasy_id, dashou_name, dashou_avatar, to_data['orderId'] = order_id if order_desc: to_data['orderDesc'] = order_desc[:300] + if order_zuangtai is not None: + to_data['orderZhuangtai'] = order_zuangtai if viewer_goeasy_id == dashou_goeasy_id: to_data['name'] = partner_name @@ -220,6 +222,7 @@ def build_pair_to_data_for_order(order, viewer_goeasy_id=None): partner_goeasy_id, partner_name, partner_avatar, order_id=order.dingdan_id, order_desc=order_desc, is_cross=order.is_cross or 0, viewer_goeasy_id=viewer_goeasy_id, + order_zuangtai=order.zhuangtai, ) @@ -832,4 +835,107 @@ def _create_local_group_only(order, dashou_goeasy_id, dashou_name, dashou_avatar "订单已接单。", None, order_id=order.dingdan_id, is_cross=order.is_cross, to_data_extra=to_data, ) - return subscribe_ok or msg_ok \ No newline at end of file + return subscribe_ok or msg_ok + + +STATUS_NOTIFY_TEXT = { + 2: '订单已开始服务', + 3: '订单已完成', + 4: '订单退款申请中', + 5: '订单已退款', + 6: '退款未通过', + 8: '打手已提交交付，订单进入结算中', +} + + +def schedule_order_status_chat_push(dingdan_id): + """订单状态变更后，事务提交后异步推送群聊状态更新""" + if not dingdan_id: + return + + def _run(): + try: + from dingdan.tasks import push_order_status_chat_task + push_order_status_chat_task.apply_async(args=[dingdan_id], countdown=1) + except Exception as e: + logger.error(f"调度订单状态聊天推送失败 {dingdan_id}: {e}", exc_info=True) + + try: + from django.db import transaction + transaction.on_commit(_run) + except Exception: + _run() + + +def push_order_status_chat_update(dingdan_id): + """向订单群推送状态提示 + 最新订单卡片（含实时 zhuangtai）""" + try: + order = Dingdan.objects.select_related('pingtai_kuozhan', 'shangjia_kuozhan').get(dingdan_id=dingdan_id) + except Dingdan.DoesNotExist: + logger.warning(f"push_order_status_chat_update: 订单 {dingdan_id} 不存在") + return False + + dashou_uid = order.jiedan_dashou_id + if not dashou_uid: + return False + + appkey = _get_self_goeasy_appkey() + secret = _get_self_goeasy_secret() + if not appkey: + return False + + dashou_goeasy_id = f'Ds{dashou_uid}' + dashou_name, dashou_avatar = _resolve_goeasy_user_display(dashou_goeasy_id) + partner_goeasy_id, partner_name, partner_avatar = _get_local_partner_info(order) + if not partner_goeasy_id: + return False + + group_id = resolve_group_id_for_order(order, dashou_uid) + order_desc = (order.jieshao or '').strip() + if order.beizhu: + order_desc = f"{order_desc} | 备注:{order.beizhu}" if order_desc else f"备注:{order.beizhu}" + + pair_to_data = _build_pair_group_to_data( + dashou_goeasy_id, dashou_name, dashou_avatar, + partner_goeasy_id, partner_name, partner_avatar, + order_id=order.dingdan_id, order_desc=order_desc, + is_cross=order.is_cross or 0, order_zuangtai=order.zhuangtai, + ) + + user_ids = list({dashou_goeasy_id, partner_goeasy_id}) + _retry_call(_subscribe_users_to_group, 3, 0.4, user_ids, [group_id], appkey, secret) + + status_text = STATUS_NOTIFY_TEXT.get(order.zhuangtai, '订单状态已更新') + notify_text = f"【订单状态】{status_text}" + _retry_call( + _send_group_message, 3, 0.4, + appkey, secret, group_id, dashou_goeasy_id, dashou_name, dashou_avatar, + notify_text, None, order_id=order.dingdan_id, + is_cross=order.is_cross or 0, to_data_extra=pair_to_data, + ) + card_ok = _retry_call( + _push_order_card_message, 3, 0.4, + appkey, secret, order, group_id, dashou_goeasy_id, dashou_name, dashou_avatar, + pair_to_data, + ) + + partner_order_id = order.partner_order_id + partner_club_id = order.partner_club_id + if order.is_cross == 1 and partner_order_id and partner_club_id: + try: + partner_cfg = ClubConfig.objects.get(club_id=partner_club_id, is_self=0) + if partner_cfg and partner_cfg.chat_api_key: + partner_appkey = partner_cfg.chat_api_key + partner_secret = partner_cfg.chat_api_id or '' + partner_group_id = f"group_{partner_order_id}" + agent_id = f"{order.dingdan_id}_{dashou_goeasy_id}" + agent_name = f"[状态]{dashou_name}" + _retry_call( + _send_group_message, 2, 0.4, + partner_appkey, partner_secret, partner_group_id, agent_id, agent_name, '', + notify_text, None, order_id=partner_order_id, is_cross=1, + ) + except Exception as e: + logger.warning(f"跨平台状态推送失败: {e}") + + return card_ok \ No newline at end of file diff --git a/utils/order_broadcast.py b/utils/order_broadcast.py new file mode 100644 index 0000000..45b93c6 --- /dev/null +++ b/utils/order_broadcast.py @@ -0,0 +1,72 @@ +""" +订单服务号模板消息广播 — 统一入口（Celery 异步，接口立即返回） +""" +import logging + +from django.conf import settings + +logger = logging.getLogger('weixin_broadcast') + + +def _game_type_name(leixing_id): + try: + from shangpin.models import ShangpinLeixing + gt = ShangpinLeixing.objects.filter(id=leixing_id).first() + return gt.jieshao if gt else '游戏订单' + except Exception: + return '游戏订单' + + +def build_order_broadcast_payload(dingdan, order_type='普通订单', game_type=None): + return { + 'dingdan_id': dingdan.dingdan_id, + 'game_type': game_type or _game_type_name(dingdan.leixing_id), + 'amount': str(dingdan.jine), + 'order_desc': (dingdan.jieshao or '')[:50], + 'order_type': order_type, + } + + +def submit_dingdan_guangbo(order_info): + """提交 Celery 广播任务（broadcast 优先队列）""" + if not getattr(settings, 'WEIXIN_BROADCAST_ENABLED', True): + logger.info('服务号广播已关闭，跳过') + return False + if not order_info or not order_info.get('dingdan_id'): + return False + try: + from dingdan.tongzhi_tasks import dingdan_guangbo + dingdan_guangbo.apply_async( + args=[order_info], + queue='broadcast', + priority=9, + countdown=0, + ) + logger.info(f'已提交服务号广播任务: {order_info.get("dingdan_id")}') + return True + except Exception as e: + logger.error(f'Celery 广播提交失败，降级线程发送: {e}', exc_info=True) + _fallback_thread_broadcast(order_info) + return False + + +def schedule_dingdan_guangbo_from_order(dingdan, order_type='普通订单', game_type=None): + """根据订单对象构建 payload 并提交广播""" + order_info = build_order_broadcast_payload(dingdan, order_type=order_type, game_type=game_type) + return submit_dingdan_guangbo(order_info) + + +def _fallback_thread_broadcast(order_info): + """Celery 不可用时，后台线程并发发送（不阻塞下单接口）""" + import threading + from utils.weixin_broadcast import WeixinBroadcastSender + + def _run(): + try: + sender = WeixinBroadcastSender() + sender.broadcast_parallel(order_info) + except Exception as ex: + logger.error(f'线程降级广播失败: {ex}', exc_info=True) + + t = threading.Thread(target=_run, daemon=True) + t.start() diff --git a/utils/weixin_broadcast.py b/utils/weixin_broadcast.py index d7a2cb7..70be339 100644 --- a/utils/weixin_broadcast.py +++ b/utils/weixin_broadcast.py @@ -1,31 +1,29 @@ -# utils/weixin_broadcast.py - 强调试版本 -import time +# utils/weixin_broadcast.py — 服务号模板消息并发广播 import logging -import json # 新增，用于格式化打印 +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + import requests -from django.core.cache import cache from django.conf import settings +from django.core.cache import cache + from yonghu.models import OfficialAccountUser logger = logging.getLogger('weixin_broadcast') +_RATE_LIMIT_CODES = {45009, 45011, 45047} + class WeixinBroadcastSender: - """ - 微信模板消息广播发送器 - 调试版 - """ + """微信服务号模板消息发送器（连接复用 + 线程池并发）""" def __init__(self): self.appid = getattr(settings, 'WEIXIN_OFFICIAL_APPID', '') self.secret = getattr(settings, 'WEIXIN_OFFICIAL_SECRET', '') self.template_id = getattr(settings, 'WEIXIN_TEMPLATE_ID', '') - - # 发送控制参数 - self.batch_size = 100 - self.delay_between_batches = 1.5 + self._session = requests.Session() def get_access_token(self): - """获取服务号access_token""" cache_key = f'weixin_official_token_{self.appid}' token = cache.get(cache_key) if token: @@ -34,168 +32,114 @@ class WeixinBroadcastSender: url = 'https://api.weixin.qq.com/cgi-bin/token' params = {'grant_type': 'client_credential', 'appid': self.appid, 'secret': self.secret} try: - response = requests.get(url, params=params, timeout=10) - result = response.json() + resp = self._session.get(url, params=params, timeout=10) + result = resp.json() if 'access_token' in result: token = result['access_token'] expires_in = result.get('expires_in', 7200) - 300 cache.set(cache_key, token, expires_in) - logger.info("✅ 获取access_token成功") return token - else: - logger.error(f"❌ 获取access_token失败: {result}") - return None + logger.error(f'获取 access_token 失败: {result}') except Exception as e: - logger.error(f"❌ 获取access_token异常: {str(e)}") - return None + logger.error(f'获取 access_token 异常: {e}') + return None def build_order_message_data(self, order_info): - """ - 构建订单消息数据 - 【调试关键点1】 - 打印传入的 order_info 和构建出的最终数据 - """ - # 🟡 调试点1：打印传入的参数 - logger.info( - f"🟡【调试-输入】build_order_message_data 收到的 order_info: {json.dumps(order_info, ensure_ascii=False, indent=2)}") - - # 这是你现在的构建逻辑，我们原样保留但加上调试 - # 注意：这里保持你目前的字段名（short_thing5, short_thing4, thing15, amount8） - data = { - "short_thing5": { - "value": order_info.get('order_type', '平台发单'), - "color": "#173177" - }, - "short_thing4": { - "value": order_info.get('game_type', '游戏订单'), - "color": "#173177" - }, - "thing15": { - "value": str(order_info.get('order_desc', '新订单'))[:20], - "color": "#173177" - }, - "amount8": { - "value": f"{order_info.get('amount', '0')}元", - "color": "#173177" - } + order_type = order_info.get('order_type') or ( + '商家派单' if order_info.get('fadan_pingtai') == 2 else '平台发单' + ) + return { + 'short_thing5': {'value': str(order_type)[:20], 'color': '#173177'}, + 'short_thing4': {'value': str(order_info.get('game_type', '游戏订单'))[:20], 'color': '#173177'}, + 'thing15': {'value': str(order_info.get('order_desc', '新订单'))[:20], 'color': '#173177'}, + 'amount8': {'value': f"{order_info.get('amount', '0')}元", 'color': '#173177'}, } - # 🟡 调试点2：打印构建出的最终数据结构 - logger.info(f"🟡【调试-输出】构建完成的模板数据 data 字段：") - for key, content in data.items(): - logger.info(f" 字段名: '{key}' -> 值: '{content.get('value', '')}'") - logger.info(f"🟡【调试-输出】完整 data 结构: {json.dumps(data, ensure_ascii=False)}") + def send_to_user(self, official_openid, template_data, order_id=None, access_token=None): + token = access_token or self.get_access_token() + if not token: + return {'success': False, 'msg': 'access_token 失败', 'retry': True} - # 🔴 关键检查：确认 short_thing5 是否存在，值是否为空 - if 'short_thing5' not in data: - logger.error("❌【致命错误】构建的数据中根本没有 'short_thing5' 这个字段！") - else: - st5_value = data['short_thing5'].get('value', '') - if not st5_value: - logger.error(f"❌【致命错误】'short_thing5' 字段的值为空！") - else: - logger.info(f"✅ 'short_thing5' 字段存在，值为: '{st5_value}'") + request_data = { + 'touser': official_openid, + 'template_id': self.template_id, + 'data': template_data, + } + if order_id and getattr(settings, 'WEIXIN_APPID', None): + request_data['miniprogram'] = { + 'appid': settings.WEIXIN_APPID, + 'pagepath': f'pages/jiedan/jiedan?id={order_id}', + } - return data + url = f'https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={token}' + try: + resp = self._session.post(url, json=request_data, timeout=12) + result = resp.json() + if result.get('errcode') == 0: + return {'success': True, 'msgid': result.get('msgid')} - def send_to_user(self, official_openid, template_data, order_id=None): - """发送模板消息给单个用户 - 【调试关键点2】""" - # 🟡 调试点3：打印每次发送前的数据 - logger.info(f"🟡【调试-发送前】准备发送给 {official_openid[:10]}... 的数据:") - logger.info(f" 模板ID: {self.template_id}") - logger.info( - f" pagepath: pages/dingdan-xiangqing/dingdan-xiangqing?id={order_id}" if order_id else " 无pagepath") - for key, content in template_data.items(): - logger.info(f" 字段 '{key}': '{content.get('value', '')}'") + errcode = result.get('errcode') + if errcode in (43004, 43101): + OfficialAccountUser.objects.filter(official_openid=official_openid).update(is_subscribed=False) + retry = errcode in _RATE_LIMIT_CODES + return {'success': False, 'msg': result.get('errmsg'), 'errcode': errcode, 'retry': retry} + except Exception as e: + return {'success': False, 'msg': str(e), 'retry': True} + + def broadcast_parallel(self, order_info): + order_id = order_info.get('dingdan_id') + template_data = self.build_order_message_data(order_info) + + openids = list( + OfficialAccountUser.objects.filter(is_subscribed=True).values_list('official_openid', flat=True) + ) + total = len(openids) + if total == 0: + return {'success': True, 'msg': '无订阅用户', 'total': 0, 'success_count': 0, 'fail_count': 0} + + max_workers = min( + int(getattr(settings, 'WEIXIN_BROADCAST_WORKERS', 30)), + max(8, total), + ) + logger.info(f'广播开始 订单={order_id} 人数={total} 并发={max_workers}') access_token = self.get_access_token() if not access_token: - return {'success': False, 'msg': '获取access_token失败', 'retry': True} + return {'success': False, 'msg': 'access_token 失败', 'total': total} - request_data = { - "touser": official_openid, - "template_id": self.template_id, - "data": template_data - } + success = 0 + fail = 0 + rate_limited = 0 - if order_id and hasattr(settings, 'WEIXIN_APPID'): - request_data["miniprogram"] = { - "appid": settings.WEIXIN_APPID, - "pagepath": f"pages/jiedan/jiedan?id={order_id}" - } + def _send_one(openid): + return self.send_to_user(openid, template_data, order_id, access_token=access_token) - send_url = f'https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={access_token}' - try: - response = requests.post(send_url, json=request_data, timeout=10) - result = response.json() - logger.info(f"🟡【调试-微信响应】{official_openid[:10]}... 的发送结果: {result}") + with ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = {pool.submit(_send_one, oid): oid for oid in openids} + for fut in as_completed(futures): + oid = futures[fut] + try: + res = fut.result(timeout=30) + if res.get('success'): + success += 1 + else: + fail += 1 + if res.get('retry'): + rate_limited += 1 + except Exception as e: + fail += 1 + logger.warning(f'发送异常 {oid[:8]}...: {e}') - if result.get('errcode') == 0: - return {'success': True, 'msg': '发送成功', 'msgid': result.get('msgid')} - else: - # 如果是未关注或拒收，更新本地状态 - if result.get('errcode') in [43004, 43101]: - try: - OfficialAccountUser.objects.filter(official_openid=official_openid).update(is_subscribed=False) - except: - pass - return {'success': False, 'msg': f"[{result.get('errcode')}]{result.get('errmsg')}", 'retry': False} + if rate_limited > 5: + time.sleep(1.5) - except Exception as e: - logger.error(f"❌ 发送给用户异常: {official_openid[:10]}..., 错误: {str(e)}") - return {'success': False, 'msg': f'请求异常: {str(e)}', 'retry': True} - - def broadcast_order(self, order_info): - """广播订单消息给所有关注用户 - 【调试入口】""" - logger.info(f"🚀【开始广播】订单: {order_info.get('dingdan_id')}") - - # 1. 获取所有关注用户 - try: - subscribed_users = OfficialAccountUser.objects.filter(is_subscribed=True).values_list('official_openid', - flat=True) - total_users = subscribed_users.count() - logger.info(f"📊【广播统计】将从数据库查询到 {total_users} 个订阅用户") - if total_users == 0: - return {'success': True, 'msg': '无订阅用户', 'total': 0} - except Exception as e: - logger.error(f"❌ 获取订阅用户列表失败: {str(e)}") - return {'success': False, 'msg': f'获取用户失败: {str(e)}'} - - # 2. 构建模板数据（这里会触发上面的调试打印） - template_data = self.build_order_message_data(order_info) - order_id = order_info.get('dingdan_id') - - # 3. 分批发送 - success_count = 0 - fail_count = 0 - current_batch = 0 - openid_list = list(subscribed_users) - - logger.info(f"🟡【调试-广播循环】开始遍历 {len(openid_list)} 个用户进行发送...") - for i, openid in enumerate(openid_list): - try: - result = self.send_to_user(openid, template_data, order_id) - if result['success']: - success_count += 1 - else: - fail_count += 1 - if fail_count <= 5: # 只打印前几个失败原因 - logger.warning(f"⚠️ 发送失败 {openid[:10]}...: {result.get('msg')}") - - # 分批控制 - current_batch += 1 - if current_batch >= self.batch_size: - time.sleep(self.delay_between_batches) - current_batch = 0 - - except Exception as e: - fail_count += 1 - logger.error(f"❌ 处理用户 {openid[:10]}... 异常: {str(e)}") - - logger.info(f"🏁【广播完成】订单 {order_id}，总计: {total_users}，成功: {success_count}，失败: {fail_count}") + logger.info(f'广播完成 订单={order_id} 成功={success} 失败={fail} 总计={total}') return { 'success': True, - 'msg': f'广播完成: 成功{success_count}, 失败{fail_count}', - 'total': total_users, - 'success_count': success_count, - 'fail_count': fail_count - } \ No newline at end of file + 'total': total, + 'success_count': success, + 'fail_count': fail, + } + + def broadcast_order(self, order_info): + return self.broadcast_parallel(order_info)