95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
import logging
|
||
|
||
import requests
|
||
from django.conf import settings
|
||
from django.core.cache import cache
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
WX_TOKEN_CACHE_KEY = 'wx_mini_access_token'
|
||
|
||
|
||
def invalidate_weixin_mini_access_token():
|
||
cache.delete(WX_TOKEN_CACHE_KEY)
|
||
|
||
|
||
def get_weixin_mini_access_token(*, force_refresh=False):
|
||
"""
|
||
获取小程序 access_token。
|
||
force_refresh=True 时清除缓存并强制向微信重新获取，避免 token 非最新导致 40001。
|
||
优先调用 stable_token，失败时回退普通 token 接口。
|
||
"""
|
||
if not force_refresh:
|
||
token = cache.get(WX_TOKEN_CACHE_KEY)
|
||
if token:
|
||
return token
|
||
|
||
if force_refresh:
|
||
invalidate_weixin_mini_access_token()
|
||
|
||
appid = getattr(settings, 'WEIXIN_APPID', '')
|
||
secret = getattr(settings, 'WEIXIN_SECRET', '')
|
||
if not appid or not secret:
|
||
logger.error('微信小程序配置缺失: WEIXIN_APPID 或 WEIXIN_SECRET 未设置')
|
||
return None
|
||
|
||
token, expires_in = _fetch_stable_token(appid, secret, force_refresh)
|
||
if not token:
|
||
token, expires_in = _fetch_legacy_token(appid, secret)
|
||
|
||
if token:
|
||
ttl = max(60, int(expires_in or 7200) - 200)
|
||
cache.set(WX_TOKEN_CACHE_KEY, token, ttl)
|
||
return token
|
||
|
||
|
||
def is_weixin_token_invalid(errcode=None, errmsg=''):
|
||
if errcode in (40001, 40014, 42001):
|
||
return True
|
||
msg = (errmsg or '').lower()
|
||
return 'invalid credential' in msg or 'access_token is invalid' in msg
|
||
|
||
|
||
def _fetch_stable_token(appid, secret, force_refresh):
|
||
url = 'https://api.weixin.qq.com/cgi-bin/stable_token'
|
||
payload = {
|
||
'grant_type': 'client_credential',
|
||
'appid': appid,
|
||
'secret': secret,
|
||
'force_refresh': bool(force_refresh),
|
||
}
|
||
try:
|
||
resp = requests.post(url, json=payload, timeout=5)
|
||
data = resp.json()
|
||
token = data.get('access_token')
|
||
if token:
|
||
logger.info('成功获取微信 stable access_token')
|
||
return token, data.get('expires_in', 7200)
|
||
logger.warning(
|
||
'stable_token 获取失败 errcode=%s errmsg=%s',
|
||
data.get('errcode'), data.get('errmsg'),
|
||
)
|
||
except Exception as e:
|
||
logger.warning('请求 stable_token 异常: %s', e)
|
||
return None, None
|
||
|
||
|
||
def _fetch_legacy_token(appid, secret):
|
||
url = 'https://api.weixin.qq.com/cgi-bin/token'
|
||
params = {
|
||
'grant_type': 'client_credential',
|
||
'appid': appid,
|
||
'secret': secret,
|
||
}
|
||
try:
|
||
resp = requests.get(url, params=params, timeout=5)
|
||
data = resp.json()
|
||
token = data.get('access_token')
|
||
if token:
|
||
logger.info('成功获取微信 access_token (legacy)')
|
||
return token, data.get('expires_in', 7200)
|
||
logger.error('legacy token 获取失败: %s', data)
|
||
except Exception as e:
|
||
logger.error('请求 legacy token 异常: %s', e)
|
||
return None, None
|