32 KiB

Raw Permalink Blame History

微信登录服务

文档版本：v2.2
最后更新：2026-01-29

服务概述

微信登录服务负责处理微信扫码登录、账号绑定等功能，支持微信公众号和开放平台两种登录方式。

职责

生成微信登录二维码
处理微信授权回调
获取微信用户信息
绑定微信账号

支持平台

公众号（mp）：适用于微信内网页
开放平台（open）：适用于网站应用

核心功能

1. 生成登录二维码

生成唯一 scene_id（UUID）
调用微信 API 生成二维码
将 scene_id 存入 Redis（有效期 5 分钟）
返回二维码 URL

2. 处理微信回调

接收微信授权 code
通过 code 换取 access_token
获取用户 openid、unionid、昵称、头像
将用户信息存入 Redis

3. 轮询登录结果

前端通过 scene_id 轮询
从 Redis 获取用户信息
通过 openid 查找或创建用户
返回 JWT Token

4. 绑定微信账号

已登录用户绑定微信
检查 openid 是否已被其他账号使用
更新用户表

数据库设计

users 表微信相关字段

-- 微信登录相关字段
ALTER TABLE users ADD COLUMN IF NOT EXISTS wechat_openid VARCHAR(64);
ALTER TABLE users ADD COLUMN IF NOT EXISTS wechat_unionid VARCHAR(64);
ALTER TABLE users ADD COLUMN IF NOT EXISTS wechat_platform SMALLINT;

-- 创建索引（重要：无外键约束时必须创建索引）
CREATE INDEX IF NOT EXISTS idx_users_wechat_openid 
    ON users(wechat_openid) WHERE wechat_openid IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_users_wechat_unionid 
    ON users(wechat_unionid) WHERE wechat_unionid IS NOT NULL;

-- 添加注释
COMMENT ON COLUMN users.wechat_openid IS '微信 OpenID（平台唯一标识）';
COMMENT ON COLUMN users.wechat_unionid IS '微信 UnionID（跨平台唯一标识）';
COMMENT ON COLUMN users.wechat_platform IS '微信平台类型（1=mp公众号，2=open开放平台）';

微信平台枚举值映射表

值	名称	说明
1	mp	公众号
2	open	开放平台

Model 定义

# app/models/user.py
from sqlmodel import SQLModel, Field, Column
from sqlalchemy import SmallInteger
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
from typing import Optional
from uuid import UUID
from datetime import datetime, timezone  # ✅ 必须导入 timezone
from enum import IntEnum
from app.utils.id_generator import generate_uuid

class WechatPlatform(IntEnum):
    """微信平台枚举"""
    MP = 1    # 公众号
    OPEN = 2  # 开放平台
    
    @classmethod
    def from_string(cls, value: str) -> 'WechatPlatform':
        """从字符串转换为枚举值"""
        mapping = {
            'mp': cls.MP,
            'open': cls.OPEN
        }
        result = mapping.get(value.lower())
        if result is None:
            raise ValueError(f"Invalid platform: {value}")
        return result
    
    def to_string(self) -> str:
        """转换为字符串"""
        mapping = {
            self.MP: 'mp',
            self.OPEN: 'open'
        }
        return mapping[self]
    
    @classmethod
    def get_display_name(cls, value: int) -> str:
        """获取显示名称"""
        names = {
            cls.MP: "公众号",
            cls.OPEN: "开放平台"
        }
        return names.get(value, "未知平台")

class User(SQLModel, table=True):
    __tablename__ = "users"
    
    user_id: UUID = Field(
        sa_column=Column(
            PG_UUID(as_uuid=True), 
            primary_key=True,
            default=generate_uuid  # ✅ 应用层生成 UUID
        )
    )
    
    # 微信登录字段
    wechat_openid: Optional[str] = Field(
        default=None, 
        max_length=64, 
        index=True,
        description="微信 OpenID"
    )
    wechat_unionid: Optional[str] = Field(
        default=None, 
        max_length=64, 
        index=True,
        description="微信 UnionID"
    )
    wechat_platform: Optional[int] = Field(
        default=None,
        sa_column=Column(SmallInteger),
        description="微信平台：1=mp(公众号), 2=open(开放平台)"
    )
    
    # 时间戳字段（使用 UTC aware datetime）
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        description="创建时间（UTC）"
    )
    updated_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        description="更新时间（UTC）"
    )
    deleted_at: Optional[datetime] = Field(
        default=None,
        description="软删除时间（UTC）"
    )

