71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
# dingdan/tongzhi_tasks.py
|
||
import logging
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
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)
|
||
def dingdan_guangbo(self, order_info):
|
||
"""
|
||
异步并发广播订单模板消息给所有关注服务号的用户
|
||
order_info 字典字段示例：
|
||
{
|
||
'dingdan_id': 'PT123456',
|
||
'game_type': '王者荣耀',
|
||
'amount': '88.00',
|
||
'order_desc': '星耀上王者'
|
||
}
|
||
"""
|
||
sender = WeixinBroadcastSender()
|
||
template_data = sender.build_order_message_data(order_info)
|
||
order_id = order_info.get('dingdan_id')
|
||
|
||
# 查询所有已关注用户
|
||
try:
|
||
openids = list(
|
||
OfficialAccountUser.objects
|
||
.filter(is_subscribed=True)
|
||
.values_list('official_openid', flat=True)
|
||
)
|
||
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
|
||
} |