XingQue/along_django
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化服务号订单广播(Celery并发)与群聊状态实时推送

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-06-24 00:43:30 +08:00
parent 56c66846cc
commit 49ba67ecaf
9 changed files with 458 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
a_long_dianjing/settings.py
View File

@@ -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'},
}
# ==================== 定时任务时间配置 ====================

24
dingdan/tasks.py
View File

@@ -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():

82
dingdan/tongzhi_tasks.py
View File

@@ -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
}
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}

176
dingdan/views.py
View File

@@ -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}")

7
houtai/view.py
View File

@@ -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}")

55
peizhi/views.py
View File

@@ -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)

110
utils/chat_utils.py
View File

@@ -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
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

72
utils/order_broadcast.py Normal file
View File

@@ -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()

252
utils/weixin_broadcast.py
View File

@@ -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
}
'total': total,
'success_count': success,
'fail_count': fail,
}
def broadcast_order(self, order_info):
return self.broadcast_parallel(order_info)