"""拼团 (group-buy) funnel query orchestration (docs/02 §5/§6 v3). Pipeline: validate the request -> route by ``period`` to the right table/row -> pick the five UV columns -> derive conversion/dropoff -> resolve data_status -> assemble response. Routing (docs/02 §6 v3): * ``day`` -> ``ads_trd_group_funnel_daily``; a given ``snapshot_dt`` selects that row, else the latest ``dt``. Columns: ``uv_{step}``. * ``last_7d`` -> ``ads_trd_group_funnel_rolling`` (single row). Columns: ``uv_{step}_7d``. * ``last_30d`` -> same rolling row. Columns: ``uv_{step}_30d``. No bitmaps, no OR, no cross-day aggregation. The repository layer is an abstract protocol so the service can be unit-tested without Postgres (and so a fake in-memory source can stand in when no DB is configured). """ from __future__ import annotations from dataclasses import dataclass from datetime import date, timedelta from typing import Protocol from app.schemas import ( DataStatus, FunnelQueryResponse, FunnelStepResult, Period, ) # Fixed funnel order: 启动 -> 曝光 -> 拼团详情 -> 下单 -> 成功 (docs/03 §11). # Each entry: (event_key, Chinese display name). FUNNEL_STEPS: list[tuple[str, str]] = [ ("start", "启动"), ("show", "曝光"), ("detail", "拼团详情"), ("order", "下单"), ("paid", "成功"), ] # period -> rolling-table column suffix (for last_7d / last_30d). ROLLING_SUFFIX: dict[Period, str] = { Period.last_7d: "7d", Period.last_30d: "30d", } def daily_columns() -> list[str]: """Return the five ``uv_{step}`` daily-table column names, in funnel order.""" return [f"uv_{event_key}" for event_key, _ in FUNNEL_STEPS] def rolling_columns(period: Period) -> list[str]: """Return the five ``uv_{step}_{suffix}`` rolling-table columns for a period.""" suffix = ROLLING_SUFFIX[period] return [f"uv_{event_key}_{suffix}" for event_key, _ in FUNNEL_STEPS] # --------------------------------------------------------------------------- # # Validation # # --------------------------------------------------------------------------- # class SnapshotDateError(ValueError): """Raised when ``snapshot_dt`` is today or in the future (data is T+1).""" def validate_snapshot_dt(snapshot_dt: date) -> None: """Reject a ``snapshot_dt`` later than yesterday. Data is T+1: today's data is not computed yet, so the max queryable day is always yesterday. today or future -> :class:`SnapshotDateError`. """ yesterday = date.today() - timedelta(days=1) if snapshot_dt > yesterday: raise SnapshotDateError( f"snapshot_dt {snapshot_dt.isoformat()} is later than yesterday " f"({yesterday.isoformat()}); data is T+1 and today's data is not " "computed yet." ) # --------------------------------------------------------------------------- # # Repository protocol + snapshot container # # --------------------------------------------------------------------------- # @dataclass(frozen=True) class FunnelSnapshot: """A funnel source row as returned by the repository. ``values`` maps column name to its value (NULL columns map to ``None``). ``dt`` is the row's ``dt`` (yyyyMMdd): the day for daily, the as-of day for rolling. """ dt: str values: dict[str, int | None] class FunnelRepository(Protocol): """Data access boundary. Implementations may hit Postgres or be in-memory.""" async def fetch_daily(self, dt: str | None) -> FunnelSnapshot | None: """Return a row from the daily table. ``dt`` (yyyyMMdd) selects that specific day; ``None`` returns the latest ``dt`` row. ``None`` when no matching row exists. """ ... async def fetch_rolling(self) -> FunnelSnapshot | None: """Return the single rolling row, or ``None`` when the table is empty.""" ... # --------------------------------------------------------------------------- # # Conversion math # # --------------------------------------------------------------------------- # def build_results(uvs: list[int]) -> list[FunnelStepResult]: """Assemble per-step results with conversion/dropoff math. Step 1: conversion_rate and dropoff_rate are null. Step i: conversion = uv[i] / uv[i-1]; dropoff = 1 - conversion. If uv[i-1] == 0, conversion (and dropoff) are null. """ results: list[FunnelStepResult] = [] for idx, (uv, (event_key, name)) in enumerate(zip(uvs, FUNNEL_STEPS)): if idx == 0: conversion: float | None = None dropoff: float | None = None else: prev = uvs[idx - 1] if prev == 0: conversion = None dropoff = None else: conversion = uv / prev dropoff = 1 - conversion results.append( FunnelStepResult( step_index=idx + 1, name=name, event_key=event_key, uv=uv, conversion_rate=conversion, dropoff_rate=dropoff, ) ) return results # --------------------------------------------------------------------------- # # Orchestration # # --------------------------------------------------------------------------- # def _assemble( period: Period, snapshot: FunnelSnapshot | None, column_names: list[str], ) -> FunnelQueryResponse: """Map a fetched snapshot + period columns to the response model. Missing row or any NULL period column -> ``data_status = missing`` with no zero-fill. ``snapshot_dt`` echoes the row used (None when no row). """ if snapshot is None: return FunnelQueryResponse( period=period, snapshot_dt=None, results=[], data_status=DataStatus.missing, ) raw = [snapshot.values.get(col) for col in column_names] if any(value is None for value in raw): return FunnelQueryResponse( period=period, snapshot_dt=snapshot.dt, results=[], data_status=DataStatus.missing, ) uvs = [int(value) for value in raw] # narrow Optional -> int return FunnelQueryResponse( period=period, snapshot_dt=snapshot.dt, results=build_results(uvs), data_status=DataStatus.ready, ) async def run_funnel_query( period: Period, repo: FunnelRepository, snapshot_dt: date | None = None, ) -> FunnelQueryResponse: """Run the group-buy funnel query pipeline and return the response model. Routes by ``period``: * ``day`` -> daily table; validates ``snapshot_dt`` (<= yesterday) when given, then selects that day or the latest row. * ``last_7d`` / ``last_30d`` -> the single rolling row; ``snapshot_dt`` is ignored. """ if period is Period.day: dt_str: str | None = None if snapshot_dt is not None: validate_snapshot_dt(snapshot_dt) # may raise SnapshotDateError dt_str = snapshot_dt.strftime("%Y%m%d") snapshot = await repo.fetch_daily(dt_str) return _assemble(period, snapshot, daily_columns()) # last_7d / last_30d -> rolling row; snapshot_dt ignored. snapshot = await repo.fetch_rolling() return _assemble(period, snapshot, rolling_columns(period))