marco
7fd70a264d
- 服务号网页授权(WX_MP_APPID/SECRET,区别于 App 的 WECHAT_APP_ID):
落地页 base 静默拿 openid;点「领券」补 userinfo 拿昵称头像(交互触发避快照页)
- 新表 cps_wx_user + cps_click 加 openid + 迁移;integrations/wx_oauth
- /c/{code} 授权链路 + 回调 /wx/oauth/cb + cookie 免重复授权;click 带 openid
- admin 群详情加「群内微信用户」(领券画像)端点
- 下单归因到人需 user-level sid,本期只做身份+领券侧
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""微信服务号网页授权:构造授权链接 + code 换 openid + 拉 userinfo。
|
|
|
|
用于 CPS 落地页(coupon.shaguabijia.com)在微信内拿用户身份。流程(见微信文档):
|
|
授权链接(snsapi_base|snsapi_userinfo) → 回调带 code → sns/oauth2/access_token 换 openid
|
|
→ (userinfo 时)sns/userinfo 拿昵称头像。
|
|
|
|
注意:
|
|
- 这里的 access_token 是【网页授权 token】,与基础 access_token(cgi-bin/token)不同,
|
|
走 sns/* 接口,不需要 IP 白名单。
|
|
- secret 只在服务器,不下发客户端。
|
|
- 用 WX_MP_APPID/SECRET(已认证服务号),区别于 WECHAT_APP_ID(App 移动应用)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
_API = "https://api.weixin.qq.com"
|
|
_AUTHORIZE = "https://open.weixin.qq.com/connect/oauth2/authorize"
|
|
_TIMEOUT = 10
|
|
|
|
|
|
class WxOauthError(Exception):
|
|
"""网页授权调用失败(凭证/网络/code 失效等)。调用方兜底为无 openid。"""
|
|
|
|
|
|
def build_authorize_url(redirect_uri: str, scope: str, state: str) -> str:
|
|
"""构造网页授权跳转链接。scope: 'snsapi_base'(静默,仅 openid) | 'snsapi_userinfo'(昵称头像)。
|
|
|
|
微信要求 redirect_uri urlEncode、参数顺序固定、结尾 #wechat_redirect。
|
|
"""
|
|
return (
|
|
f"{_AUTHORIZE}?appid={settings.WX_MP_APPID}"
|
|
f"&redirect_uri={quote(redirect_uri, safe='')}"
|
|
f"&response_type=code&scope={scope}&state={quote(state, safe='')}"
|
|
f"#wechat_redirect"
|
|
)
|
|
|
|
|
|
def exchange_code(code: str) -> dict:
|
|
"""code 换网页授权 access_token + openid。返回含 openid/access_token/scope(/unionid)。
|
|
|
|
code 5 分钟内一次性有效。失败(errcode)抛 WxOauthError。
|
|
"""
|
|
resp = httpx.get(
|
|
f"{_API}/sns/oauth2/access_token",
|
|
params={
|
|
"appid": settings.WX_MP_APPID,
|
|
"secret": settings.WX_MP_SECRET,
|
|
"code": code,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
timeout=_TIMEOUT,
|
|
)
|
|
data = resp.json()
|
|
if "openid" not in data:
|
|
raise WxOauthError(f"exchange_code failed: {data}")
|
|
return data
|
|
|
|
|
|
def get_userinfo(access_token: str, openid: str) -> dict:
|
|
"""拉用户信息(nickname/headimgurl/unionid)。需 scope=snsapi_userinfo 的 access_token。"""
|
|
resp = httpx.get(
|
|
f"{_API}/sns/userinfo",
|
|
params={"access_token": access_token, "openid": openid, "lang": "zh_CN"},
|
|
timeout=_TIMEOUT,
|
|
)
|
|
resp.encoding = "utf-8" # 昵称含中文/emoji,强制 utf-8 避免乱码
|
|
data = resp.json()
|
|
if "openid" not in data:
|
|
raise WxOauthError(f"get_userinfo failed: {data}")
|
|
return data
|