docs/v1.0/server/changelogs/2026-01-27-wechat-service-implementation.md

微信登录服务实现

变更日期：2026-01-27
变更类型：新功能
影响范围：用户服务、认证系统

---

变更概述

实现完整的微信登录服务，支持微信公众号和开放平台两种登录方式，包括扫码登录、账号绑定、自动注册等功能。

---

主要变更

1. 数据库层

新增迁移脚本

• 文件：app/migrations/012_wechat_login_fields.py
• SQL：app/migrations/sql/012_wechat_login_fields.sql

users 表新增字段

ALTER TABLE users ADD COLUMN wechat_openid VARCHAR(64);
ALTER TABLE users ADD COLUMN wechat_unionid VARCHAR(64);
ALTER TABLE users ADD COLUMN wechat_platform SMALLINT;

-- 索引
CREATE INDEX idx_users_wechat_openid ON users(wechat_openid);
CREATE INDEX idx_users_wechat_unionid ON users(wechat_unionid);

字段说明

字段	类型	说明
wechat_openid	VARCHAR(64)	微信 OpenID（平台唯一标识）
wechat_unionid	VARCHAR(64)	微信 UnionID（跨平台唯一标识）
wechat_platform	SMALLINT	平台类型（1=mp公众号，2=open开放平台）

---

2. Model 层

WechatPlatform 枚举

class WechatPlatform(IntEnum):
    MP = 1    # 公众号
    OPEN = 2  # 开放平台
    
    @classmethod
    def from_string(cls, value: str) -> 'WechatPlatform':
        """字符串 → 枚举值"""
        
    def to_string(self) -> str:
        """枚举值 → 字符串"""
        
    @classmethod
    def get_display_name(cls, value: int) -> str:
        """获取显示名称"""

User 模型扩展

• 已在 app/models/user.py 中包含微信字段
• 使用 SMALLINT 存储平台枚举值

---

3. Schema 层

新增文件：app/schemas/wechat.py

请求 Schema：

• WechatBindRequest：绑定微信账号请求

响应 Schema：

• WechatQrcodeResponse：二维码响应
• UserInfoResponse：用户信息响应
• WechatLoginResultResponse：登录结果响应

字段命名转换

• API 层：camelCase（sceneId, qrcodeUrl）
• 数据库层：snake_case（scene_id, qrcode_url）
• 使用 Pydantic 的 alias 和 field_serializer 自动转换

---

4. Repository 层

UserRepository 新增方法

async def get_by_wechat_openid(openid: str, platform: int) -> Optional[User]
async def get_by_wechat_unionid(unionid: str) -> Optional[User]
async def update_wechat_info(user_id: UUID, openid: str, unionid: Optional[str], platform: int) -> Optional[User]
async def clear_wechat_info(user_id: UUID) -> Optional[User]

---

5. Service 层

新增文件：app/services/wechat_service.py

核心功能：

1. 生成二维码：generate_qrcode(platform)

   • 生成 UUID v7 作为 scene_id
   • 调用微信 API 生成授权 URL
   • 存储到 Redis（5分钟 TTL）

2. 处理回调：handle_callback(code, state, platform)

   • 通过 code 换取 access_token
   • 获取用户 openid、unionid、昵称、头像
   • 存储到 Redis

3. 轮询结果：get_login_result(scene_id)

   • 从 Redis 获取用户信息
   • 查找或创建用户
   • 生成 JWT Token

4. 绑定微信：bind_wechat(user_id, code, platform)

   • 验证授权码
   • 检查 openid 是否已被绑定
   • 更新用户信息

5. 解绑微信：unbind_wechat(user_id)

   • 清除微信信息

技术特点

• 异步编程：使用 AsyncSession 和 Redis
• ThreadPoolExecutor：处理同步微信 API
• 枚举转换：API 层字符串 ↔ 数据库层整数

---

6. API 层

新增文件：app/api/v1/wechat.py

端点列表：

方法	路径	说明	认证
GET	/api/v1/auth/wechat/qrcode	获取登录二维码	❌
GET	/api/v1/auth/wechat/callback	微信回调（内部）	❌
GET	/api/v1/auth/wechat/result	轮询登录结果	❌
POST	/api/v1/auth/wechat/bind	绑定微信账号	✅
DELETE	/api/v1/auth/wechat/bind	解绑微信账号	✅

统一响应格式

{
  "code": 200,
  "message": "Success",
  "data": { ... }
}

---

7. 配置层

Config 新增配置

