| 123456789101112131415161718192021222324252627282930313233343536373839 |
- from fastapi import APIRouter, HTTPException
- from pydantic import BaseModel
- import httpx
- from app.core.security import create_access_token
- from app.core.config import get_settings
- settings = get_settings()
- router = APIRouter(prefix="/auth", tags=["鉴权"])
- class WechatLoginRequest(BaseModel):
- code: str
- class LoginResponse(BaseModel):
- access_token: str
- token_type: str = "bearer"
- user_id: str
- @router.post("/login/wechat", response_model=LoginResponse)
- async def wechat_login(req: WechatLoginRequest):
- url = (
- f"https://api.weixin.qq.com/sns/jscode2session"
- f"?appid={settings.wechat_appid}"
- f"&secret={settings.wechat_secret}"
- f"&js_code={req.code}"
- f"&grant_type=authorization_code"
- )
- async with httpx.AsyncClient() as client:
- resp = await client.get(url)
- data = resp.json()
- if "errcode" in data:
- raise HTTPException(status_code=400, detail=f"Wechat login failed: {data.get('errmsg')}")
- openid = data["openid"]
- access_token = create_access_token(data={"sub": openid})
- return LoginResponse(access_token=access_token, user_id=openid)
|
