55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class WechatService:
|
|
async def login(self, code: str, phone_code: str | None = None) -> tuple[str, str | None]:
|
|
if settings.wechat_mock_login:
|
|
return f"mock-openid-{code}", "13800000000" if phone_code else None
|
|
|
|
if not settings.wechat_app_id or not settings.wechat_app_secret:
|
|
raise RuntimeError("Wechat credentials are not configured")
|
|
|
|
async with httpx.AsyncClient(timeout=8) as client:
|
|
session_resp = await client.get(
|
|
"https://api.weixin.qq.com/sns/jscode2session",
|
|
params={
|
|
"appid": settings.wechat_app_id,
|
|
"secret": settings.wechat_app_secret,
|
|
"js_code": code,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
)
|
|
session_data = session_resp.json()
|
|
openid = session_data.get("openid")
|
|
if not openid:
|
|
raise RuntimeError(session_data.get("errmsg", "Wechat login failed"))
|
|
|
|
phone_number = None
|
|
if phone_code:
|
|
access_token = await self._get_access_token(client)
|
|
phone_resp = await client.post(
|
|
f"https://api.weixin.qq.com/wxa/business/getuserphonenumber?access_token={access_token}",
|
|
json={"code": phone_code},
|
|
)
|
|
phone_data = phone_resp.json()
|
|
phone_number = phone_data.get("phone_info", {}).get("phoneNumber")
|
|
|
|
return openid, phone_number
|
|
|
|
async def _get_access_token(self, client: httpx.AsyncClient) -> str:
|
|
token_resp = await client.get(
|
|
"https://api.weixin.qq.com/cgi-bin/token",
|
|
params={
|
|
"grant_type": "client_credential",
|
|
"appid": settings.wechat_app_id,
|
|
"secret": settings.wechat_app_secret,
|
|
},
|
|
)
|
|
token_data = token_resp.json()
|
|
access_token = token_data.get("access_token")
|
|
if not access_token:
|
|
raise RuntimeError(token_data.get("errmsg", "Wechat access token failed"))
|
|
return access_token
|