82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
import httpx
|
|
|
|
from app.config import settings
|
|
from app.services.notify_service import get_access_token
|
|
|
|
|
|
async def code2session(code: str) -> dict:
|
|
if not settings.wx_appid or not settings.wx_secret:
|
|
return {
|
|
"openid": f"dev_openid_{code or 'anonymous'}",
|
|
"unionid": None,
|
|
}
|
|
|
|
if settings.wx_appid.startswith("your_") or settings.wx_secret.startswith("your_"):
|
|
return {
|
|
"openid": f"dev_openid_{code or 'anonymous'}",
|
|
"unionid": None,
|
|
}
|
|
|
|
url = "https://api.weixin.qq.com/sns/jscode2session"
|
|
params = {
|
|
"appid": settings.wx_appid,
|
|
"secret": settings.wx_secret,
|
|
"js_code": code,
|
|
"grant_type": "authorization_code",
|
|
}
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, params=params)
|
|
return response.json()
|
|
|
|
|
|
async def generate_unlimited_qrcode(*, page: str, scene: str) -> bytes | None:
|
|
if not settings.wx_appid or not settings.wx_secret:
|
|
return None
|
|
if settings.wx_appid.startswith("your_") or settings.wx_secret.startswith("your_"):
|
|
return None
|
|
|
|
access_token = await get_access_token()
|
|
if not access_token:
|
|
return None
|
|
|
|
url = f"https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={access_token}"
|
|
payload = {
|
|
"page": page,
|
|
"scene": scene,
|
|
"check_path": False,
|
|
"env_version": settings.wx_miniprogram_env_version or "release",
|
|
}
|
|
async with httpx.AsyncClient(timeout=20.0) as client:
|
|
response = await client.post(url, json=payload)
|
|
content_type = response.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
data = response.json()
|
|
if data.get("errcode"):
|
|
return None
|
|
return response.content
|
|
|
|
|
|
async def generate_url_link(*, path: str, query: str = "", env_version: str | None = None) -> tuple[str | None, str | None]:
|
|
if not settings.wx_appid or not settings.wx_secret:
|
|
return None, "微信小程序 AppID / Secret 未配置"
|
|
if settings.wx_appid.startswith("your_") or settings.wx_secret.startswith("your_"):
|
|
return None, "微信小程序 AppID / Secret 仍为占位配置"
|
|
|
|
access_token = await get_access_token()
|
|
if not access_token:
|
|
return None, "微信 access_token 获取失败"
|
|
|
|
url = f"https://api.weixin.qq.com/wxa/generate_urllink?access_token={access_token}"
|
|
payload = {
|
|
"path": path,
|
|
"query": query,
|
|
"is_expire": False,
|
|
"env_version": env_version or settings.wx_miniprogram_env_version or "release",
|
|
}
|
|
async with httpx.AsyncClient(timeout=20.0) as client:
|
|
response = await client.post(url, json=payload)
|
|
data = response.json()
|
|
if data.get("errcode"):
|
|
return None, data.get("errmsg") or f"微信接口错误：{data.get('errcode')}"
|
|
return data.get("url_link"), None
|