krz/hutch-stats

  1from __future__ import annotations
  2
  3from dataclasses import dataclass
  4from datetime import date
  5
  6from sqlalchemy import func, select
  7from sqlalchemy.orm import Session
  8
  9from srht_contrib.models import ContributionEvent, TrackedActor
 10from srht_contrib.schemas import (
 11    ContributionCalendarResponse,
 12    ContributionDay,
 13    ContributionIndexMetadata,
 14    ContributionStatsResponse,
 15)
 16from srht_contrib.utils.dates import date_range, date_to_utc_bounds
 17
 18
 19@dataclass(slots=True)
 20class DailyAggregate:
 21    date: date
 22    count: int
 23    score: float
 24
 25
 26class ContributionAggregator:
 27    def build_calendar(self, db: Session, actor: str, start: date, end: date) -> ContributionCalendarResponse:
 28        aggregates = self._query_daily_aggregates(db, actor, start, end)
 29        by_day = {row.date: row for row in aggregates}
 30        days = [
 31            ContributionDay(
 32                date=day,
 33                count=by_day.get(day, DailyAggregate(date=day, count=0, score=0.0)).count,
 34                score=by_day.get(day, DailyAggregate(date=day, count=0, score=0.0)).score,
 35            )
 36            for day in date_range(start, end)
 37        ]
 38        metadata = self._index_metadata(db, actor)
 39        return ContributionCalendarResponse(
 40            actor=actor,
 41            from_date=start,
 42            to_date=end,
 43            days=days,
 44            **metadata.model_dump(),
 45        )
 46
 47    def build_stats(self, db: Session, actor: str, start: date, end: date) -> ContributionStatsResponse:
 48        calendar = self.build_calendar(db, actor, start, end)
 49        active_days = [day for day in calendar.days if day.count > 0]
 50        streaks = self._streak_lengths(calendar.days)
 51        current_streak = self._current_streak(calendar.days)
 52
 53        return ContributionStatsResponse(
 54            actor=actor,
 55            from_date=start,
 56            to_date=end,
 57            total_events=sum(day.count for day in calendar.days),
 58            total_score=round(sum(day.score for day in calendar.days), 2),
 59            active_days=len(active_days),
 60            longest_streak=max(streaks, default=0),
 61            current_streak=current_streak,
 62            is_indexed=calendar.is_indexed,
 63            last_polled_at=calendar.last_polled_at,
 64            indexing_state=calendar.indexing_state,
 65            is_recent_window_backfilled=calendar.is_recent_window_backfilled,
 66            recent_backfill_state=calendar.recent_backfill_state,
 67            recent_backfill_completed_at=calendar.recent_backfill_completed_at,
 68        )
 69
 70    def _index_metadata(self, db: Session, actor: str) -> ContributionIndexMetadata:
 71        tracked_actor = db.scalar(select(TrackedActor).where(TrackedActor.actor == actor))
 72        has_indexed_events = db.scalar(select(ContributionEvent.id).where(ContributionEvent.actor == actor).limit(1)) is not None
 73        last_poll_status = tracked_actor.last_poll_status if tracked_actor is not None else None
 74        is_indexed = has_indexed_events or (tracked_actor is not None and tracked_actor.last_polled_at is not None)
 75
 76        if last_poll_status == "error":
 77            indexing_state = "error"
 78        elif is_indexed:
 79            indexing_state = "indexed"
 80        else:
 81            indexing_state = "pending"
 82
 83        return ContributionIndexMetadata(
 84            is_indexed=is_indexed,
 85            last_polled_at=tracked_actor.last_polled_at if tracked_actor is not None else None,
 86            indexing_state=indexing_state,
 87            is_recent_window_backfilled=(tracked_actor.recent_backfill_status == "completed") if tracked_actor is not None else False,
 88            recent_backfill_state=(tracked_actor.recent_backfill_status if tracked_actor is not None else "pending"),
 89            recent_backfill_completed_at=tracked_actor.recent_backfill_completed_at if tracked_actor is not None else None,
 90        )
 91
 92    def _query_daily_aggregates(self, db: Session, actor: str, start: date, end: date) -> list[DailyAggregate]:
 93        start_dt, _ = date_to_utc_bounds(start)
 94        _, end_dt = date_to_utc_bounds(end)
 95
 96        stmt = (
 97            select(
 98                func.date(ContributionEvent.occurred_at).label("day"),
 99                func.count(ContributionEvent.id).label("count"),
100                func.coalesce(func.sum(ContributionEvent.weight), 0.0).label("score"),
101            )
102            .where(ContributionEvent.actor == actor)
103            .where(ContributionEvent.occurred_at >= start_dt)
104            .where(ContributionEvent.occurred_at <= end_dt)
105            .group_by(func.date(ContributionEvent.occurred_at))
106            .order_by(func.date(ContributionEvent.occurred_at))
107        )
108        rows = db.execute(stmt).all()
109        return [
110            DailyAggregate(
111                date=date.fromisoformat(str(row.day)),
112                count=int(row.count),
113                score=round(float(row.score), 2),
114            )
115            for row in rows
116        ]
117
118    @staticmethod
119    def _streak_lengths(days: list[ContributionDay]) -> list[int]:
120        streaks: list[int] = []
121        current = 0
122        for day in days:
123            if day.count > 0:
124                current += 1
125            elif current > 0:
126                streaks.append(current)
127                current = 0
128        if current > 0:
129            streaks.append(current)
130        return streaks
131
132    @staticmethod
133    def _current_streak(days: list[ContributionDay]) -> int:
134        streak = 0
135        for day in reversed(days):
136            if day.count > 0:
137                streak += 1
138            else:
139                break
140        return streak