simplex-chat/packages/simplex-chat-python/src/simplex_chat/bot.py

"""`Bot` — Client extended with server-side features (address, auto-accept, commands)."""

from __future__ import annotations

from dataclasses import dataclass

from . import util
from .api import Db
from .client import (
    BotProfile,
    ChatMessage,
    Client,
    CommandHandler,
    EventHandler,
    FileMessage,
    ImageMessage,
    LinkMessage,
    Message,
    MessageHandler,
    Middleware,
    ParsedCommand,
    Profile,
    ReportMessage,
    TextMessage,
    UnknownMessage,
    VideoMessage,
    VoiceMessage,
    log,
)
from .core import MigrationConfirmation
from .types import T


@dataclass(slots=True)
class BotCommand:
    """One entry in the bot's advertised slash-command list (wire-side
    `groupPreferences.commands` or profile `preferences.commands`).

    `keyword` and `label` are required: `keyword` is what the user
    types after `/`; `label` is the human-readable description shown
    next to the keyword in the SimpleX client's commands menu.

    `params` is an optional placeholder string that controls how the
    client behaves when the user taps the command in the menu:

      * `params=None` (default) — the client SENDS `/<keyword>`
        immediately on tap; no input-box detour. Use this for
        zero-argument commands (`/help`, `/ping`) where the action is
        unambiguous.

      * `params="<value>"` — the client PASTES `/<keyword> <params>`
        into the input box and positions the cursor at the end. The
        user edits the placeholder and sends. Use this for commands
        that take a required argument (`/review <pr-url>`,
        `/order <number>`) so the user sees the expected shape
        without having to remember it.

    Mirrors `CBCCommand` in the Haskell core
    (`Simplex.Chat.Types.Preferences`) and the wire TypedDict
    `ChatBotCommand_command`. Both SimpleX clients (Android/Kotlin
    and iOS/Swift) implement the paste-vs-send branch on the
    `params` field; see `CommandsMenuView.{kt,swift}` for the
    reference UI behaviour.
    """
    keyword: str
    label: str
    params: str | None = None


class Bot(Client):
    """SimpleX bot — Client extended with server-side features.

    On top of `Client` (identity + messaging + connect_to/send_and_wait/events),
    a Bot:
      - creates and announces its own contact address
      - auto-accepts incoming contact requests (configurable)
      - advertises a list of slash-commands in its profile preferences
      - sets `peerType=bot` and disables calls/voice in profile prefs
      - sends a `welcome` message to new contacts via the auto-reply address setting

    If you want just identity + messaging without any of that, use `Client`
    directly.
    """

    def __init__(
        self,
        *,
        profile: Profile,
        db: Db,
        welcome: str | T.MsgContent | None = None,
        commands: list[BotCommand] | None = None,
        confirm_migrations: MigrationConfirmation = MigrationConfirmation.YES_UP,
        create_address: bool = True,
        update_address: bool = True,
        update_profile: bool = True,
        auto_accept: bool = True,
        business_address: bool = False,
        allow_files: bool = False,
        log_contacts: bool = True,
        log_network: bool = False,
    ) -> None:
        super().__init__(
            profile=profile,
            db=db,
            confirm_migrations=confirm_migrations,
            update_profile=update_profile,
            log_contacts=log_contacts,
            log_network=log_network,
        )
        self._welcome = welcome
        self._commands = commands or []
        self._create_address = create_address
        self._update_address = update_address
        self._auto_accept = auto_accept
        self._business_address = business_address
        self._allow_files = allow_files

    # ------------------------------------------------------------------ #
    # Profile + address sync (overrides hooks in Client)
    # ------------------------------------------------------------------ #

    async def _post_start(self, user: T.User) -> None:
        """Bots sync address first, then embed the link in the profile."""
        link = await self._sync_address(user)
        await self._maybe_sync_profile(user, contact_link=link)

    async def _sync_address(self, user: T.User) -> str | None:
        """Address sync. Returns the public link if any, for embedding in the profile."""
        api = self.api
        user_id = user["userId"]

        address = await api.api_get_user_address(user_id)
        if address is None:
            if self._create_address:
                log.info("Bot has no address, creating...")
                await api.api_create_user_address(user_id)
                address = await api.api_get_user_address(user_id)
                if address is None:
                    raise RuntimeError("Failed reading newly created user address")
            else:
                log.warning("Bot has no address")

        link: str | None = None
        if address is not None:
            link = util.contact_address_str(address["connLinkContact"])
            log.info("Bot address: %s", link)

        # Address settings (auto-accept + welcome message). Mirrors bot.ts:185-194.
        # autoAccept present → accept; absent → no auto-accept (mirrors Node bot.ts).
        if address is not None and self._update_address:
            desired: T.AddressSettings = {"businessAddress": self._business_address}
            if self._auto_accept:
                desired["autoAccept"] = {"acceptIncognito": False}
            if self._welcome is not None:
                desired["autoReply"] = (
                    {"type": "text", "text": self._welcome}
                    if isinstance(self._welcome, str)
                    else self._welcome
                )
            if address["addressSettings"] != desired:
                log.info("Bot address settings changed, updating...")
                await api.api_set_address_settings(user_id, desired)

        return link

    def _profile_to_wire(self) -> T.Profile:
        """Bot profile: base profile + peerType=bot, command list, calls/voice prefs disabled.

        Mirrors Node `mkBotProfile` (bot.ts:88-102).
        """
        p = super()._profile_to_wire()
        prefs: T.Preferences = {
            "calls": {"allow": "no"},
            "voice": {"allow": "no"},
            "files": {"allow": "yes" if self._allow_files else "no"},
        }
        if self._commands:
            cmds: list[T.ChatBotCommand] = []
            for c in self._commands:
                entry: T.ChatBotCommand_command = {
                    "type": "command",
                    "keyword": c.keyword,
                    "label": c.label,
                }
                # `params` is `NotRequired[str]` on the wire; omit the
                # key entirely when None so the Haskell parser sees
                # `Nothing` rather than `Just ""`. The two have
                # different client semantics: `Nothing` (`params=None`)
                # triggers an immediate send on tap; `Just ""` would
                # paste `/<keyword> ` (with a trailing space) into the
                # input box, which is rarely what the operator wants.
                if c.params is not None:
                    entry["params"] = c.params
                cmds.append(entry)
            prefs["commands"] = cmds
        p["preferences"] = prefs
        p["peerType"] = "bot"
        return p


__all__ = [
    "Bot",
    "BotCommand",
    "BotProfile",
    "ChatMessage",
    "Client",
    "CommandHandler",
    "EventHandler",
    "FileMessage",
    "ImageMessage",
    "LinkMessage",
    "Message",
    "MessageHandler",
    "Middleware",
    "ParsedCommand",
    "Profile",
    "ReportMessage",
    "TextMessage",
    "UnknownMessage",
    "VideoMessage",
    "VoiceMessage",
]