along_django/utils/weixin_broadcast.py

# utils/weixin_broadcast.py — 服务号模板消息并发广播
import logging
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests
from django.conf import settings
from django.core.cache import cache

from utils.xcx_sys_config import wx_cfg
from yonghu.models import OfficialAccountUser

logger = logging.getLogger('weixin_broadcast')

_RATE_LIMIT_CODES = {45009, 45011, 45047}


class WeixinBroadcastSender:
    """微信服务号模板消息发送器（连接复用 + 线程池并发）"""

    def __init__(self):
        self.appid = wx_cfg.WEIXIN_OFFICIAL_APPID
        self.secret = wx_cfg.WEIXIN_OFFICIAL_SECRET
        self.template_id = wx_cfg.WEIXIN_TEMPLATE_ID
        self._session = requests.Session()

    def get_access_token(self):
        cache_key = f'weixin_official_token_{self.appid}'
        token = cache.get(cache_key)
        if token:
            return token

        url = 'https://api.weixin.qq.com/cgi-bin/token'
        params = {'grant_type': 'client_credential', 'appid': self.appid, 'secret': self.secret}
        try:
            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)
                return token
            logger.error(f'获取 access_token 失败: {result}')
        except Exception as e:
            logger.error(f'获取 access_token 异常: {e}')
        return None

    def build_order_message_data(self, order_info):
        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'},
        }

    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}

        request_data = {
            'touser': official_openid,
            'template_id': self.template_id,
            'data': template_data,
        }
        if order_id and wx_cfg.WEIXIN_APPID:
            request_data['miniprogram'] = {
                'appid': wx_cfg.WEIXIN_APPID,
                'pagepath': f'pages/jiedan/jiedan?id={order_id}',
            }

        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')}

            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(wx_cfg.WEIXIN_BROADCAST_WORKERS),
            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 失败', 'total': total}

        success = 0
        fail = 0
        rate_limited = 0

        def _send_one(openid):
            return self.send_to_user(openid, template_data, order_id, access_token=access_token)

        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 rate_limited > 5:
            time.sleep(1.5)

        logger.info(f'广播完成 订单={order_id} 成功={success} 失败={fail} 总计={total}')
        return {
            'success': True,
            'total': total,
            'success_count': success,
            'fail_count': fail,
        }

    def broadcast_order(self, order_info):
        return self.broadcast_parallel(order_info)