signal-bot/signal_bot/suggestions.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""What the group wished the bot could do, kept where git cannot reach it.

Two tables and one deliberate asymmetry between them. ``suggestions`` holds
notes somebody wrote *on purpose*, with !zaproponuj, knowing they were
writing to whoever maintains the bot. ``unknown_commands`` holds every "!"
word the registry did not recognise, body and all, which is a far less
considered kind of text -- it is full of half-typed names, and a misspelled
name is one the personal-data gate cannot know to refuse.

That asymmetry is why this lives in SQLite rather than in a tracked file.
The database is gitignored; the export the management page offers carries
the deliberate notes in full and the unrecognised words as counts only.

Shares the Store's connection, the same way stats.events.EventStore does.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    import sqlite3


@dataclass(frozen=True, slots=True)
class Suggestion:
    """One idea somebody filed with !zaproponuj."""

    group_id: str
    command: str
    body: str
    sent_at_ms: int


@dataclass(frozen=True, slots=True)
class UnknownTally:
    """One unrecognised word, how often it was tried, and what was typed."""

    group_id: str
    word: str
    hits: int
    last_seen: int
    bodies: tuple[str, ...]


class SuggestionStore:
    """Reads and writes the wish list. Shares the Store's connection."""

    def __init__(self, conn: sqlite3.Connection) -> None:
        """Wrap an open connection whose schema has already been applied."""
        self._conn = conn

    def record_unknown(self, group_id: str, word: str, body: str, sent_at_ms: int) -> None:
        """Note one unrecognised "!" word, ignoring a redelivery of it.

        ``OR IGNORE`` against the envelope's own timestamp, not the wall
        clock. signal-cli replays undelivered frames when the bot
        reconnects; keyed on a clock reading every replay looks new, and
        without OR IGNORE a genuine collision raises IntegrityError out of
        the dispatcher, where there is nothing above it to catch one.
        """
        self._conn.execute(
            "INSERT OR IGNORE INTO unknown_commands"
            " (group_id, word, body, sent_at_ms) VALUES (?, ?, ?, ?)",
            (group_id, word, body, sent_at_ms),
        )
        self._conn.commit()

    def record_suggestion(self, group_id: str, command: str, body: str, sent_at_ms: int) -> None:
        """File one idea against the command name it is asking for."""
        self._conn.execute(
            "INSERT OR IGNORE INTO suggestions"
            " (group_id, command, body, sent_at_ms) VALUES (?, ?, ?, ?)",
            (group_id, command, body, sent_at_ms),
        )
        self._conn.commit()

    def filed(self) -> list[Suggestion]:
        """Every filed suggestion, newest first."""
        rows = self._conn.execute(
            "SELECT group_id, command, body, sent_at_ms FROM suggestions"
            " ORDER BY sent_at_ms DESC, command ASC"
        )
        return [
            Suggestion(
                group_id=str(row["group_id"]),
                command=str(row["command"]),
                body=str(row["body"]),
                sent_at_ms=int(row["sent_at_ms"]),
            )
            for row in rows
        ]

    def unknown(self) -> list[UnknownTally]:
        """Every unrecognised word, most-tried first, with what was typed.

        Two passes rather than one grouped query with ``group_concat``: a
        body may itself contain the separator, and a tally that silently
        splits somebody's sentence in half is worse than a second SELECT.
        """
        tallies = self._conn.execute(
            "SELECT group_id, word, COUNT(*) AS hits, MAX(sent_at_ms) AS last_seen"
            " FROM unknown_commands GROUP BY group_id, word"
            " ORDER BY hits DESC, last_seen DESC, word ASC"
        ).fetchall()
        return [
            UnknownTally(
                group_id=str(row["group_id"]),
                word=str(row["word"]),
                hits=int(row["hits"]),
                last_seen=int(row["last_seen"]),
                bodies=self._bodies(str(row["group_id"]), str(row["word"])),
            )
            for row in tallies
        ]

    def _bodies(self, group_id: str, word: str) -> tuple[str, ...]:
        """The distinct messages that produced one unrecognised word."""
        rows = self._conn.execute(
            "SELECT DISTINCT body FROM unknown_commands"
            " WHERE group_id = ? AND word = ? ORDER BY body ASC",
            (group_id, word),
        )
        return tuple(str(row["body"]) for row in rows)