75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from jose import JWTError
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.services.admin_service import get_system_configs
|
|
from app.schemas.auth import TokenRefreshRequest, WxLoginRequest, WxLoginResponse
|
|
from app.services.auth_service import login_with_wechat
|
|
from app.utils.rate_limit import client_ip, enforce_rate_limit
|
|
from app.utils.security import create_access_token, decode_access_token
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/public-config")
|
|
async def public_config(session: AsyncSession = Depends(get_db)) -> dict:
|
|
configs = await get_system_configs(session)
|
|
return {
|
|
"code": 0,
|
|
"message": "ok",
|
|
"data": {
|
|
"api_base_url": configs.get("api_base_url") or "http://127.0.0.1:8000/api/v1",
|
|
"customer_service_wechat": configs.get("customer_service_wechat") or "service_wechat_001",
|
|
},
|
|
}
|
|
|
|
|
|
@router.post("/wx-login")
|
|
async def wx_login(
|
|
payload: WxLoginRequest,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
) -> dict:
|
|
await enforce_rate_limit(
|
|
key=f"rate:login:{client_ip(request)}",
|
|
limit=10,
|
|
window_seconds=60,
|
|
message="登录过于频繁，请稍后再试",
|
|
)
|
|
try:
|
|
result = await login_with_wechat(session, payload)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
|
|
|
|
data = WxLoginResponse(**result)
|
|
return {"code": 0, "message": "ok", "data": data.model_dump()}
|
|
|
|
|
|
@router.post("/refresh")
|
|
async def refresh_token(payload: TokenRefreshRequest) -> dict:
|
|
try:
|
|
token_payload = decode_access_token(payload.refresh_token)
|
|
except JWTError as exc:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="刷新令牌无效或已过期") from exc
|
|
|
|
if token_payload.get("token_type") != "refresh":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="刷新令牌无效")
|
|
|
|
user_id = token_payload.get("user_id")
|
|
if not user_id:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="刷新令牌无效")
|
|
|
|
token = create_access_token({"user_id": user_id, "token_type": "access"})
|
|
return {
|
|
"code": 0,
|
|
"message": "ok",
|
|
"data": {
|
|
"access_token": token,
|
|
"token_type": "bearer",
|
|
"expires_in": 7200,
|
|
"user_id": int(user_id),
|
|
},
|
|
}
|