API overview

This bot showcases how to use most of the features in the library. Check the commands section to see the implementation of each command.

import logging
import os

from examples.commands import (
    AttachmentCommand,
    DeleteCommand,
    DeleteLocalAttachmentCommand,
    EditCommand,
    HelpCommand,
    PingCommand,
    ReceiveDeleteCommand,
    RegexTriggeredCommand,
    ReplyCommand,
    StylesCommand,
    TriggeredCommand,
    TypingCommand,
)
from signalbot import SignalBot, enable_console_logging


def main() -> None:
    enable_console_logging(logging.INFO)

    signal_service = os.environ["SIGNAL_SERVICE"]
    phone_number = os.environ["PHONE_NUMBER"]

    config = {
        "signal_service": signal_service,
        "phone_number": phone_number,
    }
    bot = SignalBot(config)

    bot.register(HelpCommand())

    # enable a chat command for all contacts and all groups
    bot.register(PingCommand())
    bot.register(ReplyCommand())

    # enable a chat command only for groups
    bot.register(AttachmentCommand(), contacts=False, groups=True)

    # enable a chat command for one specific group with the name "My Group"
    bot.register(TypingCommand(), groups=["My Group"])

    # chat command is enabled for all groups and one specific contact
    bot.register(TriggeredCommand(), contacts=["+490123456789"], groups=True)

    bot.register(RegexTriggeredCommand())

    bot.register(EditCommand())
    bot.register(DeleteCommand())
    bot.register(ReceiveDeleteCommand())
    bot.register(DeleteLocalAttachmentCommand())
    bot.register(StylesCommand())
    bot.start()


if __name__ == "__main__":
    main()

Commands

AttachmentCommand

import base64
from pathlib import Path

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, triggered


class AttachmentCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "friday: 🦀 Send and delete an image."

    @triggered("friday")
    async def handle(self, c: Context) -> None:
        with open(Path(__file__).parent / "image.jpeg", "rb") as f:  # noqa: ASYNC230, PTH123
            image = str(base64.b64encode(f.read()), encoding="utf-8")

        await c.send(
            "https://www.youtube.com/watch?v=pU2SdH1HBuk",
            base64_attachments=[image],
        )

DeleteCommand & DeleteLocalAttachmentCommand & ReceiveDeleteCommand

import asyncio
from datetime import datetime
from pathlib import Path

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, MessageType, triggered


class DeleteCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "delete: 🗑️ Delete a message."

    @triggered("delete")
    async def handle(self, c: Context) -> None:
        timestamp = await c.send("This message will be deleted in two seconds.")
        await asyncio.sleep(2)
        await c.remote_delete(timestamp=timestamp)


class DeleteLocalAttachmentCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "delete_attachment: 🗑️ Delete the local copy of an attachment."

    @triggered("delete_attachment")
    async def handle(self, c: Context) -> None:
        local_filenames = c.message.attachments_local_filenames
        if local_filenames is None or len(local_filenames) == 0:
            await c.send("Please send an attachment to delete.")

        for attachment_filename in local_filenames:
            attachment_path: Path = (
                Path.home()
                / ".local/share/signal-api/attachments"
                / attachment_filename
            )

            if attachment_path.exists():
                print(f"Received file {attachment_path}")  # noqa: T201

            await c.bot.delete_attachment(attachment_filename)

            if not attachment_path.exists():
                print(f"Deleted file {attachment_path}")  # noqa: T201


class ReceiveDeleteCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "N/A: 🗑️ Receive a message has been deleted notification."

    async def handle(self, c: Context) -> None:
        if c.message.type == MessageType.DELETE_MESSAGE:
            deleted_at = datetime.fromtimestamp(  # noqa: DTZ006
                c.message.remote_delete_timestamp / 1000
            )
            await c.send(f"You've deleted a message, which was sent at {deleted_at}.")

EditCommand

import asyncio

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, triggered


class EditCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "edit: ✏️ Edit a message."

    @triggered("edit")
    async def handle(self, c: Context) -> None:
        timestamp = await c.send("This message will be edited in two seconds.")
        await asyncio.sleep(2)
        await c.edit("This message has been edited.", timestamp)

HelpCommand

from abc import abstractmethod

from signalbot import Command, Context, triggered


class CommandWithHelpMessage(Command):
    @abstractmethod
    def help_message(self) -> str:
        pass


class HelpCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "help: 🆘 Shows information about available commands."

    @triggered("help")
    async def handle(self, c: Context) -> None:
        help_message = "Available commands:\n"
        command: CommandWithHelpMessage
        for command, _, _, _ in self.bot.commands:
            help_message += f"\t - {command.help_message()}\n"
        await c.send(help_message)

TriggeredCommand

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, triggered


class TriggeredCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "command_1, command_2 or command_3: 😤😤😤 Decorator example."

    # add case_sensitive=True for case sensitive triggers
    @triggered("command_1", "Command_2", "CoMmAnD_3")
    async def handle(self, c: Context) -> None:
        await c.send("I am triggered")

PingCommand

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, triggered


class PingCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "ping: 🏓 Listen for a ping and send a pong reply."

    @triggered("ping")
    async def handle(self, c: Context) -> None:
        await c.send("pong")

RegexTriggeredCommand

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, regex_triggered


class RegexTriggeredCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "^[\\w\\.-]+@gmail\\.com$: 😤 Regular expression decorator example."

    @regex_triggered(r"^[\w\.-]+@gmail\.com$")
    async def handle(self, c: Context) -> None:
        await c.send("Detected a Gmail address!")

ReplyCommand

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, triggered


class ReplyCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "reply: 💬 Reply to a message."

    @triggered("reply")
    async def handle(self, c: Context) -> None:
        await c.reply("This is a reply.")

StylesCommand

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, triggered


class StylesCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "styles: 🎨 Demonstrates different text styles."

    @triggered("styles")
    async def handle(self, c: Context) -> None:
        await c.send("**Bold style**", text_mode="styled")
        await c.send("*Italic style*", text_mode="styled")
        await c.send("~Strikethrough style~", text_mode="styled")
        await c.send("||Spoiler style||", text_mode="styled")
        await c.send("`Monospaced style`", text_mode="styled")

TypingCommand

import asyncio

from examples.commands.help import CommandWithHelpMessage
from signalbot import Context, triggered


class TypingCommand(CommandWithHelpMessage):
    def help_message(self) -> str:
        return "typing: ⌨️ Demonstrates typing indicator for a few seconds."

    @triggered("typing")
    async def handle(self, c: Context) -> None:
        await c.start_typing()
        seconds = 5
        await asyncio.sleep(seconds)
        await c.stop_typing()
        await c.send(f"Typed for {seconds}s")

The code shown here can be found the examples folder.