signal-bot/signal_bot/stats/render_extra.py
Krzysztof kuhy Rudnicki bfc2bcb3b8 feat: nine commands for asking the statistics questions
!staty, !wczoraj, !tydzień, !miesiąc, !rok, !rekordy, !korona, !dni and
!ja, under 29 triggers with both Polish spellings of each.

None takes an argument. registry.resolve matches the whole folded body
against a trigger, so "!ja ala" resolves to nothing; teaching it to
split on whitespace would silently make "!kot pls" fire where today it
does nothing. Each question gets its own trigger instead.

!wczoraj renders through the same code path as the nightly post, which
is not a convenience: the bot cannot trigger its own 21:37 report, so
without byte-identical output there is no way to check the headline
feature of this release short of waiting for the evening.

Two guards worth naming:

- test_every_trigger_is_unique builds the whole registry.
  DuplicateTriggerError subclasses ValueError, and ValueError is in
  startup_retry.RETRYABLE -- so a colliding alias would not crash the
  bot, it would retry forever logging "signal-cli not reachable yet".
  Checked at test time instead of never.
- No trigger may contain a digit: fold() maps 0134578 onto letters, so
  "!7dni" would resolve as "!tdni". Also asserted.

Every command answers on an empty history rather than crashing, which is
the state the group will actually be in for the next few days: records,
the crown and the weekday table all say "za mało danych" because their
only day so far is the backfilled one, and they exclude it.

Verified against a copy of the live database -- all nine render
correctly on the real group's data.

One test bug caught while writing it: it seeded "today" and excluded
2026-08-30, which on the day it was written were the same date, so it
passed for the wrong reason. Fixed dates now.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015d48t4qsQLBJnDwDeo2Nom
2026-08-30 19:49:41 +02:00

147 lines
5.3 KiB
Python

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The on-demand answers: records, the crown, one person, the weekdays.
Each of these is asked for rather than pushed, so each says something even
when there is nothing interesting yet -- "za mało danych" is a useful answer
to a question, where it would be noise in a nightly post nobody asked for.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Final
from signal_bot.stats import records, streaks, text
from signal_bot.stats.periods import WEEKDAY_PLURAL, pl_day_month
if TYPE_CHECKING:
from signal_bot.stats.events import EventStore, Person
from signal_bot.stats.records import DayRecord
TROPHY: Final = "\U0001f3c6"
CROWN: Final = "\U0001f451"
CALENDAR: Final = "\U0001f4c5"
CAMERA: Final = "\U0001f4f7"
QUIET: Final = "\U0001f92b"
NOTHING_YET: Final = "Za mało danych — wróć za kilka dni."
def _record_line(label: str, record: DayRecord | None, unit: str) -> str | None:
"""One "label: value (date)" line, or None when there is no record yet."""
if record is None:
return None
return f"{label}: {record.value} {unit} ({pl_day_month(record.day)})"
def render_records(
history: dict[str, int],
media: DayRecord | None,
silence: records.Silence | None,
personal: list[tuple[str, DayRecord]],
) -> str:
"""Every all-time record the group has set."""
lines = [f"{TROPHY} Rekordy"]
candidates = [
_record_line("Najwięcej w jeden dzień", records.busiest_day(history), "wiadomości"),
_record_line("Najciszej (ale ktoś pisał)", records.quietest_day(history), "wiadomości"),
_record_line(f"Najwięcej zdjęć {CAMERA}", media, "zdjęć"),
]
lines.extend(line for line in candidates if line)
if silence:
lines.append(
f"{QUIET} Najdłuższa cisza: {text.days(silence.days)} "
f"({pl_day_month(silence.start)} - {pl_day_month(silence.end)})"
)
if personal:
lines.append("")
lines.append("Rekordy życiowe:")
lines.extend(
f" {name} — {record.value} ({pl_day_month(record.day)})" for name, record in personal
)
return "\n".join(lines) if len(lines) > 1 else NOTHING_YET
def render_crown(
winners: dict[str, list[tuple[str, str]]],
people: list[Person],
) -> str:
"""Who holds the crown now, who held it longest, and who has held it most."""
if not winners:
return NOTHING_YET
# Both are guaranteed once winners is non-empty, which the guard above
# already established: the latest day always has at least one holder, and
# a day with a holder is a run of at least one.
current = streaks.current_streaks(winners)
# `or current[0]` is a type narrowing, not a fallback: longest_ever only
# returns None for an empty history, which the guard above ruled out.
longest = streaks.longest_ever(winners) or current[0]
names = " i ".join(s.display_name for s in current if s.length == current[0].length)
lines = [
f"{CROWN} Korona",
f"Teraz: {names} — {text.days(current[0].length)} z rzędu",
(
f"Najdłuższa seria: {longest.display_name} — {text.days(longest.length)} "
f"(do {pl_day_month(longest.last_day)})"
),
]
held = sorted(
((person.display_name, streaks.held_days(winners, person.uuid)) for person in people),
key=lambda pair: (-pair[1], pair[0]),
)
lines.append("")
lines.append("Dni z koroną łącznie:")
lines.extend(f" {name} — {text.days(days)}" for name, days in held if days)
return "\n".join(lines)
def render_weekdays(buckets: dict[int, list[int]], today: int) -> str:
"""The Monday-to-Sunday picture, with today's weekday called out."""
if not buckets:
return NOTHING_YET
lines = [f"{CALENDAR} Dni tygodnia"]
for index in range(len(WEEKDAY_PLURAL)):
days = buckets.get(index)
if not days:
continue
mean = round(sum(days) / len(days))
marker = " ← dziś" if index == today else ""
lines.append(
f"{WEEKDAY_PLURAL[index]}: średnio {mean} ({text.days(len(days))} w historii){marker}"
)
return "\n".join(lines)
def render_me(
name: str,
total: int,
best: DayRecord | None,
share: int,
favourite_hour: int | None,
streak: int,
) -> str:
"""One person's own numbers."""
lines = [f"{name} — {text.messages(total)} w historii"]
if best:
lines.append(f"{TROPHY} Rekord życiowy: {best.value} ({pl_day_month(best.day)})")
lines.append(f"Udział w rozmowie: {share}%")
if favourite_hour is not None:
lines.append(f"Najczęściej piszesz o {favourite_hour:02d}:00")
if streak:
lines.append(f"{CROWN} Korona: {text.days(streak)} z rzędu")
return "\n".join(lines)
def favourite_hour(events: EventStore, group_id: str, uuid: str) -> int | None:
"""The hour somebody writes most in, or None if they never have."""
row = events.connection.execute(
"""
SELECT local_hour, COUNT(*) AS messages
FROM message_events
WHERE group_id = ? AND uuid = ?
GROUP BY local_hour
ORDER BY messages DESC, local_hour ASC
LIMIT 1
""",
(group_id, uuid),
).fetchone()
return row["local_hour"] if row else None