77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
|
"""
|
||
|
|
微信公众平台开放接口封装
|
||
|
|
|
||
|
|
- get_access_token(): 获取并缓存 access_token（有效期 7200 秒）
|
||
|
|
- create_temp_qrcode(): 创建带字符串场景值的临时二维码，返回图片 URL
|
||
|
|
|
||
|
|
仅服务端调用，appid/secret 来自 .env。
|
||
|
|
"""
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import time
|
||
|
|
from urllib.parse import quote
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from config import WECHAT_APPID, WECHAT_SECRET
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
API_BASE = "https://api.weixin.qq.com/cgi-bin"
|
||
|
|
QRCODE_URL = "https://mp.weixin.qq.com/cgi-bin/showqrcode"
|
||
|
|
|
||
|
|
# access_token 内存缓存（单 worker 部署足够；提前 5 分钟过期避免边界失效）
|
||
|
|
_token_cache = {"value": "", "expires_at": 0.0}
|
||
|
|
|
||
|
|
|
||
|
|
async def get_access_token() -> str:
|
||
|
|
"""获取 access_token，命中缓存则直接返回"""
|
||
|
|
now = time.time()
|
||
|
|
if _token_cache["value"] and now < _token_cache["expires_at"]:
|
||
|
|
return _token_cache["value"]
|
||
|
|
|
||
|
|
if not WECHAT_APPID or not WECHAT_SECRET:
|
||
|
|
raise RuntimeError("WECHAT_APPID / WECHAT_SECRET 未配置")
|
||
|
|
|
||
|
|
params = {
|
||
|
|
"grant_type": "client_credential",
|
||
|
|
"appid": WECHAT_APPID,
|
||
|
|
"secret": WECHAT_SECRET,
|
||
|
|
}
|
||
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
||
|
|
resp = await client.get(f"{API_BASE}/token", params=params)
|
||
|
|
data = resp.json()
|
||
|
|
|
||
|
|
if "access_token" not in data:
|
||
|
|
raise RuntimeError(f"获取 access_token 失败: {data}")
|
||
|
|
|
||
|
|
_token_cache["value"] = data["access_token"]
|
||
|
|
_token_cache["expires_at"] = now + int(data.get("expires_in", 7200)) - 300
|
||
|
|
return _token_cache["value"]
|
||
|
|
|
||
|
|
|
||
|
|
async def create_temp_qrcode(scene_str: str, expire_seconds: int) -> str:
|
||
|
|
"""创建临时二维码（字符串场景值），返回扫码图片 URL"""
|
||
|
|
token = await get_access_token()
|
||
|
|
payload = {
|
||
|
|
"expire_seconds": expire_seconds,
|
||
|
|
"action_name": "QR_STR_SCENE",
|
||
|
|
"action_info": {"scene": {"scene_str": scene_str}},
|
||
|
|
}
|
||
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
||
|
|
resp = await client.post(
|
||
|
|
f"{API_BASE}/qrcode/create",
|
||
|
|
params={"access_token": token},
|
||
|
|
json=payload,
|
||
|
|
)
|
||
|
|
data = resp.json()
|
||
|
|
|
||
|
|
ticket = data.get("ticket")
|
||
|
|
if not ticket:
|
||
|
|
# 40001 等 token 失效错误：清空缓存，下次重新获取
|
||
|
|
if data.get("errcode") in (40001, 42001):
|
||
|
|
_token_cache["value"] = ""
|
||
|
|
raise RuntimeError(f"创建二维码失败: {data}")
|
||
|
|
|
||
|
|
logger.info("已创建临时二维码 scene_str=%s", scene_str)
|
||
|
|
return f"{QRCODE_URL}?ticket={quote(ticket)}"
|