"""Admin dashboard stats with precise date range reporting and smart start detection.""" from datetime import datetime, timedelta, timezone from typing import Any, Optional import redis as redis_lib from sqlalchemy import func, select from sqlmodel import Session from app.models.join_team_members import JoinTeamMembers from app.models.link_clicked import LinkClicked from app.models.team import Team from app.models.user_invite import UserInvite _ALLTIME_START_KEY = "admin:alltime_start" _ALLTIME_START_TTL = 86400 # 24 hours # Semantic time range → (days, interval) TIME_RANGES: dict[str, tuple[int, str]] = { "past_7_days": (7, "day"), "past_14_days": (14, "day"), "past_30_days": (30, "day"), "past_90_days": (90, "day"), "past_1_year": (365, "month"), "past_2_years": (730, "month"), "past_5_years": (1825, "year"), "all_time": (36500, "year"), } def _get_alltime_start(session: Session, redis: Optional[redis_lib.Redis]) -> datetime: """Return the earliest data point as a year-start datetime. Checks Redis cache first; falls back to DB query and populates cache.""" now = datetime.now(timezone.utc) if redis is not None: try: cached = redis.get(_ALLTIME_START_KEY) if cached: return datetime.fromisoformat(cached) except Exception: pass # Redis unavailable, continue to DB starts = [] for col in (LinkClicked.click_time, UserInvite.invited_time): earliest = session.scalar(select(func.min(col))) if earliest: starts.append(earliest) if starts: start_dt = min(starts) if start_dt.tzinfo is None: start_dt = start_dt.replace(tzinfo=timezone.utc) start = start_dt.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0) else: start = now - timedelta(days=365) if redis is not None: try: redis.set(_ALLTIME_START_KEY, start.isoformat(), ex=_ALLTIME_START_TTL) except Exception: pass return start def get_admin_dashboard_stats( session: Session, time_range: str = "past_14_days", redis: Optional[redis_lib.Redis] = None, ) -> dict[str, Any]: days, interval = TIME_RANGES.get(time_range, TIME_RANGES["past_14_days"]) now = datetime.now(timezone.utc) # ── Start Time Logic ── if days > 10000: # all_time start = _get_alltime_start(session, redis) else: raw_start = now - timedelta(days=days) if interval == "month": start = raw_start.replace(day=1, hour=0, minute=0, second=0, microsecond=0) elif interval == "year": start = raw_start.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0) else: start = raw_start.replace(hour=0, minute=0, second=0, microsecond=0) # Summary window (rolling or absolute) summary_start = datetime(1970, 1, 1, tzinfo=timezone.utc) if days > 10000 else now - timedelta(days=days) # ── Actual Display Dates ── # If it's "All Time", we show from the detected start. Otherwise, from requested window. display_start = start.strftime("%b %d, %Y") if days > 10000 else summary_start.strftime("%b %d, %Y") display_end = now.strftime("%b %d, %Y") team_count = session.scalar(select(func.count(Team.team_id))) or 0 clicks_period = session.scalar(select(func.count(LinkClicked.id)).where(LinkClicked.click_time >= summary_start)) or 0 invites_period = session.scalar(select(func.count(UserInvite.id)).where(UserInvite.invited_time >= summary_start)) or 0 joins_period = session.scalar(select(func.count(JoinTeamMembers.id)).where(JoinTeamMembers.team_join_time >= summary_start)) or 0 click_to_invite = round((invites_period / clicks_period * 100), 1) if clicks_period > 0 else 0 invite_to_join = round((joins_period / invites_period * 100), 1) if invites_period > 0 else 0 # Time-series data clicks_by_interval = _count_by_interval(session, LinkClicked, LinkClicked.click_time, start, now, interval) invites_by_interval = _count_by_interval(session, UserInvite, UserInvite.invited_time, start, now, interval) joins_by_interval = _count_by_interval(session, JoinTeamMembers, JoinTeamMembers.team_join_time, start, now, interval) labels = [] curr = start while curr <= now: if interval == "month": labels.append(curr.strftime("%Y-%m")) curr = (curr.replace(day=28) + timedelta(days=4)).replace(day=1) elif interval == "year": labels.append(curr.strftime("%Y")) curr = curr.replace(year=curr.year + 1) else: labels.append(curr.strftime("%Y-%m-%d")) curr += timedelta(days=1) def series_for_labels(by_val: dict[str, int]) -> list[int]: return [by_val.get(label, 0) for label in labels] # Analysis referrer_stmt = select(func.coalesce(LinkClicked.referrer, "direct").label("source"), func.count(LinkClicked.id).label("cnt")).where(LinkClicked.click_time >= summary_start, LinkClicked.click_time <= now) referrer_results = session.exec(referrer_stmt.group_by(func.coalesce(LinkClicked.referrer, "direct"))).all() sources = sorted([{"name": str(r[0] if isinstance(r, (tuple, list)) else r.source), "value": int(r[1] if isinstance(r, (tuple, list)) else r.cnt)} for r in referrer_results], key=lambda x: -x["value"])[:10] channel_stmt = select(UserInvite.channel.label("chan"), func.count(UserInvite.id).label("cnt")).where(UserInvite.invited_time >= summary_start, UserInvite.invited_time <= now) channel_results = session.exec(channel_stmt.group_by(UserInvite.channel)).all() channels = sorted([{"name": str(r[0] if isinstance(r, (tuple, list)) else r.chan), "value": int(r[1] if isinstance(r, (tuple, list)) else r.cnt)} for r in channel_results], key=lambda x: -x["value"])[:10] return { "summary": { "teams": team_count, "clicks": clicks_period, "invites": invites_period, "joins": joins_period, "cr_click_invite": click_to_invite, "cr_invite_join": invite_to_join, }, "chart": { "labels": labels, "link_clicks": series_for_labels(clicks_by_interval), "invites": series_for_labels(invites_by_interval), "joins": series_for_labels(joins_by_interval), }, "sources": sources, "channels": channels, "display_range": {"start": display_start, "end": display_end}, "time_range": time_range, "interval": interval } def _count_by_interval(session, model, date_col, start, end, interval): if interval == "month": date_expr = func.strftime("%Y-%m", date_col) if "sqlite" in str(session.bind.url) else func.date_format(date_col, "%Y-%m") elif interval == "year": date_expr = func.strftime("%Y", date_col) if "sqlite" in str(session.bind.url) else func.date_format(date_col, "%Y") else: date_expr = func.date(date_col) stmt = select(date_expr.label("p"), func.count().label("c")).select_from(model).where(date_col >= start, date_col <= end) rows = session.exec(stmt.group_by(date_expr)).all() return {str(r[0] if isinstance(r, (tuple, list)) else r.p): int(r[1] if isinstance(r, (tuple, list)) else r.c) for r in rows if r}