Karlcarlo
1eef3ae614
- 新增 createDummyToken() 从前端构建本地 JWT（后端不验签） - LoginPage 检测 URL 参数 uid/openid/phone 直接登录 - 新增 /api/auth/params-login 后端接口（后续可选） - 移除 paramsLogin API 调用，简化流程 - Debug 面板默认显示 - 修复 JWT base64url 解码
126 lines
3.4 KiB
Python
126 lines
3.4 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
import os
|
||
|
||
from fastapi import APIRouter, HTTPException
|
||
from jose import jwt
|
||
from pydantic import BaseModel
|
||
|
||
from app.config import settings
|
||
from app.models.user import User
|
||
from app.services.wechat import code_to_session
|
||
|
||
router = APIRouter()
|
||
|
||
DEV_MODE = os.environ.get("DEV_MODE", "").lower() in ("1", "true", "yes")
|
||
# 固定测试账号 openid（dev 模式下使用）
|
||
DEV_OPENID = "dev_test_account_001"
|
||
|
||
|
||
class ExchangeCodeRequest(BaseModel):
|
||
code: str
|
||
mobile: str | None = None
|
||
|
||
|
||
class TokenResponse(BaseModel):
|
||
token: str
|
||
user_id: str
|
||
openid: str
|
||
mobile: str
|
||
|
||
|
||
def _build_token(user: User) -> str:
|
||
"""生成自包含 JWT：uid + openid + phone（兼容外部 token 格式）"""
|
||
expire = datetime.now(timezone.utc) + timedelta(days=30)
|
||
payload = {
|
||
"uid": str(user.id),
|
||
"openid": user.openid,
|
||
"phone": user.phone or "",
|
||
"exp": expire,
|
||
"iss": "localhost",
|
||
}
|
||
return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
|
||
|
||
|
||
@router.post("/exchange-code")
|
||
async def exchange_code(req: ExchangeCodeRequest) -> TokenResponse:
|
||
"""小程序登录：code → openid → 创建/查找 User → 返回自包含 JWT"""
|
||
wechat_data = await code_to_session(req.code)
|
||
openid = wechat_data["openid"]
|
||
|
||
user = await User.find_one(User.openid == openid)
|
||
if not user:
|
||
user = User(
|
||
openid=openid,
|
||
session_key=wechat_data.get("session_key"),
|
||
union_id=wechat_data.get("unionid"),
|
||
phone=req.mobile or None,
|
||
)
|
||
await user.insert()
|
||
else:
|
||
# 更新 session_key + 手机号
|
||
user.session_key = wechat_data.get("session_key")
|
||
if req.mobile:
|
||
user.phone = req.mobile
|
||
await user.save()
|
||
|
||
token = _build_token(user)
|
||
return TokenResponse(
|
||
token=token,
|
||
user_id=str(user.id),
|
||
openid=user.openid,
|
||
mobile=user.phone or "",
|
||
)
|
||
|
||
|
||
class ParamsLoginRequest(BaseModel):
|
||
uid: str = ""
|
||
openid: str
|
||
phone: str = ""
|
||
|
||
|
||
@router.post("/params-login")
|
||
async def params_login(req: ParamsLoginRequest) -> TokenResponse:
|
||
"""小程序传 uid/openid/phone → 查找或创建用户 → 返回自签 token"""
|
||
user = await User.find_one(User.openid == req.openid)
|
||
if not user:
|
||
user = User(openid=req.openid, phone=req.phone or None)
|
||
await user.insert()
|
||
else:
|
||
if req.phone and req.phone != (user.phone or ""):
|
||
user.phone = req.phone
|
||
await user.save()
|
||
|
||
token = _build_token(user)
|
||
return TokenResponse(
|
||
token=token,
|
||
user_id=str(user.id),
|
||
openid=user.openid,
|
||
mobile=user.phone or "",
|
||
)
|
||
|
||
|
||
@router.post("/dev-login")
|
||
async def dev_login() -> TokenResponse:
|
||
"""开发环境：使用固定测试账号，不依赖微信"""
|
||
if not DEV_MODE:
|
||
raise HTTPException(status_code=403, detail="仅开发环境可用")
|
||
|
||
user = await User.find_one(User.openid == DEV_OPENID)
|
||
if not user:
|
||
user = User(
|
||
openid=DEV_OPENID,
|
||
name="开发测试用户",
|
||
phone="13800000000",
|
||
)
|
||
await user.insert()
|
||
|
||
token = _build_token(user)
|
||
return TokenResponse(
|
||
token=token,
|
||
user_id=str(user.id),
|
||
openid=user.openid,
|
||
mobile=user.phone or "",
|
||
)
|