signal-bot/signal_bot/manage/routes_suggestions.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""What the group asked for, read back on the management page.

Read-only, and connected read-only: ``mode=ro`` rather than a plain connect,
which is the same URI ``scripts/personal_data_sources.py`` uses. Two reasons.
The bot has this file open and commits on every inbound message, so a second
writer is a lock waiting to happen; and ``Store`` applies its schema on open,
which drops the ephemeral tables whenever the versions disagree -- a
management page must never be able to do that to a running bot.

The export deliberately does **not** write into the repository. See
``render_export``.
"""

from __future__ import annotations

import contextlib
import sqlite3
from typing import TYPE_CHECKING, Any, Final

from fastapi import APIRouter

from signal_bot.state_dir import state_dir
from signal_bot.suggestions import SuggestionStore

if TYPE_CHECKING:
    from pathlib import Path

    from signal_bot.manage.deps import ManageDeps

# Named for the repository's markdown convention, so that a copy pasted
# into the tree is already compliant -- but written outside it. See
# default_export_path.
EXPORT_NAME: Final = "DOCS-suggestions.md"


def default_export_path() -> Path:
    """Where the export goes: the state directory, never the repository."""
    return state_dir() / EXPORT_NAME


# What a body is truncated to in the export. Long enough for a sentence,
# short enough that a pasted paragraph cannot bury the file.
EXPORT_BODY_LIMIT = 200

HEADER = """# Suggestions

Written by the management page, from the bot's own database. Not generated
at build time and not checked by anything -- read it, act on what is worth
acting on, and delete it.

Bodies are included for deliberate `!zaproponuj` notes only. Unrecognised
commands are counts, because that half of the table is full of half-typed
words and a half-typed name is one the personal-data gate cannot recognise.
"""


def collect(path: Path) -> dict[str, list[dict[str, Any]]]:
    """Everything the page shows, or empty lists when there is nothing to read.

    Opening and reading are inside one try because every way this can fail
    is the same answer -- a bot that has never run, a database older than
    these tables, a half-written file -- and none of them is worth breaking
    the page over. The page is most useful exactly when something is wrong.

    Closed explicitly: the bot holds this file open and commits on every
    inbound message, so a reader left dangling is a lock waiting to happen.
    """
    if not path.is_file():
        return {"suggestions": [], "unknown": []}
    try:
        with contextlib.closing(sqlite3.connect(f"file:{path}?mode=ro", uri=True)) as conn:
            conn.row_factory = sqlite3.Row
            store = SuggestionStore(conn)
            filed = store.filed()
            unknown = store.unknown()
    except sqlite3.Error:
        return {"suggestions": [], "unknown": []}
    return {
        "suggestions": [
            {
                "command": one.command,
                "body": one.body,
                "sent_at_ms": one.sent_at_ms,
                "group_id": one.group_id,
            }
            for one in filed
        ],
        "unknown": [
            {
                "word": one.word,
                "hits": one.hits,
                "last_seen": one.last_seen,
                "group_id": one.group_id,
                "bodies": list(one.bodies),
            }
            for one in unknown
        ],
    }


def render_export(data: dict[str, list[dict[str, Any]]]) -> str:
    """Turn the collected rows into the markdown the export writes.

    Deliberate notes keep their text; unrecognised words are reduced to a
    count. That asymmetry is the whole point of the file. A body typed into
    a group chat is free text, and this repository is public and has leaked
    real names once already -- the personal-data gate learns the names it
    refuses from the live database, so it catches "anna" and cannot possibly
    catch "!annna". Counts cannot leak a name; sentences can.
    """
    lines = [HEADER, "## Asked for", ""]
    if not data["suggestions"]:
        lines.append("_Nothing yet._")
    lines.extend(
        f"- `{one['command']}` — {one['body'][:EXPORT_BODY_LIMIT]}" for one in data["suggestions"]
    )
    lines.extend(["", "## Mistyped, or tried and not there", ""])
    if not data["unknown"]:
        lines.append("_Nothing yet._")
    lines.extend(f"- `{one['word']}` x {one['hits']}" for one in data["unknown"])
    return "\n".join(lines) + "\n"


def router(deps: ManageDeps) -> APIRouter:
    """The suggestions endpoints.

    Args:
        deps: The running app's configuration and paths.

    Returns:
        A router to include in the management app.
    """
    api = APIRouter(prefix="/api")

    @api.get("/suggestions")
    async def suggestions() -> dict[str, list[dict[str, Any]]]:
        """Every filed idea and every unrecognised word, newest first."""
        return collect(deps.database_path)

    @api.post("/suggestions/export")
    async def export() -> dict[str, Any]:
        """Write the export and say where it went."""
        data = collect(deps.database_path)
        deps.export_path.parent.mkdir(parents=True, exist_ok=True)
        deps.export_path.write_text(render_export(data), encoding="utf-8")
        return {
            "path": str(deps.export_path),
            "suggestions": len(data["suggestions"]),
            "unknown": len(data["unknown"]),
        }

    return api