Schema 定义

请求 Schema

# app/schemas/wechat.py
from pydantic import BaseModel, Field
from typing import Literal

class WechatBindRequest(BaseModel):
    """绑定微信账号请求"""
    code: str = Field(..., description="微信授权码")
    platform: Literal["mp", "open"] = Field(..., description="平台类型")
    
    # ✅ 不在 Schema 层转换，保持字符串类型
    # 转换逻辑在 Service 层处理
    
    class Config:
        json_schema_extra = {
            "example": {
                "code": "061a2b3c4d5e6f7g8h9i0j",
                "platform": "mp"
            }
        }

响应 Schema

# app/schemas/wechat.py
from pydantic import BaseModel, Field, field_serializer
from typing import Optional
from uuid import UUID

class WechatQrcodeResponse(BaseModel):
    """微信二维码响应"""
    sceneId: str = Field(..., alias="scene_id", description="场景ID")
    qrcodeUrl: str = Field(..., alias="qrcode_url", description="二维码URL")
    expiresIn: int = Field(..., alias="expires_in", description="过期时间（秒）")
    
    class Config:
        populate_by_name = True
        json_schema_extra = {
            "example": {
                "sceneId": "550e8400-e29b-41d4-a716-446655440000",
                "qrcodeUrl": "https://open.weixin.qq.com/connect/oauth2/authorize?...",
                "expiresIn": 300
            }
        }

class UserInfoResponse(BaseModel):
    """用户信息响应"""
    userId: UUID = Field(..., alias="user_id")
    username: str
    avatarUrl: Optional[str] = Field(None, alias="avatar_url")
    wechatOpenid: Optional[str] = Field(None, alias="wechat_openid")
    wechatPlatform: Optional[int] = Field(None, alias="wechat_platform")
    
    @field_serializer('wechatPlatform')
    def serialize_platform(self, value: Optional[int]) -> Optional[str]:
        """API 输出：整数 → 字符串"""
        if value is None:
            return None
        from app.models.user import WechatPlatform
        return WechatPlatform(value).to_string()
    
    class Config:
        populate_by_name = True
        from_attributes = True

class WechatLoginResultResponse(BaseModel):
    """微信登录结果响应"""
    status: Literal["pending", "success"] = Field(..., description="登录状态")
    user: Optional[UserInfoResponse] = Field(None, description="用户信息")
    accessToken: Optional[str] = Field(None, alias="access_token", description="访问令牌")
    refreshToken: Optional[str] = Field(None, alias="refresh_token", description="刷新令牌")
    tokenType: Optional[str] = Field(None, alias="token_type", description="令牌类型")
    
    class Config:
        populate_by_name = True

Repository 层

UserRepository 扩展方法

# app/repositories/user_repository.py
from typing import Optional
from uuid import UUID
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from app.models.user import User
from app.repositories.base_repository import BaseRepository
from app.core.exceptions import NotFoundError
from datetime import datetime, timezone

class UserRepository(BaseRepository[User]):
    """用户数据访问层"""
    
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)
    
    async def get_by_wechat_openid(
        self,
        openid: str,
        platform: int
    ) -> Optional[User]:
        """根据微信 OpenID 和平台查找用户
        
        Args:
            openid: 微信 OpenID
            platform: 平台类型（整数枚举值：1=mp, 2=open）
        """
        result = await self.session.execute(
            select(User).where(
                User.wechat_openid == openid,
                User.wechat_platform == platform,
                User.deleted_at.is_(None)
            )
        )
        return result.scalar_one_or_none()
    
    async def get_by_wechat_unionid(
        self,
        unionid: str
    ) -> Optional[User]:
        """根据微信 UnionID 查找用户"""
        result = await self.session.execute(
            select(User).where(
                User.wechat_unionid == unionid,
                User.deleted_at.is_(None)
            )
        )
        return result.scalar_one_or_none()
    
    async def update_wechat_info(
        self,
        user_id: UUID,
        openid: str,
        unionid: Optional[str],
        platform: int
    ) -> User:
        """更新用户微信信息
        
        Args:
            user_id: 用户 ID
            openid: 微信 OpenID
            unionid: 微信 UnionID
            platform: 平台类型（整数枚举值：1=mp, 2=open）
        """
        user = await self.session.get(User, user_id)
        if not user:
            raise NotFoundError("用户不存在")
        
        user.wechat_openid = openid
        user.wechat_unionid = unionid
        user.wechat_platform = platform
        user.updated_at = datetime.now(timezone.utc)
        
        await self.session.commit()
        await self.session.refresh(user)
        return user
    
    async def clear_wechat_info(self, user_id: UUID) -> User:
        """清除用户微信信息"""
        user = await self.session.get(User, user_id)
        if not user:
            raise NotFoundError("用户不存在")
        
        user.wechat_openid = None
        user.wechat_unionid = None
        user.wechat_platform = None
        user.updated_at = datetime.now(timezone.utc)
        
        await self.session.commit()
        await self.session.refresh(user)
        return user

