diff --git a/signal_bot/stats/queries_period.py b/signal_bot/stats/queries_period.py new file mode 100644 index 0000000..fb1102a --- /dev/null +++ b/signal_bot/stats/queries_period.py @@ -0,0 +1,173 @@ +# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. +"""Weeks, months, years, and the per-weekday picture. + +Every query takes an inclusive span of stored Warsaw day strings, so a +daylight-saving change cannot move a boundary -- see ``stats.periods``. + +Several of them take ``exclude``: the one day carried over from the pre-0.1.0 +counter has real totals but no hours, no lengths, and a 22-hour window +pretending to be a day. It belongs in a total and nowhere near an average. + +The exclusion is a bound parameter compared against ``''`` rather than a +clause pasted into the statement. No real day is the empty string, so one +static query serves both cases -- and the alternative, an f-string holding +half a WHERE clause, is a SQL-injection shape even when every input is a +module constant. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from signal_bot.stats.periods import weekday_of +from signal_bot.stats.queries_day import DayRow + +if TYPE_CHECKING: + from signal_bot.stats.events import EventStore + from signal_bot.stats.periods import Period + +PERIOD_ROWS = """ + SELECT uuid, + display_name, + COUNT(*) AS messages, + SUM(media_count) AS media, + SUM(is_command) AS commands, + SUM(char_len) AS chars + FROM message_events + WHERE group_id = ? AND day BETWEEN ? AND ? AND day <> ? + GROUP BY uuid + ORDER BY messages DESC, display_name ASC +""" + +DAILY_SERIES = """ + SELECT day, COUNT(*) AS messages + FROM message_events + WHERE group_id = ? AND day BETWEEN ? AND ? AND day <> ? + GROUP BY day + ORDER BY day +""" + +EVERY_DAY = """ + SELECT day, COUNT(*) AS messages + FROM message_events + WHERE group_id = ? AND day <> ? + GROUP BY day + ORDER BY day +""" + + +def period_rows( + events: EventStore, + group_id: str, + period: Period, + *, + exclude: str | None = None, +) -> list[DayRow]: + """Everyone who wrote during a period, most talkative first.""" + rows = events.connection.execute( + PERIOD_ROWS, (group_id, period.start, period.end, exclude or "") + ).fetchall() + return [ + DayRow( + uuid=r["uuid"], + display_name=r["display_name"], + messages=r["messages"], + media=r["media"], + commands=r["commands"], + chars=r["chars"], + ) + for r in rows + ] + + +def daily_series( + events: EventStore, + group_id: str, + start: str, + end: str, + *, + exclude: str | None = None, +) -> dict[str, int]: + """Messages per day across a span, missing days absent rather than zero. + + The caller decides what a missing day means: for a streak it is silence, + for an average it is a day nobody was there to record. + """ + rows = events.connection.execute(DAILY_SERIES, (group_id, start, end, exclude or "")).fetchall() + return {r["day"]: r["messages"] for r in rows} + + +def every_day( + events: EventStore, + group_id: str, + *, + exclude: str | None = None, +) -> dict[str, int]: + """Messages per day across the whole history.""" + rows = events.connection.execute(EVERY_DAY, (group_id, exclude or "")).fetchall() + return {r["day"]: r["messages"] for r in rows} + + +def weekday_buckets(days: dict[str, int]) -> dict[int, list[int]]: + """Bucket a day series by weekday, 0=Monday through 6=Sunday. + + Returns the days themselves rather than a mean, so a caller can see how + many a figure rests on: "wtorki mają średnio 74" off a single Tuesday is a + number that should not be said out loud. + """ + buckets: dict[int, list[int]] = {} + for day, messages in days.items(): + buckets.setdefault(weekday_of(day), []).append(messages) + return buckets + + +def total_for( + events: EventStore, + group_id: str, + start: str, + end: str, + *, + exclude: str | None = None, +) -> int: + """How many messages a span holds.""" + return sum(daily_series(events, group_id, start, end, exclude=exclude).values()) + + +@dataclass(frozen=True, slots=True) +class Trend: + """One period measured against the one before it.""" + + now: int + before: int + + @property + def delta(self) -> int: + """Absolute change, which may be negative.""" + return self.now - self.before + + @property + def percent(self) -> int | None: + """Rounded percentage change, or None when there is no baseline. + + None rather than 0: "nothing to compare against yet" is a different + statement from "no change", and the renderers say each differently. + """ + if not self.before: + return None + return round(self.delta / self.before * 100) + + @property + def arrow(self) -> str: + """An arrow, or "" when there is nothing to compare against.""" + if self.percent is None or self.delta == 0: + return "" + return "↑" if self.delta > 0 else "↓" + + +def share_of_voice(rows: list[DayRow]) -> list[tuple[str, int]]: + """Each person's percentage of the messages, largest first.""" + total = sum(row.messages for row in rows) + if not total: + return [] + return [(row.display_name, round(row.messages / total * 100)) for row in rows] diff --git a/signal_bot/stats/records.py b/signal_bot/stats/records.py new file mode 100644 index 0000000..12c86a7 --- /dev/null +++ b/signal_bot/stats/records.py @@ -0,0 +1,184 @@ +# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. +"""All-time records: busiest days, quietest days, personal bests, silences. + +Everything here is derived by query rather than kept in a cache table. Two +sources of truth for one number is how a leaderboard comes to disagree with +itself, and the history is small enough that recomputing is free. + +The backfilled day is excluded from every record. It is a 22-hour window +filed as a day, so leaving it in would hand somebody a personal best they +never set -- see ``stats.backfill``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from signal_bot.stats.periods import shift + +if TYPE_CHECKING: + from signal_bot.stats.events import EventStore + +PERSON_DAYS = """ + SELECT day, COUNT(*) AS messages + FROM message_events + WHERE group_id = ? AND uuid = ? AND day <> ? + GROUP BY day + ORDER BY messages DESC, day ASC +""" + +MEDIA_DAYS = """ + SELECT day, SUM(media_count) AS media + FROM message_events + WHERE group_id = ? AND day <> ? + GROUP BY day + HAVING media > 0 + ORDER BY media DESC, day ASC +""" + +DAILY_LEADERS = """ + SELECT day, uuid, display_name, COUNT(*) AS messages + FROM message_events + WHERE group_id = ? AND day <> ? + GROUP BY day, uuid + ORDER BY day ASC +""" + + +@dataclass(frozen=True, slots=True) +class DayRecord: + """One day and the number that makes it a record.""" + + day: str + value: int + + +def busiest_day(days: dict[str, int]) -> DayRecord | None: + """The day with the most messages, earliest day winning a tie.""" + if not days: + return None + day = min(days, key=lambda d: (-days[d], d)) + return DayRecord(day, days[day]) + + +def quietest_day(days: dict[str, int]) -> DayRecord | None: + """The quietest day on which anybody spoke at all. + + Days with no messages are absent from the series rather than zero, and + that is the right reading: a day nobody spoke is a silence, which + ``longest_silence`` measures. A "quietest day" of zero would just name + whichever day the bot happened to be offline. + """ + if not days: + return None + day = min(days, key=lambda d: (days[d], d)) + return DayRecord(day, days[day]) + + +def personal_best( + events: EventStore, + group_id: str, + uuid: str, + *, + exclude: str | None = None, +) -> DayRecord | None: + """Somebody's busiest day ever, or None if they have never written.""" + row = events.connection.execute(PERSON_DAYS, (group_id, uuid, exclude or "")).fetchone() + return DayRecord(row["day"], row["messages"]) if row else None + + +def previous_best( + events: EventStore, + group_id: str, + uuid: str, + before: str, + *, + exclude: str | None = None, +) -> DayRecord | None: + """Somebody's best day *excluding* one day, for "poprzedni rekord". + + Announcing "rekord życiowy: 76" is worth little without what it beat, and + the day being announced must not be its own previous best. + """ + rows = events.connection.execute(PERSON_DAYS, (group_id, uuid, exclude or "")).fetchall() + for row in rows: + if row["day"] != before: + return DayRecord(row["day"], row["messages"]) + return None + + +def most_media_day( + events: EventStore, + group_id: str, + *, + exclude: str | None = None, +) -> DayRecord | None: + """The day carrying the most photos and videos.""" + row = events.connection.execute(MEDIA_DAYS, (group_id, exclude or "")).fetchone() + return DayRecord(row["day"], row["media"]) if row else None + + +@dataclass(frozen=True, slots=True) +class Silence: + """A stretch of days on which nobody wrote anything.""" + + start: str + end: str + days: int + + +def longest_silence(days: dict[str, int], until: str) -> Silence | None: + """The longest run of consecutive silent days, up to and including ``until``. + + Walks the calendar rather than the recorded days, because the whole point + is the days that are *missing* from the series. Returns None when the + history is too short to contain a gap. + """ + if not days: + return None + spoke = set(days) + first = min(spoke) + best: Silence | None = None + run_start: str | None = None + day = first + + while day <= until: + if day in spoke: + if run_start is not None: + run = _silence(run_start, shift(day, -1)) + best = run if best is None or run.days > best.days else best + run_start = None + elif run_start is None: + run_start = day + day = shift(day, 1) + + if run_start is not None: + run = _silence(run_start, until) + best = run if best is None or run.days > best.days else best + return best + + +def _silence(start: str, end: str) -> Silence: + """Build a Silence, counting both ends.""" + length = 1 + day = start + while day < end: + day = shift(day, 1) + length += 1 + return Silence(start, end, length) + + +def silent_since(days: dict[str, int], uuid_days: dict[str, int], until: str) -> int: + """How many days somebody has gone without writing, as of ``until``. + + Counts back from ``until`` to their last message. Zero means they wrote + today; the group's own recorded days bound it, so a person who joined + yesterday is not reported as silent for a year. + """ + if not days: + return 0 + if not uuid_days: + return sum(1 for day in days if day <= until) + last = max(uuid_days) + return sum(1 for day in days if last < day <= until) diff --git a/signal_bot/stats/streaks.py b/signal_bot/stats/streaks.py new file mode 100644 index 0000000..bb7ee73 --- /dev/null +++ b/signal_bot/stats/streaks.py @@ -0,0 +1,126 @@ +# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. +"""Who holds the crown, and for how long. + +A tie shares the crown and keeps every tied streak alive. That is a decision, +not an oversight: breaking somebody's forty-day run on a technicality is the +kind of thing that starts an argument in a group chat, and "we were both first" +is what the people involved would say anyway. + +Streaks count *recorded* days, not calendar days. A day nobody spoke does not +end a run -- there was no crown to lose -- and neither does a day the bot was +offline for, which it could not have judged. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterable + + from signal_bot.stats.events import EventStore + +DAILY_COUNTS = """ + SELECT day, uuid, display_name, COUNT(*) AS messages + FROM message_events + WHERE group_id = ? AND day <> ? + GROUP BY day, uuid + ORDER BY day ASC, messages DESC, display_name ASC +""" + + +@dataclass(frozen=True, slots=True) +class Streak: + """A run of consecutive recorded days holding the crown.""" + + uuid: str + display_name: str + length: int + last_day: str + + +def daily_winners( + events: EventStore, + group_id: str, + *, + exclude: str | None = None, +) -> dict[str, list[tuple[str, str]]]: + """Who led each recorded day, as (uuid, display_name), ties included.""" + rows = events.connection.execute(DAILY_COUNTS, (group_id, exclude or "")).fetchall() + best: dict[str, int] = {} + winners: dict[str, list[tuple[str, str]]] = {} + for row in rows: + day, count = row["day"], row["messages"] + # Ordered messages-first, so the day's leader is always the row that + # creates the entry and everything after it either ties or loses. + if day not in best: + best[day] = count + winners[day] = [(row["uuid"], row["display_name"])] + elif count == best[day]: + winners[day].append((row["uuid"], row["display_name"])) + return winners + + +def current_streaks(winners: dict[str, list[tuple[str, str]]]) -> list[Streak]: + """Everyone whose run reaches the most recent recorded day. + + Sorted longest first, then by name, so two people on the same run are + reported in a stable order rather than whichever the dict happened to + yield. + """ + if not winners: + return [] + days = sorted(winners) + latest = days[-1] + holders = dict(winners[latest]) + + lengths = dict.fromkeys(holders, 0) + running = set(holders) + for day in reversed(days): + today = {uuid for uuid, _ in winners[day]} + for uuid in sorted(running): + if uuid in today: + lengths[uuid] += 1 + else: + # Their run ended on this day. What they had already + # accumulated is kept -- dropping them from the tally, as an + # earlier spelling did, reported every streak as zero. + running.discard(uuid) + if not running: + break + + return sorted( + (Streak(uuid, holders[uuid], length, latest) for uuid, length in lengths.items()), + key=lambda s: (-s.length, s.display_name), + ) + + +def longest_ever(winners: dict[str, list[tuple[str, str]]]) -> Streak | None: + """The longest crown run anybody has ever put together.""" + running: dict[str, int] = {} + names: dict[str, str] = {} + best: Streak | None = None + + for day in sorted(winners): + today = dict(winners[day]) + names.update(today) + running = {uuid: running.get(uuid, 0) + 1 for uuid in today} + for uuid, length in running.items(): + if ( + best is None + or length > best.length + or (length == best.length and names[uuid] < best.display_name) + ): + best = Streak(uuid, names[uuid], length, day) + return best + + +def held_days(winners: dict[str, list[tuple[str, str]]], uuid: str) -> int: + """How many days somebody has held the crown in total, ties included.""" + return sum(1 for holders in winners.values() if any(u == uuid for u, _ in holders)) + + +def names_of(holders: Iterable[tuple[str, str]]) -> list[str]: + """Display names of a day's crown holders, alphabetical for stability.""" + return sorted(name for _, name in holders) diff --git a/tests/stats_seed.py b/tests/stats_seed.py new file mode 100644 index 0000000..4a44159 --- /dev/null +++ b/tests/stats_seed.py @@ -0,0 +1,64 @@ +# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. +"""A deterministic history to run the period, record and streak queries over. + +Not a conftest fixture: it is a builder several test modules call with a +different history each, rather than one shared shape. Counts come from +arithmetic, never from ``random`` -- a fixture that changes between runs turns +a real failure into a flake. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from signal_bot.stats.events import MessageEvent + +if TYPE_CHECKING: + from signal_bot.store import Store + +GROUP = "group-a==" +MITHRA = ("u-ala", "Ala") +KUHY = ("u-kuhy", "kuhy") +IGA = ("u-iga", "Basia") + + +@dataclass(frozen=True, slots=True) +class Shape: + """The optional detail of a seeded day, so ``wrote`` stays legible.""" + + media: int = 0 + hour: int = 12 + chars: int = 10 + commands: int = 0 + group_id: str = GROUP + + +def wrote( + store: Store, + day: str, + who: tuple[str, str], + count: int, + shape: Shape | None = None, +) -> None: + """Put ``count`` messages from one person into one group on one day.""" + shape = shape or Shape() + uuid, name = who + for index in range(count): + store.events.record( + MessageEvent( + group_id=shape.group_id, + uuid=uuid, + display_name=name, + # Unique within (group, uuid) so the primary key does not + # absorb them, and stable so two runs agree. + sent_at_ms=int(day.replace("-", "")) * 1000 + index, + day=day, + local_hour=shape.hour, + media_kind="image" if index < shape.media else "", + media_count=1 if index < shape.media else 0, + char_len=shape.chars, + is_command=index < shape.commands, + ) + ) + store.events.note_person(shape.group_id, uuid, name, "", day) diff --git a/tests/test_stats_queries_period.py b/tests/test_stats_queries_period.py new file mode 100644 index 0000000..9a8e9b3 --- /dev/null +++ b/tests/test_stats_queries_period.py @@ -0,0 +1,121 @@ +# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. +"""Weeks, months, years, trends and the per-weekday picture.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from signal_bot.stats import periods +from signal_bot.stats.queries_period import ( + Trend, + daily_series, + every_day, + period_rows, + share_of_voice, + total_for, + weekday_buckets, +) +from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote + +if TYPE_CHECKING: + from signal_bot.store import Store + +# 2026-08-31 is a Monday, so this spans one ISO week exactly. +WEEK = periods.week_of("2026-09-01") +PREVIOUS = periods.previous(WEEK) + + +def history(store: Store) -> None: + """A full week, plus one day in the week before it.""" + wrote(store, "2026-08-25", KUHY, 8) + wrote(store, "2026-08-31", MITHRA, 10, Shape(media=2, commands=1)) + wrote(store, "2026-09-01", MITHRA, 20) + wrote(store, "2026-09-01", KUHY, 10) + wrote(store, "2026-09-02", IGA, 5) + + +def test_period_rows_are_ordered_by_volume(store: Store) -> None: + history(store) + rows = period_rows(store.events, GROUP, WEEK) + assert [(r.display_name, r.messages) for r in rows] == [ + ("Ala", 30), + ("kuhy", 10), + ("Basia", 5), + ] + + +def test_period_rows_carry_media_commands_and_length(store: Store) -> None: + history(store) + ala = period_rows(store.events, GROUP, WEEK)[0] + assert (ala.media, ala.commands, ala.chars) == (2, 1, 300) + + +def test_a_period_stops_at_its_own_boundary(store: Store) -> None: + """The 25th belongs to the previous week and must not leak in.""" + history(store) + assert total_for(store.events, GROUP, WEEK.start, WEEK.end) == 45 + assert total_for(store.events, GROUP, PREVIOUS.start, PREVIOUS.end) == 8 + + +def test_daily_series_omits_days_nobody_spoke(store: Store) -> None: + """Absent, not zero: the caller decides what a missing day means.""" + history(store) + series = daily_series(store.events, GROUP, WEEK.start, WEEK.end) + assert series == {"2026-08-31": 10, "2026-09-01": 30, "2026-09-02": 5} + + +def test_every_day_spans_the_whole_history(store: Store) -> None: + history(store) + assert set(every_day(store.events, GROUP)) == { + "2026-08-25", + "2026-08-31", + "2026-09-01", + "2026-09-02", + } + + +def test_weekday_buckets_index_monday_as_zero(store: Store) -> None: + history(store) + buckets = weekday_buckets(every_day(store.events, GROUP)) + # 2026-08-25 is a Tuesday, 08-31 a Monday, 09-01 a Tuesday, 09-02 a Wednesday. + assert buckets[0] == [10] + assert sorted(buckets[1]) == [8, 30] + assert buckets[2] == [5] + + +def test_trend_reports_direction_and_size() -> None: + trend = Trend(now=45, before=30) + assert (trend.delta, trend.percent, trend.arrow) == (15, 50, "↑") + down = Trend(now=15, before=30) + assert (down.delta, down.percent, down.arrow) == (-15, -50, "↓") + + +def test_a_flat_period_gets_no_arrow() -> None: + assert Trend(now=30, before=30).arrow == "" + assert Trend(now=30, before=30).percent == 0 + + +def test_no_baseline_is_not_the_same_as_no_change() -> None: + """The first week ever has nothing to compare against, and says so.""" + first = Trend(now=45, before=0) + assert first.percent is None + assert first.arrow == "" + assert first.delta == 45 + + +def test_share_of_voice(store: Store) -> None: + history(store) + rows = period_rows(store.events, GROUP, WEEK) + assert share_of_voice(rows) == [("Ala", 67), ("kuhy", 22), ("Basia", 11)] + + +def test_share_of_voice_of_nothing_is_empty() -> None: + assert share_of_voice([]) == [] + + +def test_the_backfilled_day_can_be_left_out_of_every_query(store: Store) -> None: + history(store) + assert total_for(store.events, GROUP, WEEK.start, WEEK.end, exclude="2026-09-01") == 15 + assert "2026-09-01" not in every_day(store.events, GROUP, exclude="2026-09-01") + rows = period_rows(store.events, GROUP, WEEK, exclude="2026-09-01") + assert [r.display_name for r in rows] == ["Ala", "Basia"] diff --git a/tests/test_stats_records.py b/tests/test_stats_records.py new file mode 100644 index 0000000..be25cb8 --- /dev/null +++ b/tests/test_stats_records.py @@ -0,0 +1,135 @@ +# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. +"""All-time records, and the silences between them.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from signal_bot.stats.queries_period import every_day +from signal_bot.stats.records import ( + busiest_day, + longest_silence, + most_media_day, + personal_best, + previous_best, + quietest_day, + silent_since, +) +from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote + +if TYPE_CHECKING: + from signal_bot.store import Store + + +def history(store: Store) -> None: + """Four days, a two-day gap, ties, and somebody who stops writing.""" + wrote(store, "2026-09-01", MITHRA, 10, Shape(media=3)) + wrote(store, "2026-09-01", KUHY, 5) + wrote(store, "2026-09-02", MITHRA, 4) + wrote(store, "2026-09-02", KUHY, 4) + wrote(store, "2026-09-03", IGA, 2) + # 09-04 and 09-05 silent. + wrote(store, "2026-09-06", KUHY, 9, Shape(media=1)) + + +def test_no_history_yields_no_records(store: Store) -> None: + """Day one of this release: everything empty, nothing crashing.""" + days = every_day(store.events, GROUP) + assert busiest_day(days) is None + assert quietest_day(days) is None + assert longest_silence(days, "2026-09-06") is None + assert most_media_day(store.events, GROUP) is None + assert personal_best(store.events, GROUP, "u-kuhy") is None + + +def test_busiest_and_quietest(store: Store) -> None: + history(store) + days = every_day(store.events, GROUP) + assert busiest_day(days).day == "2026-09-01" + assert busiest_day(days).value == 15 + assert quietest_day(days).day == "2026-09-03" + assert quietest_day(days).value == 2 + + +def test_the_quietest_day_is_one_somebody_spoke_on(store: Store) -> None: + """A silent day is a silence, not a quiet day, and is measured separately.""" + history(store) + days = every_day(store.events, GROUP) + assert quietest_day(days).value > 0 + assert "2026-09-04" not in days + + +def test_a_tie_on_the_busiest_day_goes_to_the_earlier(store: Store) -> None: + wrote(store, "2026-09-02", KUHY, 7) + wrote(store, "2026-09-01", MITHRA, 7) + assert busiest_day(every_day(store.events, GROUP)).day == "2026-09-01" + + +def test_personal_best_and_what_it_beat(store: Store) -> None: + history(store) + best = personal_best(store.events, GROUP, "u-kuhy") + assert (best.day, best.value) == ("2026-09-06", 9) + previous = previous_best(store.events, GROUP, "u-kuhy", "2026-09-06") + assert (previous.day, previous.value) == ("2026-09-01", 5) + + +def test_a_first_ever_day_has_nothing_to_beat(store: Store) -> None: + """The "poprzedni rekord" clause has to be omitted, not printed as 0.""" + wrote(store, "2026-09-01", IGA, 3) + assert previous_best(store.events, GROUP, "u-iga", "2026-09-01") is None + + +def test_most_media_day(store: Store) -> None: + history(store) + record = most_media_day(store.events, GROUP) + assert (record.day, record.value) == ("2026-09-01", 3) + + +def test_a_history_with_no_media_has_no_media_record(store: Store) -> None: + wrote(store, "2026-09-01", KUHY, 4) + assert most_media_day(store.events, GROUP) is None + + +def test_longest_silence(store: Store) -> None: + history(store) + gap = longest_silence(every_day(store.events, GROUP), "2026-09-06") + assert (gap.start, gap.end, gap.days) == ("2026-09-04", "2026-09-05", 2) + + +def test_a_silence_running_to_today_counts(store: Store) -> None: + """The gap the group is currently in is the one worth mentioning.""" + wrote(store, "2026-09-01", KUHY, 1) + gap = longest_silence(every_day(store.events, GROUP), "2026-09-05") + assert (gap.start, gap.end, gap.days) == ("2026-09-02", "2026-09-05", 4) + + +def test_an_unbroken_history_has_no_silence(store: Store) -> None: + wrote(store, "2026-09-01", KUHY, 1) + wrote(store, "2026-09-02", KUHY, 1) + assert longest_silence(every_day(store.events, GROUP), "2026-09-02") is None + + +def test_silent_since_counts_recorded_days_only(store: Store) -> None: + """Bounded by the group's own history, so a newcomer is not "silent for a year".""" + history(store) + days = every_day(store.events, GROUP) + ala = {d: n for d, n in days.items() if d in {"2026-09-01", "2026-09-02"}} + assert silent_since(days, ala, "2026-09-06") == 2 + assert silent_since(days, {"2026-09-06": 9}, "2026-09-06") == 0 + + +def test_somebody_who_never_wrote_is_silent_for_the_whole_history(store: Store) -> None: + history(store) + days = every_day(store.events, GROUP) + assert silent_since(days, {}, "2026-09-06") == len(days) + assert silent_since({}, {}, "2026-09-06") == 0 + + +def test_the_backfilled_day_is_kept_out_of_records(store: Store) -> None: + """A 22-hour window filed as a day must not become somebody's best.""" + wrote(store, "2026-08-30", MITHRA, 98) + wrote(store, "2026-09-01", MITHRA, 10) + assert personal_best(store.events, GROUP, "u-ala").value == 98 + excluded = personal_best(store.events, GROUP, "u-ala", exclude="2026-08-30") + assert (excluded.day, excluded.value) == ("2026-09-01", 10) + assert "2026-08-30" not in every_day(store.events, GROUP, exclude="2026-08-30") diff --git a/tests/test_stats_streaks.py b/tests/test_stats_streaks.py new file mode 100644 index 0000000..da49da6 --- /dev/null +++ b/tests/test_stats_streaks.py @@ -0,0 +1,118 @@ +# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. +"""The crown: who holds it, for how long, and what a tie does to a run.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from signal_bot.stats.streaks import ( + current_streaks, + daily_winners, + held_days, + longest_ever, + names_of, +) +from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, wrote + +if TYPE_CHECKING: + from signal_bot.store import Store + + +def test_no_history_has_no_crown(store: Store) -> None: + winners = daily_winners(store.events, GROUP) + assert winners == {} + assert current_streaks(winners) == [] + assert longest_ever(winners) is None + assert held_days(winners, "u-kuhy") == 0 + + +def test_the_busiest_person_holds_the_day(store: Store) -> None: + wrote(store, "2026-09-01", MITHRA, 10) + wrote(store, "2026-09-01", KUHY, 5) + winners = daily_winners(store.events, GROUP) + assert names_of(winners["2026-09-01"]) == ["Ala"] + + +def test_a_tie_shares_the_crown(store: Store) -> None: + """kuhy's call: shared, because the alternative starts arguments.""" + wrote(store, "2026-09-01", MITHRA, 4) + wrote(store, "2026-09-01", KUHY, 4) + wrote(store, "2026-09-01", IGA, 1) + winners = daily_winners(store.events, GROUP) + assert names_of(winners["2026-09-01"]) == ["Ala", "kuhy"] + + +def test_a_tie_keeps_every_tied_run_alive(store: Store) -> None: + """Three days, both tied throughout: both hold a three-day streak.""" + for day in ("2026-09-01", "2026-09-02", "2026-09-03"): + wrote(store, day, MITHRA, 3) + wrote(store, day, KUHY, 3) + streaks = current_streaks(daily_winners(store.events, GROUP)) + assert [(s.display_name, s.length) for s in streaks] == [("Ala", 3), ("kuhy", 3)] + + +def test_a_run_survives_a_day_nobody_spoke(store: Store) -> None: + """There was no crown to lose on a silent day, so the run continues.""" + wrote(store, "2026-09-01", KUHY, 5) + # 09-02 and 09-03 silent. + wrote(store, "2026-09-04", KUHY, 5) + streaks = current_streaks(daily_winners(store.events, GROUP)) + assert [(s.display_name, s.length) for s in streaks] == [("kuhy", 2)] + + +def test_losing_a_day_ends_a_run_but_keeps_what_it_reached(store: Store) -> None: + """The bug this test exists for reported every streak as zero.""" + wrote(store, "2026-09-01", MITHRA, 10) + wrote(store, "2026-09-02", MITHRA, 10) + wrote(store, "2026-09-03", MITHRA, 4) + wrote(store, "2026-09-03", KUHY, 4) + wrote(store, "2026-09-06", KUHY, 9) + streaks = current_streaks(daily_winners(store.events, GROUP)) + assert [(s.display_name, s.length) for s in streaks] == [("kuhy", 2)] + assert streaks[0].last_day == "2026-09-06" + + +def test_longest_ever_survives_the_run_that_set_it_ending(store: Store) -> None: + wrote(store, "2026-09-01", MITHRA, 10) + wrote(store, "2026-09-02", MITHRA, 10) + wrote(store, "2026-09-03", MITHRA, 10) + wrote(store, "2026-09-04", KUHY, 10) + best = longest_ever(daily_winners(store.events, GROUP)) + assert (best.display_name, best.length, best.last_day) == ("Ala", 3, "2026-09-03") + + +def test_held_days_counts_shared_days_for_everyone_tied(store: Store) -> None: + wrote(store, "2026-09-01", MITHRA, 10) + wrote(store, "2026-09-02", MITHRA, 4) + wrote(store, "2026-09-02", KUHY, 4) + winners = daily_winners(store.events, GROUP) + assert held_days(winners, "u-ala") == 2 + assert held_days(winners, "u-kuhy") == 1 + assert held_days(winners, "u-iga") == 0 + + +def test_somebody_who_merely_took_part_holds_nothing(store: Store) -> None: + """Neither leader nor tied: the common case, and the one worth pinning.""" + wrote(store, "2026-09-01", MITHRA, 10) + wrote(store, "2026-09-01", KUHY, 4) + wrote(store, "2026-09-01", IGA, 1) + winners = daily_winners(store.events, GROUP) + assert names_of(winners["2026-09-01"]) == ["Ala"] + assert held_days(winners, "u-iga") == 0 + + +def test_streak_order_is_stable(store: Store) -> None: + """Longest first, then by name -- not whichever the dict happened to yield.""" + wrote(store, "2026-09-01", MITHRA, 3) + wrote(store, "2026-09-01", KUHY, 3) + wrote(store, "2026-09-01", IGA, 3) + streaks = current_streaks(daily_winners(store.events, GROUP)) + assert [s.display_name for s in streaks] == ["Basia", "Ala", "kuhy"] + + +def test_the_backfilled_day_can_be_left_out(store: Store) -> None: + wrote(store, "2026-08-30", MITHRA, 98) + wrote(store, "2026-09-01", KUHY, 5) + winners = daily_winners(store.events, GROUP, exclude="2026-08-30") + assert "2026-08-30" not in winners + assert [s.display_name for s in current_streaks(winners)] == ["kuhy"]