35 lines
1.2 KiB
Python
35 lines
1.2 KiB
Python
import time
|
||
import asyncio
|
||
import httpx
|
||
from app.config import settings
|
||
|
||
# token 缓存
|
||
_token_cache: dict = {}
|
||
_lock = asyncio.Lock()
|
||
|
||
|
||
async def get_access_token() -> str:
|
||
"""获取 access_token，带缓存和并发刷新保护"""
|
||
now = int(time.time())
|
||
if _token_cache.get("access_token") and _token_cache.get("expires_at", 0) > now + 300:
|
||
return _token_cache["access_token"]
|
||
|
||
async with _lock:
|
||
# 双重检查
|
||
if _token_cache.get("access_token") and _token_cache.get("expires_at", 0) > now + 300:
|
||
return _token_cache["access_token"]
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
|
||
resp = await client.get(url, params={
|
||
"corpid": settings.corpid,
|
||
"corpsecret": settings.secret,
|
||
})
|
||
data = resp.json()
|
||
if data.get("errcode") != 0:
|
||
raise Exception(f"获取 access_token 失败: {data}")
|
||
|
||
_token_cache["access_token"] = data["access_token"]
|
||
_token_cache["expires_at"] = now + data["expires_in"]
|
||
return _token_cache["access_token"]
|