Service 层

WechatService 类

# app/services/wechat_service.py
from typing import Optional, Dict, Any
from datetime import datetime
from uuid import UUID
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor

from sqlmodel.ext.asyncio.session import AsyncSession
from redis.asyncio import Redis
from wechatpy import WeChatOAuth
from wechatpy.exceptions import WeChatOAuthException

from app.models.user import User, WechatPlatform
from app.repositories.user_repository import UserRepository
from app.core.exceptions import ValidationError, NotFoundError
from app.core.config import settings
from app.core.security import create_access_token, create_refresh_token
from app.utils.id_generator import generate_uuid

class WechatService:
    """微信登录服务（异步）"""
    
    def __init__(
        self, 
        session: AsyncSession,
        redis_client: Redis,
        executor: Optional[ThreadPoolExecutor] = None
    ):
        self.repository = UserRepository(session)
        self.session = session
        self.redis = redis_client
        self.executor = executor or ThreadPoolExecutor(max_workers=4)
    
    async def generate_qrcode(
        self,
        platform: str = 'mp'
    ) -> Dict[str, Any]:
        """生成微信登录二维码
        
        Args:
            platform: 平台类型字符串（'mp' 或 'open'）
        """
        # 验证并转换平台类型
        try:
            platform_enum = WechatPlatform.from_string(platform)
        except ValueError:
            raise ValidationError("无效的平台类型")
        
        # 生成唯一 scene_id
        scene_id = str(generate_uuid())
        
        # 获取配置
        appid, secret, redirect_uri = self._get_platform_config(platform)
        
        # 在 executor 中运行同步微信 API
        loop = asyncio.get_event_loop()
        authorize_url = await loop.run_in_executor(
            self.executor,
            self._create_authorize_url,
            appid, secret, redirect_uri, scene_id
        )
        
        # 将 scene_id 和平台枚举值存入 Redis（5分钟有效期）
        await self.redis.setex(
            f"wechat:scene:{scene_id}",
            300,
            str(platform_enum.value)  # 存储整数枚举值
        )
        
        return {
            'scene_id': scene_id,
            'qrcode_url': authorize_url,
            'expires_in': 300
        }
    
    async def handle_callback(
        self,
        code: str,
        state: str,
        platform: str
    ) -> None:
        """处理微信回调
        
        Args:
            code: 微信授权码
            state: 场景ID（sceneId）
            platform: 平台类型字符串（'mp' 或 'open'）
        """
        # 验证并转换平台类型
        try:
            platform_enum = WechatPlatform.from_string(platform)
        except ValueError:
            raise ValidationError("无效的平台类型")
        
        # 获取配置
        appid, secret, redirect_uri = self._get_platform_config(platform)
        
        try:
            # 在 executor 中运行同步微信 API
            loop = asyncio.get_event_loop()
            user_info = await loop.run_in_executor(
                self.executor,
                self._fetch_wechat_user_info,
                appid, secret, redirect_uri, code
            )
            
            # 提取关键信息（存储整数枚举值）
            wechat_data = {
                'openid': user_info['openid'],
                'unionid': user_info.get('unionid'),
                'nickname': user_info.get('nickname'),
                'avatar': user_info.get('headimgurl'),
                'platform': platform_enum.value  # 存储整数枚举值
            }
            
            # 将用户信息存入 Redis（scene_id 作为 key）
            scene_id = state  # state 参数即为 scene_id
            await self.redis.setex(
                f"wechat:user:{scene_id}",
                300,
                json.dumps(wechat_data, ensure_ascii=False)
            )
            
        except WeChatOAuthException as e:
            raise ValidationError(f"微信授权失败：{str(e)}")
    
    async def get_login_result(
        self,
        scene_id: str
    ) -> Optional[Dict[str, Any]]:
        """前端轮询获取登录结果"""
        # 从 Redis 获取用户信息
        user_data_str = await self.redis.get(f"wechat:user:{scene_id}")
        if not user_data_str:
            return {'status': 'pending'}
        
        # 解析用户信息（platform 字段已经是整数枚举值）
        wechat_data = json.loads(user_data_str)
        
        # 通过 openid 查找用户（platform 是整数）
        user = await self.repository.get_by_wechat_openid(
            wechat_data['openid'],
            wechat_data['platform']  # 整数枚举值
        )
        
        # 用户不存在，自动注册
        if not user:
            user = await self._create_user_from_wechat(wechat_data)
        
        # 生成 Token
        access_token = create_access_token(data={'sub': str(user.user_id)})
        refresh_token = create_refresh_token(data={'sub': str(user.user_id)})
        
        # 删除 Redis 中的临时数据
        await self.redis.delete(f"wechat:user:{scene_id}")
        await self.redis.delete(f"wechat:scene:{scene_id}")
        
        return {
            'status': 'success',
            'user': user,
            'access_token': access_token,
            'refresh_token': refresh_token,
            'token_type': 'bearer'
        }
    
    async def bind_wechat(
        self,
        user_id: UUID,
        code: str,
        platform: str
    ) -> User:
        """绑定微信账号
        
        Args:
            user_id: 用户 ID
            code: 微信授权码
            platform: 平台类型字符串（'mp' 或 'open'）
        """
        # 验证并转换平台类型
        try:
            platform_enum = WechatPlatform.from_string(platform)
        except ValueError:
            raise ValidationError("无效的平台类型")
        
        # 获取配置
        appid, secret, redirect_uri = self._get_platform_config(platform)
        
        try:
            # 在 executor 中运行同步微信 API
            loop = asyncio.get_event_loop()
            user_info = await loop.run_in_executor(
                self.executor,
                self._fetch_wechat_user_info,
                appid, secret, redirect_uri, code
            )
            
            openid = user_info['openid']
            unionid = user_info.get('unionid')
            
            # 检查 openid 是否已被其他账号绑定（使用整数枚举值）
            existing_user = await self.repository.get_by_wechat_openid(
                openid, 
                platform_enum.value
            )
            if existing_user and existing_user.user_id != user_id:
                raise ValidationError("该微信账号已被其他用户绑定")
            
            # 更新用户信息（使用整数枚举值）
            return await self.repository.update_wechat_info(
                user_id, openid, unionid, platform_enum.value
            )
            
        except WeChatOAuthException as e:
            raise ValidationError(f"微信授权失败：{str(e)}")
    
    async def unbind_wechat(self, user_id: UUID) -> User:
        """解绑微信账号"""
        return await self.repository.clear_wechat_info(user_id)
    
    # ==================== 私有方法 ====================
    
    def _get_platform_config(self, platform: str) -> tuple:
        """获取平台配置"""
        if platform == 'mp':
            return (
                settings.WECHAT_MP_APPID,
                settings.WECHAT_MP_SECRET,
                settings.WECHAT_MP_CALLBACK_URL
            )
        else:  # open
            return (
                settings.WECHAT_OPEN_APPID,
                settings.WECHAT_OPEN_SECRET,
                settings.WECHAT_OPEN_CALLBACK_URL
            )
    
    def _create_authorize_url(
        self,
        appid: str,
        secret: str,
        redirect_uri: str,
        state: str
    ) -> str:
        """创建授权 URL（同步方法，在 executor 中运行）"""
        oauth = WeChatOAuth(
            app_id=appid,
            secret=secret,
            redirect_uri=redirect_uri,
            scope='snsapi_userinfo',
            state=state
        )
        return oauth.authorize_url
    
    def _fetch_wechat_user_info(
        self,
        appid: str,
        secret: str,
        redirect_uri: str,
        code: str
    ) -> Dict[str, Any]:
        """获取微信用户信息（同步方法，在 executor 中运行）"""
        oauth = WeChatOAuth(
            app_id=appid,
            secret=secret,
            redirect_uri=redirect_uri
        )
        oauth.fetch_access_token(code)
        return oauth.get_user_info()
    
    async def _create_user_from_wechat(
        self,
        wechat_data: Dict[str, Any]
    ) -> User:
        """从微信信息创建用户"""
        # 生成用户名（时间戳 + 随机后缀）
        timestamp = int(datetime.now(timezone.utc).timestamp())
        random_suffix = str(generate_uuid())[:8]
        username = f"user_{timestamp}_{random_suffix}"
        
        # 使用事务创建用户
        async with self.session.begin():
            user = User(
                user_id=generate_uuid(),
                username=username,
                wechat_openid=wechat_data['openid'],
                wechat_unionid=wechat_data.get('unionid'),
                wechat_platform=wechat_data['platform'],
                avatar_url=wechat_data.get('avatar'),
                subscription_tier='free',
                ai_credits_remaining=10,
                ai_credits_total=10,
                username_changed=False
            )
            
            self.session.add(user)
            await self.session.flush()  # ✅ 使用 flush 获取 ID
        
        await self.session.refresh(user)
        return user