# 微信登录配置
WECHAT_MP_APPID: str
WECHAT_MP_SECRET: str
WECHAT_MP_CALLBACK_URL: str
WECHAT_OPEN_APPID: str
WECHAT_OPEN_SECRET: str
WECHAT_OPEN_CALLBACK_URL: str

Deps 新增依赖

async def get_redis() -> AsyncGenerator[Redis, None]:
    """获取 Redis 客户端"""

---

8. 依赖管理

requirements.txt 新增

wechatpy>=1.8.18

---

9. 测试脚本

新增文件：test_wechat_service.py

测试内容：

1. 平台枚举转换
2. 生成二维码
3. 模拟微信回调
4. 轮询登录结果
5. 查询用户
6. 解绑微信
7. 清理测试数据

---

登录流程

时序图

前端 → 后端：获取二维码
后端 → Redis：存储 sceneId
后端 → 前端：返回 sceneId + qrcodeUrl

用户 → 微信：扫码授权
微信 → 后端：回调（code + state）
后端 → 微信：换取 access_token
后端 → Redis：存储用户信息

前端 → 后端：轮询结果（sceneId）
后端 → Redis：查询用户信息
后端 → 数据库：查找/创建用户
后端 → 前端：返回用户信息 + Token

详细步骤

1. 前端请求二维码

   • GET /api/v1/auth/wechat/qrcode?platform=mp
   • 返回 sceneId 和 qrcodeUrl

2. 用户扫码授权

   • 微信引导用户授权

3. 微信回调后端

   • GET /api/v1/auth/wechat/callback?code=xxx&state=sceneId
   • 后端换取 access_token
   • 存储用户信息到 Redis

4. 前端轮询结果

   • GET /api/v1/auth/wechat/result?sceneId=xxx
   • 未完成返回 {"status": "pending"}
   • 完成返回用户信息和 Token

5. 自动注册或登录

   • 通过 openid 查找用户
   • 不存在则自动注册
   • 生成 JWT Token

---

技术要点

1. 枚举值映射

数据库	API	显示名称
1	"mp"	公众号
2	"open"	开放平台

2. 数据流转

API 输入（字符串）→ 验证转换 → Service 层（整数）→ 数据库（SMALLINT）
数据库（SMALLINT）→ Model 层（整数）→ Schema 序列化 → API 输出（字符串）

3. Redis 数据结构

wechat:scene:{sceneId} → platform (整数，TTL 300s)
wechat:user:{sceneId} → JSON (用户信息，TTL 300s)

4. 异步处理

• 数据库操作：AsyncSession
• Redis 操作：redis.asyncio.Redis
• 微信 API：ThreadPoolExecutor

---

配置说明

环境变量

# 微信公众号
WECHAT_MP_APPID=wx1234567890abcdef
WECHAT_MP_SECRET=your_mp_secret
WECHAT_MP_CALLBACK_URL=https://api.jointo.ai/api/v1/auth/wechat/callback

# 微信开放平台
WECHAT_OPEN_APPID=wx0987654321fedcba
WECHAT_OPEN_SECRET=your_open_secret
WECHAT_OPEN_CALLBACK_URL=https://api.jointo.ai/api/v1/auth/wechat/callback

微信平台配置

1. 公众号：

   • 登录 微信公众平台
   • 设置授权回调域名：api.jointo.ai

2. 开放平台：

   • 登录 微信开放平台
   • 设置授权回调 URL：https://api.jointo.ai/api/v1/auth/wechat/callback

---

部署步骤

1. 安装依赖

cd server
pip install -r requirements.txt

2. 运行迁移

python app/migrations/012_wechat_login_fields.py

3. 配置环境变量

编辑 .env 文件，添加微信配置。

4. 测试服务

python test_wechat_service.py

5. 重启服务

docker-compose restart api

---

测试结果

✅ 平台枚举转换测试通过
✅ 生成二维码测试通过
✅ 轮询登录结果测试通过
✅ 查询用户测试通过
✅ 解绑微信测试通过
✅ 清理测试数据完成

---

注意事项

1. 安全性

• ✅ 使用 HTTPS 传输
• ✅ Redis 数据 5 分钟自动过期
• ✅ JWT Token 有效期控制
• ✅ 检查 openid 重复绑定

2. 数据一致性

• ✅ 无外键约束，应用层验证
• ✅ 所有关联字段创建索引
• ✅ 软删除支持（deleted_at）

3. 错误处理

• ✅ 统一异常处理
• ✅ 友好错误提示
• ✅ 日志记录

---

相关文档

• 微信登录服务设计
• 用户服务重构
• 后端技术栈规范

---

变更人：AI Assistant
审核人：待审核
部署状态：待部署