zkxw/backend/app/services/wechat_mp_auth.py

"""微信服务号：access_token、oauth code 换 openid。"""
from __future__ import annotations

import logging
from datetime import datetime, timedelta
from typing import Any, Dict, Optional

import httpx

from app.core.config import settings

logger = logging.getLogger(__name__)

_token: Optional[str] = None
_token_expire: Optional[datetime] = None


class WechatMpAuthError(Exception):
    pass


def _require_mp_credentials() -> tuple:
    appid = (settings.WX_MP_APPID or "").strip()
    secret = (settings.WX_MP_SECRET or "").strip()
    if not appid or not secret:
        raise WechatMpAuthError("未配置服务号凭证（WX_MP_APPID / WX_MP_SECRET）")
    return appid, secret


def get_mp_access_token() -> str:
    global _token, _token_expire
    if _token and _token_expire and datetime.now() < _token_expire:
        return _token
    appid, secret = _require_mp_credentials()
    with httpx.Client(timeout=15.0) as client:
        r = client.get(
            "https://api.weixin.qq.com/cgi-bin/token",
            params={
                "grant_type": "client_credential",
                "appid": appid,
                "secret": secret,
            },
        )
        r.raise_for_status()
        data = r.json()
    token = data.get("access_token")
    if not token:
        logger.warning("mp access_token failed: %s", data)
        raise WechatMpAuthError("获取服务号 access_token 失败")
    _token = token
    _token_expire = datetime.now() + timedelta(seconds=int(data.get("expires_in", 7200)) - 300)
    return token


def oauth_code_to_openid(code: str) -> Dict[str, Any]:
    """snsapi_base：code → openid / unionid。"""
    c = (code or "").strip()
    if not c:
        raise WechatMpAuthError("code 不能为空")
    appid, secret = _require_mp_credentials()
    with httpx.Client(timeout=15.0) as client:
        r = client.get(
            "https://api.weixin.qq.com/sns/oauth2/access_token",
            params={
                "appid": appid,
                "secret": secret,
                "code": c,
                "grant_type": "authorization_code",
            },
        )
        r.raise_for_status()
        data = r.json()
    if data.get("errcode"):
        raise WechatMpAuthError(data.get("errmsg") or "服务号授权失败")
    openid = data.get("openid")
    if not openid:
        raise WechatMpAuthError("服务号未返回 openid")
    return {"openid": openid, "unionid": data.get("unionid")}