signal-bot/signal_bot/stats/render_extra.py

# 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"
HOURS_IN_A_DAY: Final = 24
# Every command says what it is, whether or not it has data yet. Somebody
# typing !korona on day one used to get "za mało danych" and nothing else,
# which tells them neither what the command does nor when it will work.
TITLES: Final = {
    "crown": f"{CROWN} Korona — kto danego dnia napisał najwięcej wiadomości",
    "records": f"{TROPHY} Rekordy — najlepsze i najgorsze dni w historii grupy",
    "weekdays": f"{CALENDAR} Dni tygodnia — ile średnio piszecie w każdy z nich",
}
# What each will show once there is enough history, so the empty answer is
# still an answer.
PROMISES: Final = {
    "crown": (
        "Pokażę: kto ma koronę teraz, jak długą serię, kto miał najdłuższą "
        "w historii i ile dni każdy ją nosił."
    ),
    "records": (
        "Pokażę: najgadatliwszy dzień, najcichszy, dzień z największą liczbą "
        "zdjęć, najdłuższą ciszę i rekord życiowy każdego z was."
    ),
    "weekdays": (
        "Pokażę: średnią liczbę wiadomości w poniedziałki, wtorki i tak dalej "
        "— czyli czy weekendy są gadatliwsze od środy."
    ),
}


def nothing_yet(what: str, since: str | None = None) -> str:
    """The empty answer, which still says what the command is for.

    A bare "za mało danych" tells somebody neither what they asked for nor
    when it will work, which is exactly what happened within minutes of
    0.1.0 going out.
    """
    started = f"Liczę od {pl_day_month(since)}. " if since else ""
    return "\n".join([TITLES[what], "", f"{started}Jeszcze za mało danych.", PROMISES[what]])


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}: {text.marked(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]],
    since: str | None = None,
) -> str:
    """Every all-time record the group has set."""
    lines = [TITLES["records"]]
    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} — {text.marked(record.value)} ({pl_day_month(record.day)})"
            for name, record in personal
        )
    return "\n".join(lines) if len(lines) > 1 else nothing_yet("records", since)


def render_crown(
    winners: dict[str, list[tuple[str, str]]],
    people: list[Person],
    since: str | None = None,
) -> str:
    """Who holds the crown now, who held it longest, and who has held it most."""
    if not winners:
        return nothing_yet("crown", since)
    # 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 = [
        TITLES["crown"],
        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, since: str | None = None) -> str:
    """The Monday-to-Sunday picture, with today's weekday called out."""
    if not buckets:
        return nothing_yet("weekdays", since)
    lines = [TITLES["weekdays"]]
    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)


ME_TITLE: Final = "Twoje liczby w tej grupie"


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 = [ME_TITLE, "", f"{name} — {text.messages(total)} w historii"]
    if best:
        lines.append(
            f"{TROPHY} Rekord życiowy: {text.marked(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 między {favourite_hour:02d}:00 a "
            f"{(favourite_hour + 1) % HOURS_IN_A_DAY: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 when nothing is known.

    Skips rows whose hour is UNKNOWN_HOUR. Those are backfilled from a counter
    that had no timestamps, and counting them told every member of the group
    that they write most at noon.
    """
    row = events.connection.execute(
        """
        SELECT local_hour, COUNT(*) AS messages
        FROM message_events
        WHERE group_id = ? AND uuid = ? AND local_hour >= 0
        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