taiyi
feaf4298ca
- 后端新增 /api/auth/wechat-login 接口，用手机号自动匹配系统用户 - 登录时自动绑定 open_id，为微信订阅消息通知做好准备 - 小程序登录页替换为微信手机号授权按钮，保留密码登录为备用入口 - 新增 get_user_by_mobile 按手机号查询用户方法 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
106 lines
4.1 KiB
Python
106 lines
4.1 KiB
Python
"""
|
||
认证路由模块
|
||
|
||
职责：
|
||
处理用户认证相关接口，URL 前缀为 /api/auth。
|
||
包括：登录、获取当前用户信息、登出、绑定微信 OpenID。
|
||
"""
|
||
import json
|
||
|
||
from fastapi import APIRouter, Depends, Header
|
||
from pydantic import BaseModel
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.api.deps import get_auth_service, get_current_user
|
||
from backend.app.db import get_db_session
|
||
from backend.app.schemas.auth import LoginRequest, WechatLoginRequest
|
||
from backend.app.schemas.common import success_payload
|
||
from backend.app.services.auth_service import AuthService
|
||
from backend.app.services.wechat_notification_service import wechat_notification_service
|
||
|
||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||
|
||
|
||
class BindOpenIdRequest(BaseModel):
|
||
"""绑定微信 OpenID 请求体"""
|
||
open_id: str
|
||
|
||
|
||
@router.post("/login")
|
||
def login(
|
||
payload: LoginRequest, # 登录请求体，包含用户名、密码、角色类型
|
||
auth_service: AuthService = Depends(get_auth_service), # 注入认证服务
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
) -> dict:
|
||
"""用户登录接口
|
||
|
||
用途：验证用户名和密码，返回 JWT Token。
|
||
请求参数：LoginRequest（username、password、role_type）。
|
||
返回值：登录结果，包含 token 等认证信息。
|
||
权限要求：无需认证（公开接口）。
|
||
"""
|
||
return success_payload(auth_service.login(payload.username, payload.password, payload.role_type, session))
|
||
|
||
|
||
@router.post("/wechat-login")
|
||
def wechat_login(
|
||
payload: WechatLoginRequest, # 微信手机号登录请求体
|
||
auth_service: AuthService = Depends(get_auth_service), # 注入认证服务
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
) -> dict:
|
||
"""微信手机号一键登录接口
|
||
|
||
用途：通过微信手机号授权登录，自动匹配系统用户并绑定 open_id。
|
||
请求参数：WechatLoginRequest（code、encrypted_data、iv）。
|
||
返回值：登录结果，包含 token 等认证信息。
|
||
权限要求：无需认证（公开接口）。
|
||
"""
|
||
return success_payload(auth_service.wechat_login(payload.code, payload.phone_code, session))
|
||
|
||
|
||
@router.get("/me")
|
||
def me(
|
||
current_user: dict = Depends(get_current_user), # 从 JWT Token 解析当前用户
|
||
) -> dict:
|
||
"""获取当前登录用户信息
|
||
|
||
用途：返回当前已登录用户的详细信息（角色、权限等）。
|
||
请求参数：无（通过请求头 Authorization 传递 Token）。
|
||
返回值：当前用户信息字典。
|
||
权限要求：必须已登录（携带有效 Token）。
|
||
"""
|
||
return success_payload(current_user)
|
||
|
||
|
||
@router.post("/logout")
|
||
def logout(
|
||
authorization: str | None = Header(default=None), # 请求头中的 Authorization 字段
|
||
auth_service: AuthService = Depends(get_auth_service), # 注入认证服务
|
||
) -> dict:
|
||
"""用户登出接口
|
||
|
||
用途：使当前 Token 失效，完成登出。
|
||
请求参数：通过请求头 Authorization 传递 Bearer Token。
|
||
返回值：登出结果。
|
||
权限要求：无需严格校验，但需传递 Token。
|
||
"""
|
||
token = (authorization or "").removeprefix("Bearer").strip()
|
||
return success_payload(auth_service.logout(token))
|
||
|
||
|
||
@router.post("/bind-openid")
|
||
def bind_openid(
|
||
payload: BindOpenIdRequest, # 绑定请求体，包含微信 OpenID
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
current_user: dict = Depends(get_current_user), # 当前登录用户
|
||
) -> dict:
|
||
"""绑定微信 OpenID 接口
|
||
|
||
用途：将当前用户的账号与微信 OpenID 绑定，用于微信消息通知。
|
||
请求参数：BindOpenIdRequest（open_id - 微信用户的 OpenID）。
|
||
返回值：绑定结果 {"success": true}。
|
||
权限要求：必须已登录。
|
||
"""
|
||
wechat_notification_service.bind_open_id(session, current_user["user_id"], payload.open_id)
|
||
return success_payload({"success": True})
|