API 接口

1. 获取微信登录二维码

GET /api/v1/auth/wechat/qrcode?platform=mp

查询参数：

参数	类型	必填	说明
platform	string	是	平台类型：mp=公众号，open=开放平台

成功响应 (200 OK)：

{
  "success": true,
  "code": 200,
  "message": "Success",
  "data": {
    "sceneId": "550e8400-e29b-41d4-a716-446655440000",
    "qrcodeUrl": "https://open.weixin.qq.com/connect/oauth2/authorize?...",
    "expiresIn": 300
  },
  "timestamp": "2026-01-29T08:00:00Z"
}

错误响应 (400 Bad Request)：

{
  "success": false,
  "code": 400,
  "message": "无效的平台类型",
  "data": null,
  "timestamp": "2026-01-29T08:00:00Z"
}

2. 微信回调接口（内部）

GET /api/v1/auth/wechat/callback?code=xxx&state=xxx

查询参数：

参数	类型	必填	说明
code	string	是	微信授权码
state	string	是	场景ID（sceneId）

说明：此接口由微信服务器回调，不对外暴露。成功后重定向到前端页面。

3. 轮询登录结果

GET /api/v1/auth/wechat/result?sceneId=xxx

查询参数：

参数	类型	必填	说明
sceneId	string	是	场景ID

