harness_zbt/backend/src/zbt/api/v1/h5_auth.py

"""H5 登录相关 HTTP 接口。

Pydantic 模型负责校验请求体，FastAPI 的 Depends 负责在接口执行前完成身份验证。
"""

from typing import Annotated, Any

from fastapi import APIRouter, Depends, Request, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field

from zbt.api.responses import success_response
from zbt.core.config import Settings
from zbt.core.errors import AppError
from zbt.domains.attribution.service import AttributionService
from zbt.domains.identity.models import H5User
from zbt.domains.identity.service import H5AuthService

bearer_scheme = HTTPBearer(auto_error=False)
H5_REFRESH_COOKIE = "zbt_h5_refresh"
H5_REFRESH_COOKIE_PATH = "/api/v1/h5/auth"


class MobileRequest(BaseModel):
    """请求验证码时的 JSON 请求体。"""

    mobile: str = Field(min_length=11, max_length=11)


class H5LoginRequest(MobileRequest):
    code: str = Field(min_length=6, max_length=6)
    referral_code: str | None = Field(default=None, max_length=32)


class RefreshRequest(BaseModel):
    refresh_token: str = Field(min_length=32)


def create_h5_router(
    auth_service: H5AuthService,
    settings: Settings,
    attribution_service: AttributionService | None = None,
) -> APIRouter:
    """创建 H5 认证路由，并把认证服务放入各接口的闭包中。"""
    router = APIRouter(tags=["h5-auth"])

    def bearer_token(
        credentials: Annotated[
            HTTPAuthorizationCredentials | None,
            Depends(bearer_scheme),
        ],
    ) -> str:
        # Depends 会从 Authorization: Bearer <token> 请求头中解析凭证。
        if credentials is None:
            raise AppError("AUTH_REQUIRED", "请先登录", 401)
        return credentials.credentials

    def current_user(
        token: Annotated[str, Depends(bearer_token)],
    ) -> H5User:
        # 依赖可以继续依赖另一个依赖：先取 token，再查询当前用户。
        return auth_service.authenticate_access(token)

    @router.post("/h5/auth/request-code")
    def request_code(payload: MobileRequest, request: Request) -> dict[str, Any]:
        return success_response(request, auth_service.request_code(payload.mobile))

    def set_refresh_cookie(response: Response, refresh_token: str) -> None:
        response.set_cookie(
            key=H5_REFRESH_COOKIE,
            value=refresh_token,
            max_age=settings.refresh_token_days * 24 * 60 * 60,
            httponly=True,
            secure=settings.app_env == "production",
            samesite="lax",
            path=H5_REFRESH_COOKIE_PATH,
        )

    @router.post("/h5/auth/login")
    def login(
        payload: H5LoginRequest,
        request: Request,
        response: Response,
    ) -> dict[str, Any]:
        # payload 已由 FastAPI 转成 H5LoginRequest，无需手动解析 JSON。
        result = auth_service.login(payload.mobile, payload.code)
        if payload.referral_code and attribution_service is not None:
            user = auth_service.authenticate_access(str(result["tokens"]["access_token"]))
            result["attribution"] = attribution_service.capture_visit(
                user,
                payload.referral_code,
            )
        set_refresh_cookie(response, result["tokens"]["refresh_token"])
        return success_response(request, result)

    @router.post("/h5/auth/refresh")
    def refresh(
        request: Request,
        response: Response,
        payload: RefreshRequest | None = None,
    ) -> dict[str, Any]:
        refresh_token = (
            payload.refresh_token if payload is not None else request.cookies.get(H5_REFRESH_COOKIE)
        )
        if refresh_token is None:
            raise AppError("AUTH_REQUIRED", "缺少刷新凭证", 401)
        result = auth_service.refresh(refresh_token)
        set_refresh_cookie(response, result["tokens"]["refresh_token"])
        return success_response(request, result)

    @router.post("/h5/auth/logout")
    def logout(
        request: Request,
        response: Response,
        token: Annotated[str, Depends(bearer_token)],
    ) -> dict[str, Any]:
        auth_service.logout(token)
        response.delete_cookie(
            key=H5_REFRESH_COOKIE,
            path=H5_REFRESH_COOKIE_PATH,
            samesite="lax",
        )
        return success_response(request, {"logged_out": True})

    @router.get("/h5/me")
    def me(
        request: Request,
        user: Annotated[H5User, Depends(current_user)],
    ) -> dict[str, Any]:
        return success_response(
            request,
            {
                "id": user.id,
                "mobile_masked": user.mobile_masked,
                "display_name": user.display_name,
                "status": user.status,
            },
        )

    return router