gjm
3790821ce0
- sql/schema.sql: users/authorizations/auth_scenes/usage_logs + sessions 建表 - db.py: aiomysql 连接池，lifespan 内初始化与释放 - wechat_api.py: access_token 缓存 + 临时二维码创建 - auth.py: /auth/create_scene、/auth/status，handle_scan 事务内幂等处理扫码 - wechat.py: 接入 DB 生命周期，处理 subscribe/SCAN 事件；移除多余的 openid query 参数 - 首次关注赠送 7 天免费授权，has_claimed_free 条件更新保证幂等 - config.py/.env.example: 新增 FREE_AUTH_DAYS/SCENE_TTL_SECONDS/SESSION_TTL_HOURS
78 lines
2.5 KiB
Python
78 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)}"
|