响应（未完成） (200 OK)：

{
  "success": true,
  "code": 200,
  "message": "Success",
  "data": {
    "status": "pending"
  },
  "timestamp": "2026-01-29T08:00:00Z"
}

响应（已完成） (200 OK)：

{
  "success": true,
  "code": 200,
  "message": "Success",
  "data": {
    "status": "success",
    "user": {
      "userId": "550e8400-e29b-41d4-a716-446655440000",
      "username": "user_1738012345_a3f9",
      "avatarUrl": "https://wx.qlogo.cn/...",
      "wechatOpenid": "oxxxxxxxxxxxxxx",
      "wechatPlatform": "mp"
    },
    "accessToken": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
    "refreshToken": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
    "tokenType": "bearer"
  },
  "timestamp": "2026-01-29T08:00:00Z"
}

错误响应 (404 Not Found)：

{
  "success": false,
  "code": 404,
  "message": "场景ID不存在或已过期",
  "data": null,
  "timestamp": "2026-01-29T08:00:00Z"
}

4. 绑定微信账号

POST /api/v1/users/me/bind/wechat
Authorization: Bearer <access_token>
Content-Type: application/json

请求体：

{
  "code": "061a2b3c4d5e6f7g8h9i0j",
  "platform": "mp"
}

成功响应 (200 OK)：

{
  "success": true,
  "code": 200,
  "message": "绑定成功",
  "data": {
    "userId": "550e8400-e29b-41d4-a716-446655440000",
    "username": "user_1738012345_a3f9",
    "wechatOpenid": "oxxxxxxxxxxxxxx",
    "wechatUnionid": "oxxxxxxxxxxxxxx",
    "wechatPlatform": "mp"
  },
  "timestamp": "2026-01-29T08:00:00Z"
}

