shanxia2.0/backend/app/services/wechat_oa_client.py

"""服务号传输。异常只记录错误码，不记录 URL/token/secret/微信响应原文。"""
import threading
import time

import requests

from app.core.config import settings


class WechatAPIError(Exception):
    def __init__(self, code):
        self.code = int(code)
        super().__init__(f"微信接口错误码 {self.code}")


class WechatDeliveryUnknown(Exception):
    """请求可能已被微信接收，不能盲目重发。"""


class WechatOAClient:
    def __init__(self):
        self._lock = threading.Lock()
        self._token = None
        self._expires = 0
        self._identity = None

    def _json(self, method, endpoint, *, sending=False, **kwargs):
        try:
            response = requests.request(method, f"https://api.weixin.qq.com/{endpoint}", timeout=(5, 10), **kwargs)
            response.raise_for_status()
            data = response.json()
            if not isinstance(data, dict):
                raise ValueError("invalid response")
        except (requests.RequestException, ValueError):
            if sending:
                raise WechatDeliveryUnknown("微信发送结果不确定，请核对发送记录") from None
            raise WechatAPIError(-1) from None
        if data.get("errcode", 0):
            raise WechatAPIError(data["errcode"])
        return data

    def access_token(self, force=False):
        if not settings.WECHAT_OA_APPID or not settings.WECHAT_OA_SECRET:
            raise WechatAPIError(40125)
        with self._lock:
            identity = (settings.WECHAT_OA_APPID, settings.WECHAT_OA_SECRET)
            if not force and self._identity == identity and self._token and time.monotonic() < self._expires:
                return self._token
            # stable_token 不因多实例重复取 token 而作废其他实例的有效 token。
            data = self._json("POST", "cgi-bin/stable_token", json={
                "grant_type": "client_credential", "appid": settings.WECHAT_OA_APPID,
                "secret": settings.WECHAT_OA_SECRET, "force_refresh": False,
            })
            token = data.get("access_token")
            if not token:
                raise WechatAPIError(-1)
            self._token, self._identity = token, identity
            self._expires = time.monotonic() + max(1, int(data.get("expires_in", 7200)) - 300)
            return token

    def authenticated(self, method, endpoint, **kwargs):
        for force in (False, True):
            try:
                return self._json(method, endpoint, params={"access_token": self.access_token(force)}, **kwargs)
            except WechatAPIError as exc:
                if force or exc.code not in (40001, 40014, 42001):
                    raise

    def subscribed(self, openid):
        # openid 放在独立参数中，避免打印请求异常泄露查询串。
        data = self.authenticated("POST", "cgi-bin/user/info/batchget", json={"user_list": [{"openid": openid, "lang": "zh_CN"}]})
        users = data.get("user_info_list") or []
        return bool(users and users[0].get("openid") == openid and users[0].get("subscribe") == 1)

    def oauth_openid(self, code):
        data = self._json("GET", "sns/oauth2/access_token", params={
            "appid": settings.WECHAT_OA_APPID, "secret": settings.WECHAT_OA_SECRET,
            "code": code, "grant_type": "authorization_code",
        })
        if not data.get("openid"):
            raise WechatAPIError(-1)
        return data["openid"]

    def send(self, *, openid, template_id, data, event_key, page="", url=""):
        payload = {"touser": openid, "template_id": template_id, "data": data, "client_msg_id": event_key}
        if page and settings.WECHAT_OA_MINI_APPID:
            payload["miniprogram"] = {"appid": settings.WECHAT_OA_MINI_APPID, "pagepath": page}
        if url:
            payload["url"] = url
        result = self.authenticated("POST", "cgi-bin/message/template/send", sending=True, json=payload)
        if not result.get("msgid"):
            raise WechatDeliveryUnknown("微信未返回消息编号，请核对发送记录")
        return str(result["msgid"])


client = WechatOAClient()