98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
"""模块：微信 access_token 管理
|
||
|
||
支持小程序(mini)与公众号(official)两种类型，基于 Django cache 缓存。
|
||
无 Redis 时仍可用（locmem 或内存降级，多进程下可能重复获取 token）。
|
||
|
||
配置（secrets.ini [wechat] 段）：
|
||
APPID / SECRET 小程序或默认应用
|
||
OFFICIAL_APPID / OFFICIAL_SECRET 公众号（若与小程序不同，未配置则回退 APPID/SECRET）
|
||
|
||
常用 API：
|
||
from gvsds.wechat import AccessToken
|
||
token = AccessToken.Get('mini') # 带缓存
|
||
token = AccessToken.Get('official')
|
||
token = AccessToken.Refresh('mini') # 强制刷新
|
||
"""
|
||
import logging
|
||
|
||
from django.core.cache import cache
|
||
|
||
from gvsds.secrets import get as _secret_get
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_TOKEN_URL = 'https://api.weixin.qq.com/cgi-bin/token'
|
||
_CACHE_PREFIX = 'gvsds:wechat:token'
|
||
_CACHE_TTL = 7000 # 微信 token 7200s，留 200s 余量
|
||
_VALID_TYPES = ('mini', 'official')
|
||
|
||
|
||
class AccessToken:
|
||
"""微信 access_token 管理器。
|
||
|
||
通过 Get 获取（带缓存），通过 Refresh 强制刷新。
|
||
缓存 key 形如 gvsds:wechat:token:{type}:{appid}。
|
||
"""
|
||
|
||
@staticmethod
|
||
def _GetConfig(app_type):
|
||
"""读取对应类型的 AppID/Secret，official 未配置时回退默认。"""
|
||
if app_type == 'official':
|
||
appid = _secret_get('wechat', 'OFFICIAL_APPID') or _secret_get('wechat', 'APPID')
|
||
secret = _secret_get('wechat', 'OFFICIAL_SECRET') or _secret_get('wechat', 'SECRET')
|
||
else:
|
||
appid = _secret_get('wechat', 'APPID')
|
||
secret = _secret_get('wechat', 'SECRET')
|
||
if not appid or not secret:
|
||
raise RuntimeError(f'[AccessToken] 未配置 wechat {app_type} 的 APPID/SECRET')
|
||
return appid, secret
|
||
|
||
@staticmethod
|
||
def _CacheKey(app_type, appid):
|
||
return f'{_CACHE_PREFIX}:{app_type}:{appid}'
|
||
|
||
@classmethod
|
||
def Get(cls, app_type='mini'):
|
||
"""获取 access_token，优先走缓存；缓存命中失败时自动请求微信。"""
|
||
if app_type not in _VALID_TYPES:
|
||
raise RuntimeError(f'[AccessToken] 不支持的类型: {app_type}')
|
||
appid, _ = cls._GetConfig(app_type)
|
||
key = cls._CacheKey(app_type, appid)
|
||
try:
|
||
cached = cache.get(key)
|
||
if cached:
|
||
return cached
|
||
except Exception:
|
||
logger.warning('[AccessToken] 读取缓存失败，将直接请求微信', exc_info=True)
|
||
return cls.Refresh(app_type)
|
||
|
||
@classmethod
|
||
def Refresh(cls, app_type='mini'):
|
||
"""强制刷新 access_token 并写回缓存。失败时抛 RuntimeError。"""
|
||
if app_type not in _VALID_TYPES:
|
||
raise RuntimeError(f'[AccessToken] 不支持的类型: {app_type}')
|
||
appid, secret = cls._GetConfig(app_type)
|
||
params = {
|
||
'grant_type': 'client_credential',
|
||
'appid': appid,
|
||
'secret': secret,
|
||
}
|
||
try:
|
||
import requests
|
||
resp = requests.get(_TOKEN_URL, params=params, timeout=10)
|
||
data = resp.json()
|
||
except Exception as e:
|
||
logger.error('[AccessToken] 请求微信失败: %s', e, exc_info=True)
|
||
raise RuntimeError(f'[AccessToken] 请求微信失败: {e}')
|
||
if 'access_token' not in data:
|
||
errcode = data.get('errcode')
|
||
errmsg = data.get('errmsg', 'unknown')
|
||
logger.error('[AccessToken] 微信返回错误 errcode=%s errmsg=%s', errcode, errmsg)
|
||
raise RuntimeError(f'[AccessToken] 微信返回错误: {errcode} {errmsg}')
|
||
token = data['access_token']
|
||
try:
|
||
cache.set(cls._CacheKey(app_type, appid), token, _CACHE_TTL)
|
||
except Exception:
|
||
logger.warning('[AccessToken] 写入缓存失败', exc_info=True)
|
||
return token
|