错误响应 (400 Bad Request)：

{
  "success": false,
  "code": 400,
  "message": "该微信账号已被其他用户绑定",
  "data": null,
  "timestamp": "2026-01-29T08:00:00Z"
}

错误响应 (401 Unauthorized)：

{
  "success": false,
  "code": 401,
  "message": "未认证或 Token 无效",
  "data": null,
  "timestamp": "2026-01-29T08:00:00Z"
}

5. 解绑微信账号

DELETE /api/v1/users/me/bind/wechat
Authorization: Bearer <access_token>

成功响应 (200 OK)：

{
  "success": true,
  "code": 200,
  "message": "解绑成功",
  "data": {
    "userId": "550e8400-e29b-41d4-a716-446655440000",
    "username": "user_1738012345_a3f9",
    "wechatOpenid": null,
    "wechatUnionid": null,
    "wechatPlatform": null
  },
  "timestamp": "2026-01-29T08:00:00Z"
}

错误响应 (401 Unauthorized)：

{
  "success": false,
  "code": 401,
  "message": "未认证或 Token 无效",
  "data": null,
  "timestamp": "2026-01-29T08:00:00Z"
}

错误响应 (404 Not Found)：

{
  "success": false,
  "code": 404,
  "message": "用户不存在",
  "data": null,
  "timestamp": "2026-01-29T08:00:00Z"
}

登录流程

时序图

前端                    后端                    微信服务器              Redis
 |                       |                          |                      |
 |--1. 获取二维码-------->|                          |                      |
 |                       |--生成 sceneId----------->|                      |
 |                       |--存储 sceneId----------->|                      |
 |<--sceneId, qrcodeUrl--|                          |                      |
 |                       |                          |                      |
 |--2. 展示二维码-------->用户扫码                   |                      |
 |                       |                          |                      |
 |                       |<--3. 授权回调（code）-----|                      |
 |                       |--4. 换取 access_token---->|                      |
 |                       |<--用户信息----------------|                      |
 |                       |--5. 存入 Redis（JSON）--->|                      |
 |                       |                          |                      |
 |--6. 轮询登录结果------>|                          |                      |
 |                       |--查询 Redis------------->|                      |
 |<--pending-------------|<--无数据-----------------|                      |
 |                       |                          |                      |
 |--7. 再次轮询---------->|                          |                      |
 |                       |--查询 Redis------------->|                      |
 |                       |<--用户数据（JSON）--------|                      |
 |                       |--查找/创建用户---------->|                      |
 |                       |--生成 Token------------->|                      |
 |                       |--删除临时数据----------->|                      |
 |<--success, token------|                          |                      |

详细步骤

1. 前端请求二维码

请求：GET /api/v1/auth/wechat/qrcode?platform=mp
后端生成 UUID v7 作为 sceneId
后端调用微信 API 生成授权 URL
后端将 sceneId 存入 Redis（key: wechat:scene:{sceneId}, value: platform, TTL: 300s）
返回 sceneId 和 qrcodeUrl

2. 用户扫码授权

用户使用微信扫描二维码
微信引导用户授权（获取昵称、头像等信息）

3. 微信回调后端

微信服务器回调：GET /api/v1/auth/wechat/callback?code=xxx&state=sceneId
后端通过 code 换取 access_token（在 ThreadPoolExecutor 中运行）
获取用户 openid、unionid、昵称、头像
将用户信息序列化为 JSON 存入 Redis（key: wechat:user:{sceneId}, TTL: 300s）

4. 前端轮询结果

前端每隔 2 秒轮询：GET /api/v1/auth/wechat/result?sceneId=xxx
未完成返回：{"code": 200, "data": {"status": "pending"}}
完成后返回用户信息和 Token

5. 自动注册或登录

后端通过 openid 查找用户（get_by_wechat_openid）
用户不存在则自动注册：
- 生成用户名：user_{timestamp}_{random}
- 创建用户记录（包含微信信息）
- 初始化积分：10 积分
生成 JWT Token（access_token 和 refresh_token）
删除 Redis 中的临时数据
返回用户信息和 Token

