hs-data/apps/api/app/schemas.py

"""Pydantic v2 request/response models for the 拼团 (group-buy) funnel query API.

Field names match the API contract in docs/02 §5 (v3) exactly:
  request:  period, snapshot_dt (optional)
  response: period, snapshot_dt,
            results[].step_index, results[].name, results[].event_key,
            results[].uv, results[].conversion_rate, results[].dropoff_rate,
            data_status
"""

from __future__ import annotations

from datetime import date
from enum import Enum

from pydantic import BaseModel, Field


class Period(str, Enum):
    """Supported periods (docs/02 §5 v3).

    Routing:
      day      -> table ads_trd_group_funnel_daily (single day, full history).
      last_7d  -> table ads_trd_group_funnel_rolling, columns uv_*_7d.
      last_30d -> table ads_trd_group_funnel_rolling, columns uv_*_30d.

    Any other value is rejected with HTTP 422.
    """

    day = "day"
    last_7d = "last_7d"
    last_30d = "last_30d"


class FunnelQueryRequest(BaseModel):
    """Funnel query request body.

    ``snapshot_dt`` (ISO ``YYYY-MM-DD``) is optional and only meaningful for
    ``period=day``: omitted -> latest daily row; given -> that historical day
    (must be <= yesterday). For ``last_7d`` / ``last_30d`` it is ignored.
    """

    period: Period
    snapshot_dt: date | None = Field(
        None,
        description=(
            "Optional ISO date (YYYY-MM-DD). Only meaningful for period=day; "
            "ignored for last_7d/last_30d. Must be <= yesterday."
        ),
    )


class DataStatus(str, Enum):
    """Data completeness status for a query result (docs/02 §5 v3).

      ready   - the target row exists and the period's columns are non-null.
      missing - no target row, or the period columns are NULL. Never zero-filled.
    """

    ready = "ready"
    missing = "missing"


class FunnelStepResult(BaseModel):
    """Computed UV and conversion metrics for one fixed step."""

    step_index: int = Field(..., ge=1, description="1-based step index")
    name: str = Field(..., description="Chinese display name of the step")
    event_key: str = Field(..., description="Stable step key (start/show/...)")
    uv: int = Field(..., ge=0)
    conversion_rate: float | None = Field(
        None, description="uv[i] / uv[i-1]; null for step 1 or when uv[i-1] == 0"
    )
    dropoff_rate: float | None = Field(
        None, description="1 - conversion_rate; null when conversion_rate is null"
    )


class FunnelQueryResponse(BaseModel):
    """Funnel query response body."""

    period: Period
    snapshot_dt: str | None = Field(
        None,
        description=(
            "dt (yyyyMMdd) of the row actually used; for day = that day, for "
            "7d/30d = the rolling row's as-of dt. null when missing."
        ),
    )
    results: list[FunnelStepResult]
    data_status: DataStatus