signal-bot/signal_bot/manage/analytics_panels.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""One gatherer per panel on the Analytics tab.

Split out of ``analytics_read`` so both stay under the 250-line cap. That
module owns the connection and the shapes; this one owns the six questions,
one function each, mirroring the six query modules exactly so that a panel
and its query are never more than one hop apart.

Every response carries ``window``. The bot is days old, not months, and a
month panel drawn from three days is only honest if the reader can see how
many days it is.
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from signal_bot.analytics import (
    adoption,
    content,
    demand,
    health,
    hours,
    sessions,
    trend,
    usage,
)
from signal_bot.analytics.scope import window
from signal_bot.manage.analytics_read import (
    Gathered,
    empty,
    measure,
    names_in,
    safely,
)
from signal_bot.stats.periods import today

if TYPE_CHECKING:
    import sqlite3
    from pathlib import Path

    from signal_bot.analytics.scope import Scope

# How far the retention curve is drawn. Two weeks: long enough to show a
# drop-off, short enough that most of it is not empty on a young bot.
RETENTION_DAYS = 14
# The window either side of a release that "what did this change" compares.
RELEASE_SPAN_DAYS = 3


def overview(path: Path, scope: Scope) -> Gathered:
    """Family A: what is used, by whom, and what has never been touched."""

    def gather(conn: sqlite3.Connection) -> Gathered:
        names = names_in(conn, scope)
        uses = usage.by_command(conn, scope)
        return {
            "window": measure(conn, scope),
            "totals": usage.totals(conn, scope),
            "commands": [
                {
                    "command": use.command,
                    "uses": use.uses,
                    "people": use.people,
                    "ok": use.ok,
                    "refused": use.refused,
                    "errors": use.errors,
                    "first_day": use.first_day,
                    "last_day": use.last_day,
                }
                for use in uses
            ],
            "people": [
                {
                    "uuid": person.uuid,
                    "name": names.get(person.uuid, ""),
                    "uses": person.uses,
                    "commands": person.commands,
                    "favourite": person.favourite,
                }
                for person in usage.by_person(conn, scope)
            ],
            "matrix": [
                {"uuid": uuid, "name": names.get(uuid, ""), "command": command, "uses": count}
                for uuid, command, count in usage.matrix(conn, scope)
            ],
        }

    return safely(path, gather) or {**empty("commands", "people", "matrix"), "totals": {}}


def times(path: Path, scope: Scope) -> Gathered:
    """Family B: hours, weekdays, days, months and what has gone quiet."""

    def gather(conn: sqlite3.Connection) -> Gathered:
        series = trend.dense_days(trend.by_day(conn, scope))
        measured = window(conn, scope)
        peak = hours.peak(hours.by_hour(conn, scope))
        return {
            "window": measure(conn, scope),
            "hours": hours.by_hour(conn, scope),
            "weekday_hours": hours.by_weekday_hour(conn, scope),
            "by_command_hour": hours.by_command_hour(conn, scope),
            "days": [{"day": day, "uses": uses} for day, uses in series],
            "months": [
                {"month": month, "uses": uses, "days": days}
                for month, uses, days in trend.by_month(conn, scope)
            ],
            "peak": {
                "busiest": peak.busiest,
                "busiest_uses": peak.busiest_uses,
                "quietest": peak.quietest,
                "quietest_uses": peak.quietest_uses,
            },
            "direction": trend.direction(series, measured),
            "dying": [
                {"command": one.command, "last_day": one.last_day, "idle_days": one.idle_days}
                for one in trend.dying(conn, scope, today())
            ],
        }

    return safely(path, gather) or empty("hours", "days", "months", "dying")


def wellbeing(path: Path, scope: Scope) -> Gathered:
    """Family C: what is slow, what refuses people, and what upstream does."""

    def gather(conn: sqlite3.Connection) -> Gathered:
        return {
            "window": measure(conn, scope),
            "latencies": [
                {
                    "command": one.command,
                    "samples": one.samples,
                    "handler_p50": one.handler_p50,
                    "handler_p95": one.handler_p95,
                    "send_p50": one.send_p50,
                    "upstream_p50": one.upstream_p50,
                }
                for one in health.latencies(conn, scope)
            ],
            "gates": [
                {"command": command, "gate": gate, "hits": hits}
                for command, gate, hits in health.gates(conn, scope)
            ],
            "errors": [
                {"command": command, "error_class": name, "hits": hits}
                for command, name, hits in health.errors(conn, scope)
            ],
            "upstream": [
                {
                    "host": one.host,
                    "calls": one.calls,
                    "failures": one.failures,
                    "p50_ms": one.p50_ms,
                }
                for one in health.upstreams(conn, scope)
            ],
        }

    return safely(path, gather) or empty("latencies", "gates", "errors", "upstream")


def wishes(path: Path, scope: Scope, triggers: tuple[str, ...]) -> Gathered:
    """Family D: what people nearly typed, grouped by what they nearly hit."""

    def gather(conn: sqlite3.Connection) -> Gathered:
        found = demand.collect(conn, scope, triggers)
        return {
            "window": measure(conn, scope),
            "near_misses": [
                {"target": miss.target, "hits": miss.hits, "variants": miss.variants}
                for miss in found.near_misses
            ],
            "unclaimed": found.unclaimed,
            "unclaimed_words": found.unclaimed_words,
            "filed": found.filed,
        }

    return safely(path, gather) or empty("near_misses")


def behaviour(path: Path, scope: Scope, releases: tuple[tuple[str, str], ...]) -> Gathered:
    """Family E: sittings, retention, releases and share of chat."""

    def gather(conn: sqlite3.Connection) -> Gathered:
        series = trend.dense_days(trend.by_day(conn, scope))
        sitting = sessions.collect(conn, scope)
        share = adoption.share_of_chat(conn, scope)
        return {
            "window": measure(conn, scope),
            "sessions": {
                "count": sitting.count,
                "commands": sitting.commands,
                "longest": sitting.longest,
                "average_length": sitting.average_length,
            },
            "pairs": [
                {"left": left, "right": right, "hits": hits} for left, right, hits in sitting.pairs
            ],
            "repeats": [{"command": command, "extra": extra} for command, extra in sitting.repeats],
            "retention": [
                {"offset": offset, "active": active, "cohort": cohort}
                for offset, active, cohort in adoption.retention(conn, scope, RETENTION_DAYS)
            ],
            "newcomers": [
                {"day": day, "people": people} for day, people in adoption.newcomers(conn, scope)
            ],
            "releases": [
                {"version": version, "day": day, "after": after, "before": before}
                for version, day, after, before in adoption.around_releases(
                    series, releases, RELEASE_SPAN_DAYS
                )
            ],
            "share": {
                "messages": share.messages,
                "commands": share.commands,
                "fraction": share.fraction,
            },
        }

    return safely(path, gather) or empty("pairs", "repeats", "retention", "newcomers", "releases")


def items(path: Path, scope: Scope) -> Gathered:
    """Family F: how varied each pool actually is."""

    def gather(conn: sqlite3.Connection) -> Gathered:
        return {
            "window": measure(conn, scope),
            "pools": [
                {
                    "command": pool.command,
                    "draws": pool.draws,
                    "distinct": pool.distinct,
                    "repeat_rate": pool.repeat_rate,
                }
                for pool in content.pools(conn, scope)
            ],
            "repeats": [
                {"command": command, "item_id": item, "draws": draws}
                for command, item, draws in content.repeats(conn, scope)
            ],
        }

    return safely(path, gather) or empty("pools", "repeats")