docs: add WeChat login implementation plan for mobile-h5

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-08-10 16:22:07 +08:00
1 parent 3533c42291
commit 89d9471f37
1 file changed
+543
@@ -0,0 +1,543 @@
# mobile-h5 微信授权登录 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 启用登录页微信授权:openid 已绑定直接换 JWT;未绑定用手机号匹配员工并绑定;密码登录成功后补绑。
**Architecture:** `sys_user.wx_openid` + 后端 `wechat_mini_auth` 调微信 `jscode2session` / 手机号接口;`POST /api/mini/auth/wechat-login` 与 `wechat-bind`;前端 `getPhoneNumber` 按钮 + `uni.login`;复用现有 JWT 签发逻辑。
**Tech Stack:** FastAPI、SQLAlchemy、MySQL、uni-app、微信小程序 API;配置复用 `WXPAY_MINI_APPID` / `WXPAY_SECRET`
**Spec:** `docs/superpowers/specs/2026-08-10-mobile-h5-wechat-login-design.md`
---
## File map
| 路径 | 职责 |
|------|------|
| `docs/sql/alter-sys-user-wx-openid.sql` | 增列 + 唯一索引 |
| `backend/app/models/models.py` | `SysUser.wx_openid` |
| `backend/app/services/wechat_mini_auth.py` | code2session、get_phone_number、normalize_phone |
| `backend/app/services/issue_user_tokens.py` | 从 SysUser 签发 token 包(抽自 public 登录) |
| `backend/app/api/mini_auth.py` | wechat-login / wechat-bind 路由 |
| `backend/app/api/__init__.py` + `main.py` | 注册路由 |
| `backend/app/core/config.py` | EXCLUDE 增加 mini wechat 路径 |
| `backend/app/crud/user.py` | `get_by_wx_openid` / `list_by_phone_no` |
| `backend/tests/test_wechat_mini_auth_phone.py` | 手机号规范化单测 |
| `mobile-h5/utils/auth.js` | `wechatLogin` / `bindWechatAfterPassword` |
| `mobile-h5/utils/request.js` | 可选:带 Authorization 的 post |
| `mobile-h5/pages/login/index.vue` | 启用微信按钮 |
| `mobile-h5/styles/components/login.scss` | button 重置样式 |
| `mobile-h5/manifest.json` | appid(已配可核对) |
**不修改:** 支付逻辑、web-admin 登录、静默冷启动微信登录。
---
### Task 1: SQL + ORM `wx_openid`
**Files:**
- Create: `docs/sql/alter-sys-user-wx-openid.sql`
- Modify: `backend/app/models/models.py`(`SysUser` 类)
- [ ] **Step 1: 写 SQL**
```sql
-- docs/sql/alter-sys-user-wx-openid.sql
ALTER TABLE `sys_user`
ADD COLUMN `wx_openid` varchar(64) DEFAULT NULL COMMENT '微信小程序 openid' AFTER `phone_no`;
ALTER TABLE `sys_user`
ADD UNIQUE INDEX `uk_sys_user_wx_openid` (`wx_openid`);
```
- [ ] **Step 2: ORM 增加字段**
在 `SysUser` 的 `phone_no` 后增加:
```python
wx_openid = Column(String(64), nullable=True, comment="微信小程序 openid")
```
- [ ] **Step 3: 在目标库执行 SQL**(本地/测试库)
```bash
# 使用项目实际 mysql 客户端或图形工具执行 alter 脚本
```
Expected: 列存在;多行 `NULL` 的 `wx_openid` 可共存。
- [ ] **Step 4: Commit**
```bash
git add docs/sql/alter-sys-user-wx-openid.sql backend/app/models/models.py
git commit -m "$(cat <<'EOF'
feat(db): add sys_user.wx_openid for WeChat mini login
EOF
)"
```
---
### Task 2: 手机号规范化 + 微信 HTTP 服务(可单测)
**Files:**
- Create: `backend/app/services/wechat_mini_auth.py`
- Create: `backend/tests/test_wechat_mini_auth_phone.py`
- [ ] **Step 1: 写失败单测**
```python
# backend/tests/test_wechat_mini_auth_phone.py
from app.services.wechat_mini_auth import normalize_cn_mobile
def test_normalize_strips_plus_86():
assert normalize_cn_mobile("+8613800138000") == "13800138000"
def test_normalize_strips_spaces():
assert normalize_cn_mobile(" 138 0013 8000 ") == "13800138000"
def test_normalize_empty():
assert normalize_cn_mobile("") == ""
assert normalize_cn_mobile(None) == ""
```
- [ ] **Step 2: 跑测确认失败**
```bash
cd backend && source venv/bin/activate && pytest tests/test_wechat_mini_auth_phone.py -v
```
Expected: FAIL import / not found
- [ ] **Step 3: 实现服务**
```python
# backend/app/services/wechat_mini_auth.py
from __future__ import annotations
import logging
import re
from typing import Any, Dict, Optional
import httpx
from app.core.config import settings
logger = logging.getLogger(__name__)
_WX_CODE2SESSION = "https://api.weixin.qq.com/sns/jscode2session"
_WX_TOKEN = "https://api.weixin.qq.com/cgi-bin/token"
_WX_PHONE = "https://api.weixin.qq.com/wxa/business/getuserphonenumber"
class WechatMiniAuthError(Exception):
def __init__(self, message: str, *, wx_errcode: Optional[int] = None):
super().__init__(message)
self.wx_errcode = wx_errcode
def normalize_cn_mobile(raw: Optional[str]) -> str:
if raw is None:
return ""
s = re.sub(r"\s+", "", str(raw).strip())
if s.startswith("+86"):
s = s[3:]
elif s.startswith("86") and len(s) == 13:
s = s[2:]
return s
def _require_mini_credentials() -> tuple:
appid = (settings.WXPAY_MINI_APPID or "").strip()
secret = (settings.WXPAY_SECRET or "").strip()
if not appid or not secret:
raise WechatMiniAuthError("未配置小程序凭证(WXPAY_MINI_APPID / WXPAY_SECRET)")
return appid, secret
def code2session(login_code: str) -> Dict[str, Any]:
code = (login_code or "").strip()
if not code:
raise WechatMiniAuthError("loginCode 不能为空")
appid, secret = _require_mini_credentials()
with httpx.Client(timeout=15.0) as client:
r = client.get(
_WX_CODE2SESSION,
params={
"appid": appid,
"secret": secret,
"js_code": code,
"grant_type": "authorization_code",
},
)
r.raise_for_status()
data = r.json()
if data.get("errcode"):
logger.warning("code2session failed: %s", data)
raise WechatMiniAuthError("微信登录凭证无效或已过期", wx_errcode=data.get("errcode"))
openid = data.get("openid")
if not openid:
raise WechatMiniAuthError("微信未返回 openid")
return {
"openid": openid,
"session_key": data.get("session_key"),
"unionid": data.get("unionid"),
}
def _get_access_token() -> str:
appid, secret = _require_mini_credentials()
with httpx.Client(timeout=15.0) as client:
r = client.get(
_WX_TOKEN,
params={"grant_type": "client_credential", "appid": appid, "secret": secret},
)
r.raise_for_status()
data = r.json()
token = data.get("access_token")
if not token:
logger.warning("get access_token failed: %s", data)
raise WechatMiniAuthError("获取微信 access_token 失败", wx_errcode=data.get("errcode"))
return token
def get_phone_number(phone_code: str) -> str:
code = (phone_code or "").strip()
if not code:
raise WechatMiniAuthError("phoneCode 不能为空")
token = _get_access_token()
with httpx.Client(timeout=15.0) as client:
r = client.post(
f"{_WX_PHONE}?access_token={token}",
json={"code": code},
)
r.raise_for_status()
data = r.json()
if data.get("errcode") not in (0, None):
logger.warning("getuserphonenumber failed: %s", data)
raise WechatMiniAuthError("获取手机号失败,请改用账号密码登录", wx_errcode=data.get("errcode"))
info = data.get("phone_info") or {}
phone = normalize_cn_mobile(info.get("purePhoneNumber") or info.get("phoneNumber"))
if not phone:
raise WechatMiniAuthError("微信未返回手机号")
return phone
```
确认 `httpx` 已在 `requirements.txt`;若无则加入 `httpx`。
- [ ] **Step 4: 跑测通过**
```bash
cd backend && source venv/bin/activate && pytest tests/test_wechat_mini_auth_phone.py -v
```
Expected: PASS
- [ ] **Step 5: Commit**
```bash
git add backend/app/services/wechat_mini_auth.py backend/tests/test_wechat_mini_auth_phone.py backend/requirements.txt
git commit -m "$(cat <<'EOF'
feat(backend): add WeChat mini code2session and phone helpers
EOF
)"
```
---
### Task 3: 用户查询 + 签发 token 辅助
**Files:**
- Modify: `backend/app/crud/user.py`
- Create: `backend/app/services/issue_user_tokens.py`
- [ ] **Step 1: crud 增加查询**
在 `user_crud` 同类或模块级增加:
```python
def get_by_wx_openid(self, db: Session, wx_openid: str) -> Optional[SysUser]:
oid = (wx_openid or "").strip()
if not oid:
return None
return (
db.query(SysUser)
.filter(
SysUser.wx_openid == oid,
SysUser.del_flag == 0,
)
.first()
)
def list_active_by_phone_no(self, db: Session, phone_no: str) -> list:
phone = (phone_no or "").strip()
if not phone:
return []
return (
db.query(SysUser)
.filter(
SysUser.phone_no == phone,
SysUser.del_flag == 0,
SysUser.status == "1",
)
.all()
)
```
(若 `UserCRUD` 类名不同,挂到现有 class 上。)
- [ ] **Step 2: `issue_user_tokens.py`**
抽取与 `public.web_login` 相同的签发逻辑,入参 `user: SysUser`,返回 dict:`token/refreshToken/expiresIn/userInfo`。内部调用 `user_has_active_roles`、停用检查可在调用方先做,或本函数内统一:
```python
async def issue_tokens_for_user(db: Session, user: SysUser) -> dict:
# 校验 status、roles;否则 raise ValueError 带中文 message
# 复制 public.py web_login 中 session/jwt 创建与 _build_user_info_payload
...
```
为避免循环导入,可将 `_build_user_info_payload` / `_jwt_user_payload` 小幅复制到本模块,或从 `public` 导入纯函数(若已模块级)。优先:在 `issue_user_tokens.py` 内复制最小必要 payload 构建,与 public 字段保持一致。
- [ ] **Step 3: Commit**
```bash
git add backend/app/crud/user.py backend/app/services/issue_user_tokens.py
git commit -m "$(cat <<'EOF'
feat(backend): add wx openid/phone user lookups and token issuer
EOF
)"
```
---
### Task 4: mini auth 路由
**Files:**
- Create: `backend/app/api/mini_auth.py`
- Modify: `backend/app/api/__init__.py`、`backend/app/main.py`
- Modify: `backend/app/core/config.py`(EXCLUDE_PATHS)
- Modify: 本地 `backend/.env` 的 `API_ENCRYPT_EXCLUDE_PATHS` 若覆盖了默认值,追加:
`/api/mini/auth/wechat-login`
- [ ] **Step 1: 实现路由**
```python
# backend/app/api/mini_auth.py — 要点
router = APIRouter(prefix="/mini/auth", tags=["小程序鉴权"])
@router.post("/wechat-login")
async def wechat_login(body: dict, db: Session = Depends(get_db)):
login_code = (body.get("loginCode") or "").strip()
phone_code = (body.get("phoneCode") or "").strip() or None
try:
sess = code2session(login_code)
except WechatMiniAuthError as e:
return error_response(400, str(e))
openid = sess["openid"]
user = user_crud.get_by_wx_openid(db, openid)
if user:
try:
data = await issue_tokens_for_user(db, user)
except ValueError as e:
return error_response(403, str(e))
return success_response(data)
if not phone_code:
return error_response(400, "请授权手机号登录,或使用账号密码登录")
try:
phone = get_phone_number(phone_code)
except WechatMiniAuthError as e:
return error_response(400, str(e))
users = user_crud.list_active_by_phone_no(db, phone)
if not users:
return error_response(401, "未找到与该手机号匹配的员工账号,请使用账号密码登录")
if len(users) > 1:
return error_response(400, "手机号对应多个账号,请使用账号密码登录")
user = users[0]
# 冲突:openid 已被他人占用不会走到这;检查该 openid 是否写入冲突
other = user_crud.get_by_wx_openid(db, openid)
if other and other.id != user.id:
return error_response(409, "该微信已绑定其他账号")
if user.wx_openid and user.wx_openid != openid:
return error_response(409, "该账号已绑定其他微信,请联系管理员")
user.wx_openid = openid
db.add(user)
db.commit()
db.refresh(user)
try:
data = await issue_tokens_for_user(db, user)
except ValueError as e:
return error_response(403, str(e))
return success_response(data)
@router.post("/wechat-bind")
async def wechat_bind(request: Request, body: dict, db: Session = Depends(get_db)):
# Bearer 解析同 public._resolve_current_user_from_request
# code2session → 绑定到 current user
...
```
- [ ] **Step 2: 注册 router**;EXCLUDE 增加 `/api/mini/auth/wechat-login`(bind 需加密层外的 Bearer,若全局加密开启则 bind 也需按现有鉴权接口惯例处理——与 current-user 相同,通常不在 exclude,走加密+Token)。
说明:`wechat-login` 必须 exclude;`wechat-bind` 需登录,与 `/web/auth/profile` 一样走正常加密策略。若 `API_ENCRYPT_ENABLED=false`(本地常见),两者都明文即可。
- [ ] **Step 3: Commit**
```bash
git add backend/app/api/mini_auth.py backend/app/api/__init__.py backend/app/main.py backend/app/core/config.py
git commit -m "$(cat <<'EOF'
feat(backend): add mini WeChat login and bind APIs
EOF
)"
```
---
### Task 5: mobile-h5 auth + 登录页
**Files:**
- Modify: `mobile-h5/utils/request.js`(增加可选 header 传 token)
- Modify: `mobile-h5/utils/auth.js`
- Modify: `mobile-h5/pages/login/index.vue`
- Modify: `mobile-h5/styles/components/login.scss`
- Verify: `mobile-h5/manifest.json` appid
- [ ] **Step 1: auth.js 增加**
```js
export async function wechatLogin({ loginCode, phoneCode }) {
const data = await post('/mini/auth/wechat-login', {
loginCode,
phoneCode: phoneCode || undefined
})
if (!data || !data.token) throw new Error('登录失败')
saveSession({
token: data.token,
refreshToken: data.refreshToken,
userInfo: data.userInfo
})
return data
}
export async function bindWechatQuietly() {
try {
const loginRes = await new Promise((resolve, reject) => {
uni.login({ provider: 'weixin', success: resolve, fail: reject })
})
const loginCode = loginRes.code
if (!loginCode) return
const token = getToken()
if (!token) return
await post('/mini/auth/wechat-bind', { loginCode }, {
Authorization: 'Bearer ' + token
})
} catch (e) {
console.warn('wechat bind skipped', e && e.message)
}
}
```
`post` 需支持第三参 header(已有则用)。
`passwordLogin` 成功 `saveSession` 之后调用 `bindWechatQuietly()`(勿 await 阻塞 Toast,或 void 即可)。
- [ ] **Step 2: 登录页模板**
- 去掉微信 `is-disabled`。
- 协议已同意时渲染:
```html
<button
class="login-wechat"
open-type="getPhoneNumber"
@getphonenumber="onGetPhoneNumber"
>
<image class="login-wechat__icon" src="/static/wechat-logo.png" mode="aspectFit" />
<text>微信授权登录</text>
</button>
```
- 未同意时渲染普通 `view.login-wechat`,点击 → `ensureAgree` → 同意后 `agreed=true`(用户再点一次授权按钮;或同意后提示再点微信登录)。
推荐:未同意点击 view 走 ensureAgree;同意后显示 button。
- [ ] **Step 3: `onGetPhoneNumber`**
```js
async onGetPhoneNumber(e) {
if (this.loading) return
const detail = (e && e.detail) || {}
const errMsg = detail.errMsg || ''
if (errMsg && errMsg.indexOf('ok') === -1) {
uni.showToast({ title: '需要授权手机号,或请使用账号密码登录', icon: 'none' })
return
}
const phoneCode = detail.code
if (!phoneCode) {
uni.showToast({ title: '需要授权手机号,或请使用账号密码登录', icon: 'none' })
return
}
this.loading = true
try {
const loginRes = await new Promise((resolve, reject) => {
uni.login({ provider: 'weixin', success: resolve, fail: reject })
})
await wechatLogin({ loginCode: loginRes.code, phoneCode })
uni.showToast({ title: '登录成功', icon: 'success' })
setTimeout(() => this.goHome(), 400)
} catch (err) {
uni.showToast({ title: (err && err.message) || '登录失败', icon: 'none' })
} finally {
this.loading = false
}
}
```
- [ ] **Step 4: SCSS** 重置 `button.login-wechat` 默认边框/背景,保持原视觉;可保留 `.is-disabled` 供 loading。
- [ ] **Step 5: Commit**
```bash
git add mobile-h5/utils/auth.js mobile-h5/utils/request.js mobile-h5/pages/login/index.vue mobile-h5/styles/components/login.scss mobile-h5/manifest.json
git commit -m "$(cat <<'EOF'
feat(mobile-h5): enable WeChat phone+openid login on login page
EOF
)"
```
---
### Task 6: 验收
- [ ] 后端启动;确认 `.env` 中 `WXPAY_MINI_APPID` / `WXPAY_SECRET` 已填。
- [ ] 真机或开发者工具:密码登录 → 查库 `wx_openid` 有值(补绑)。
- [ ] 清会话后再点微信登录应直接成功(若 getPhoneNumber 仍弹窗,授权后仍走 openid 优先)。
- [ ] 错误密码号 / 拒绝授权:Toast 引导密码登录。
- [ ] 回复用户 `.env` 两项清单(已配置则确认即可)。
---
## Spec coverage
| Spec | Task |
|------|------|
| wx_openid 列 | 1 |
| code2session / 手机号 | 2 |
| wechat-login / bind | 4 |
| 方案 1 按钮 | 5 |
| 密码后补绑 | 5 |
| 无静默冷启动 | 默认保持 |
| EXCLUDE 路径 | 4 |
| 验收 / env 清单 | 6 |