6. 前端存储 Token

前端接收 Token 并存储到 localStorage
跳转到主页面

配置说明

环境变量

在 server/.env 文件中配置：

# 微信公众号
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

# Redis（异步连接）
REDIS_URL=redis://localhost:6379/0

Config 类定义

在 app/core/config.py 中添加配置：

# app/core/config.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # 微信公众号配置
    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
    
    # Redis 配置
    REDIS_URL: str = "redis://localhost:6379/0"
    
    class Config:
        env_file = ".env"
        case_sensitive = True
        extra = "ignore"  # ✅ 忽略额外字段

settings = Settings()

依赖注入配置

在 app/api/deps.py 中添加 Redis 依赖：

# app/api/deps.py
from typing import AsyncGenerator
from redis.asyncio import Redis
from app.core.config import settings

async def get_redis() -> AsyncGenerator[Redis, None]:
    """获取 Redis 客户端"""
    redis = Redis.from_url(
        settings.REDIS_URL,
        encoding="utf-8",
        decode_responses=True
    )
    try:
        yield redis
    finally:
        await redis.close()

微信公众号配置

登录微信公众平台
开发 -> 接口权限 -> 网页授权
设置授权回调域名：api.jointo.ai（不含协议和路径）

微信开放平台配置

登录微信开放平台
管理中心 -> 网站应用 -> 创建应用
设置授权回调域名：https://api.jointo.ai/api/v1/auth/wechat/callback（完整 URL）

依赖安装

# 安装异步 Redis 客户端
pip install redis[hiredis]>=5.0.1

# 安装微信 SDK
pip install wechatpy>=1.8.18

技术要点

1. 异步编程

使用 AsyncSession 进行数据库操作
使用 redis.asyncio.Redis 进行缓存操作
使用 ThreadPoolExecutor 运行同步微信 API

2. 数据序列化

使用 json.dumps() 和 json.loads() 序列化 Redis 数据
避免使用 str() 和 ast.literal_eval()

3. 字段命名

API 层使用 camelCase（sceneId, qrcodeUrl）
数据库层使用 snake_case（scene_id, qrcode_url）
使用 Pydantic 的 alias 进行转换

4. 响应格式

所有 API 响应使用统一包装格式
包含 code, message, data 三个字段
错误响应同样遵循此格式

5. 无外键约束

数据库不使用外键约束
在 Service 层验证引用完整性
所有关联字段必须创建索引

变更记录

v2.2 (2026-01-29)

✅ 修复 Model 层时间戳定义，添加 timezone 导入
✅ 修复 Schema 层枚举转换逻辑，移除不必要的 validator
✅ 添加 field_serializer 装饰器到响应 Schema
✅ Repository 层继承 BaseRepository
✅ Service 层使用事务处理（async with self.session.begin()）
✅ 统一 API 响应格式（添加 success 和 timestamp 字段）
✅ 完善配置类定义（添加 extra = "ignore"）
✅ 符合 jointo-tech-stack 技术栈规范

v2.1 (2026-01-27)

初始版本

文档版本：v2.2
最后更新：2026-01-29

32 KiB Raw Permalink Blame History

微信登录服务

目录

服务概述

职责

支持平台

核心功能

1. 生成登录二维码

2. 处理微信回调

3. 轮询登录结果

4. 绑定微信账号

数据库设计

users 表微信相关字段

微信平台枚举值映射表

Model 定义

Schema 定义

请求 Schema

响应 Schema

Repository 层

UserRepository 扩展方法

Service 层

WechatService 类

API 接口

1. 获取微信登录二维码

2. 微信回调接口（内部）

3. 轮询登录结果

4. 绑定微信账号

5. 解绑微信账号

登录流程

时序图

详细步骤

1. 前端请求二维码

2. 用户扫码授权

3. 微信回调后端

4. 前端轮询结果

5. 自动注册或登录

6. 前端存储 Token

配置说明

环境变量

Config 类定义

依赖注入配置

微信公众号配置

微信开放平台配置

依赖安装

技术要点

1. 异步编程

2. 数据序列化

3. 字段命名

4. 响应格式

5. 无外键约束

相关文档

变更记录

v2.2 (2026-01-29)

v2.1 (2026-01-27)

32 KiB

Raw Permalink Blame History