feat: 更换邀请人、操作日志、组长海报token修复
复用 xiugaijilu 补全后台操作日志;新增更换打手/管事邀请人;修复组长海报 access_token 失效;修复组长列表 BossProfile 搜索 500。
This commit is contained in:
94
utils/weixin_token.py
Normal file
94
utils/weixin_token.py
Normal file
@@ -0,0 +1,94 @@
|
||||
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
|
||||
Reference in New Issue
Block a user