99 lines
2.7 KiB
Python
99 lines
2.7 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 "",
|
||
)
|
||
|
||
|
||
@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 "",
|
||
)
|