119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
from fastapi import APIRouter, Header
|
||
|
||
from app.dependencies import CurrentUser, DbSession, RedisClient
|
||
from app.schemas.auth import (
|
||
HeartbeatRequest,
|
||
HeartbeatResponse,
|
||
LoginRequest,
|
||
RegisterRequest,
|
||
TokenResponse,
|
||
UserPublic,
|
||
)
|
||
from app.core.heartbeat_verify import compute_server_verify_code
|
||
from app.schemas.common import ApiResponse
|
||
from app.services import auth as auth_service
|
||
|
||
router = APIRouter(prefix="/auth", tags=["认证"])
|
||
|
||
|
||
@router.post("/register", response_model=ApiResponse[TokenResponse])
|
||
async def register(
|
||
body: RegisterRequest,
|
||
db: DbSession,
|
||
redis: RedisClient,
|
||
device_serial: str | None = Header(default=None, convert_underscores=False),
|
||
oem_id: str | None = Header(default=None, convert_underscores=False),
|
||
agent_id: str | None = Header(default=None, convert_underscores=False),
|
||
) -> ApiResponse[TokenResponse]:
|
||
try:
|
||
parsed_oem_id = auth_service.parse_id_header(oem_id, field_name="oem_id")
|
||
parsed_agent_id = auth_service.parse_id_header(agent_id, field_name="agent_id")
|
||
result = await auth_service.register_user(
|
||
db,
|
||
username=body.username,
|
||
password=body.password,
|
||
device_serial=device_serial,
|
||
oem_id=parsed_oem_id,
|
||
agent_id=parsed_agent_id,
|
||
)
|
||
except auth_service.AuthError as exc:
|
||
return ApiResponse(ok=False, message=exc.message)
|
||
|
||
await auth_service.store_session(redis, result.access_token, result.user.id)
|
||
return ApiResponse(ok=True, message="注册成功", data=result)
|
||
|
||
|
||
@router.post("/login", response_model=ApiResponse[TokenResponse])
|
||
async def login(
|
||
body: LoginRequest,
|
||
db: DbSession,
|
||
redis: RedisClient,
|
||
device_serial: str | None = Header(default=None, convert_underscores=False),
|
||
) -> ApiResponse[TokenResponse]:
|
||
try:
|
||
result = await auth_service.login_user(
|
||
db,
|
||
username=body.username,
|
||
password=body.password,
|
||
device_serial=device_serial,
|
||
)
|
||
except auth_service.AuthError as exc:
|
||
return ApiResponse(ok=False, message=exc.message)
|
||
|
||
await auth_service.store_session(redis, result.access_token, result.user.id)
|
||
return ApiResponse(ok=True, message="登录成功", data=result)
|
||
|
||
|
||
@router.post("/heartbeat", response_model=ApiResponse[HeartbeatResponse])
|
||
async def heartbeat(
|
||
body: HeartbeatRequest,
|
||
current_user: CurrentUser,
|
||
db: DbSession,
|
||
device_serial: str | None = Header(default=None, convert_underscores=False),
|
||
) -> ApiResponse[HeartbeatResponse]:
|
||
verify_code = body.verify_code.strip()
|
||
if not verify_code:
|
||
return ApiResponse(ok=False, message="verify_code 无效")
|
||
|
||
# 无论业务成败都返回 server_verify_code，供客户端校验响应真实性
|
||
server_verify = compute_server_verify_code(verify_code)
|
||
try:
|
||
data = await auth_service.process_heartbeat(
|
||
db,
|
||
current_user,
|
||
device_serial,
|
||
verify_code,
|
||
)
|
||
except auth_service.AuthError as exc:
|
||
return ApiResponse(
|
||
ok=False,
|
||
message=exc.message,
|
||
data=HeartbeatResponse(
|
||
vip_end_time=current_user.vip_end_time,
|
||
vip_active=False,
|
||
server_verify_code=server_verify,
|
||
),
|
||
)
|
||
return ApiResponse(ok=True, message="", data=data)
|
||
|
||
|
||
@router.post("/logout", response_model=ApiResponse[None])
|
||
async def logout(
|
||
redis: RedisClient,
|
||
authorization: str | None = Header(default=None),
|
||
) -> ApiResponse[None]:
|
||
if authorization and authorization.lower().startswith("bearer "):
|
||
token = authorization[7:].strip()
|
||
if token:
|
||
await auth_service.revoke_session(redis, token)
|
||
return ApiResponse(ok=True, message="已退出登录")
|
||
|
||
|
||
@router.get("/me", response_model=ApiResponse[UserPublic])
|
||
async def me(current_user: CurrentUser) -> ApiResponse[UserPublic]:
|
||
return ApiResponse(
|
||
ok=True,
|
||
message="",
|
||
data=UserPublic.model_validate(current_user),
|
||
)
|