hs-data/apps/api/app/services/repository.py

"""Funnel repository implementations.

Two concrete sources behind the same :class:`FunnelRepository` protocol so route
and service code is identical regardless of backing store:

* :class:`SqlAlchemyFunnelRepository` - reads the two group-buy funnel tables
  (``ads_trd_group_funnel_daily`` + ``ads_trd_group_funnel_rolling``) from
  Postgres via an :class:`AsyncSession`.
* :class:`FakeFunnelRepository` - realistic in-memory data (daily history + one
  rolling row) so the page works with zero infrastructure (``USE_FAKE_DATA=true``).
"""

from __future__ import annotations

from datetime import date, datetime, timedelta

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import AdsTrdGroupFunnelDaily, AdsTrdGroupFunnelRolling
from app.services.funnel import FunnelSnapshot

# Daily table UV columns (funnel order).
_DAILY_COLUMNS: tuple[str, ...] = (
    "uv_start",
    "uv_show",
    "uv_detail",
    "uv_order",
    "uv_paid",
)

# Rolling table UV columns (7d window then 30d window).
_ROLLING_COLUMNS: tuple[str, ...] = (
    "uv_start_7d",
    "uv_show_7d",
    "uv_detail_7d",
    "uv_order_7d",
    "uv_paid_7d",
    "uv_start_30d",
    "uv_show_30d",
    "uv_detail_30d",
    "uv_order_30d",
    "uv_paid_30d",
)


class SqlAlchemyFunnelRepository:
    """Concrete repository backed by an :class:`AsyncSession` over Postgres."""

    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def fetch_daily(self, dt: str | None) -> FunnelSnapshot | None:
        stmt = select(AdsTrdGroupFunnelDaily)
        if dt is not None:
            stmt = stmt.where(AdsTrdGroupFunnelDaily.dt == dt)
        else:
            # Latest day. dt is yyyyMMdd so lexical == chronological order.
            stmt = stmt.order_by(AdsTrdGroupFunnelDaily.dt.desc()).limit(1)
        row = (await self._session.execute(stmt)).scalars().first()
        if row is None:
            return None
        values = {col: getattr(row, col) for col in _DAILY_COLUMNS}
        return FunnelSnapshot(dt=row.dt, values=values)

    async def fetch_rolling(self) -> FunnelSnapshot | None:
        # Single-row table; order by dt desc + limit 1 is defensive.
        stmt = (
            select(AdsTrdGroupFunnelRolling)
            .order_by(AdsTrdGroupFunnelRolling.dt.desc())
            .limit(1)
        )
        row = (await self._session.execute(stmt)).scalars().first()
        if row is None:
            return None
        values = {col: getattr(row, col) for col in _ROLLING_COLUMNS}
        return FunnelSnapshot(dt=row.dt, values=values)


# Cumulative keep-rate down the group-buy funnel: start, show, detail, order, paid.
_FUNNEL_KEEP = [1.0, 0.82, 0.51, 0.22, 0.18]
_STEP_KEYS = ["start", "show", "detail", "order", "paid"]


def _funnel(top: int) -> list[int]:
    """Descending UVs for the five steps from the top-step UV."""
    return [max(0, int(top * keep)) for keep in _FUNNEL_KEEP]


class FakeFunnelRepository:
    """In-memory repository with realistic group-buy funnel data, no DB needed.

    Used when no database is configured (``USE_FAKE_DATA=true``). Provides:

    * a span of ``HISTORY_DAYS`` daily rows ending yesterday, so the single-day
      date picker has history and different ``snapshot_dt`` values yield
      different data;
    * one rolling row as-of yesterday for the 7d/30d windows.

    A ``snapshot_dt`` with no matching daily row -> ``fetch_daily`` returns None,
    so the service reports ``missing``.
    """

    # Number of historical daily rows (ending yesterday).
    HISTORY_DAYS = 10

    # Top-step (start) daily UV for the most recent day; older days scale down.
    DAILY_TOP = 12000

    # Top-step UV for the rolling windows.
    ROLLING_TOP_7D = 74000
    ROLLING_TOP_30D = 295000

    def __init__(self) -> None:
        today = date.today()
        yesterday = today - timedelta(days=1)
        now = datetime.utcnow()

        # Build daily history: yyyyMMdd -> value map. Newest day = full size,
        # older days a touch smaller so rows differ.
        self._daily: dict[str, dict[str, int | None]] = {}
        for offset in range(self.HISTORY_DAYS):
            day = yesterday - timedelta(days=offset)
            dt = day.strftime("%Y%m%d")
            scale = 1.0 - 0.04 * offset
            top = int(self.DAILY_TOP * scale)
            self._daily[dt] = {
                col: uv for col, uv in zip(_DAILY_COLUMNS, _funnel(top))
            }
        self._latest_daily_dt = yesterday.strftime("%Y%m%d")

        # Rolling row, as-of yesterday.
        self._rolling_dt = yesterday.strftime("%Y%m%d")
        self._rolling: dict[str, int | None] = {}
        for col, uv in zip(_ROLLING_COLUMNS[:5], _funnel(self.ROLLING_TOP_7D)):
            self._rolling[col] = uv
        for col, uv in zip(_ROLLING_COLUMNS[5:], _funnel(self.ROLLING_TOP_30D)):
            self._rolling[col] = uv

        self._etl_time = now

    async def fetch_daily(self, dt: str | None) -> FunnelSnapshot | None:
        if dt is None:
            dt = self._latest_daily_dt
        values = self._daily.get(dt)
        if values is None:
            return None
        return FunnelSnapshot(dt=dt, values=dict(values))

    async def fetch_rolling(self) -> FunnelSnapshot | None:
        return FunnelSnapshot(dt=self._rolling_dt, values=dict(self._rolling))