razem-mattermost-bot 1.0.1

Helpers for writing your organization Mattermost bots with mattermostdriver.

Razem Mattermost Bot

razem-mattermost-bot to mała biblioteka do szybkiego pisania botów Mattermost dla Partii Razem. Opiera się na mattermostdriver, ale usuwa powtarzalny boilerplate z naszych botów: konfigurację hosta i tokenu, patch SSL dla websocketów, logowanie, filtrowanie eventów, odpowiedzi w wątkach, reakcje, edycje postów, DM-y, wyszukiwanie użytkowników, paginację API, stan JSON i podstawowe operacje na członkostwie w kanałach.

Domyślny host to https://mattermost.example.org. Token nigdy nie jest zaszyty w kodzie: można przekazać go jawnie albo ustawić MATTERMOST_TOKEN / MATTERMOST_ACCESS_TOKEN.

Instalacja

Lokalnie, podczas pracy nad pakietem:

uv add ../razem-mattermost-bot

Z repozytorium Forgejo:

uv add git+https://github.com/example/razem-mattermost-bot.git

Konfiguracja

Najprostszy .env:

MATTERMOST_TOKEN=tu_wklej_token_bota

Opcjonalnie:

MATTERMOST_HOST=https://mattermost.example.org

W kodzie można też przekazać token bezpośrednio:

from razem_mattermost_bot import RazemMattermostBot

bot = RazemMattermostBot(token="token-z-bezpiecznego-sekretu")

Bardzo prosty bot

Bot odpowiada w wątku na !ping we wszystkich kanałach, do których ma dostęp.

from razem_mattermost_bot import Post, RazemMattermostBot

bot = RazemMattermostBot(name="ping-bot")


@bot.on_trigger("!ping")
def ping(post: Post) -> None:
    bot.reply_in_thread(post, "pong")


bot.start()

Słuchanie tylko jednego kanału

Jeśli bot ma reagować tylko na konkretne kanały, podaj ich ID:

from razem_mattermost_bot import Post, RazemMattermostBot

DYZURY_CHANNEL_ID = "abc123kanal"

bot = RazemMattermostBot(name="dyzury-bot")


@bot.on_hashtag("#dyzur", channel_ids={DYZURY_CHANNEL_ID})
async def handle_shift(post: Post) -> None:
    bot.reply_in_thread(post, "Zapisane. Dzięki!")


bot.start()

Brak channel_ids oznacza: reaguj na wszystkie kanały, w których Mattermost wyśle event do bota.

Warunek logiczny zamiast komendy

Możesz zarejestrować dowolny warunek:

from razem_mattermost_bot import Post, RazemMattermostBot

bot = RazemMattermostBot(name="uwaga-bot")


def asks_for_help(post: Post) -> bool:
    text = post.message.lower()
    return "kto pomoże" in text or "potrzebuję pomocy" in text


@bot.on_message(condition=asks_for_help, include_threads=False)
def help_reply(post: Post) -> None:
    bot.reply_in_channel(post.channel_id, "Widzę prośbę o pomoc. Oznaczam dyżur.")


bot.start()

Powitanie nowych osób

user_added pozwala śledzić nowe osoby w kanałach i wysyłać im DM-y albo odpowiedź na kanale.

from razem_mattermost_bot import MattermostEvent, RazemMattermostBot

START_CHANNEL_ID = "kanal-startowy"

bot = RazemMattermostBot(name="welcome-bot")


@bot.on_user_added(channel_ids={START_CHANNEL_ID})
def welcome(event: MattermostEvent) -> None:
    if event.user_id:
        bot.send_dm(
            event.user_id,
            "Cześć! Tu kilka linków na start: ...",
        )


bot.start()

Dodawanie i usuwanie osób z kanału

Użytkownika można znaleźć po ID, mailu albo nazwie:

from razem_mattermost_bot import Post, RazemMattermostBot

bot = RazemMattermostBot(name="channel-admin-bot")


@bot.on_trigger("!dodaj")
def add_person(post: Post) -> None:
    parts = post.message.split(maxsplit=2)
    if len(parts) < 3:
        bot.reply_in_thread(post, "Użycie: `!dodaj @nick kanal_id`")
        return

    user = bot.find_user(parts[1])
    channel_id = parts[2]
    bot.add_user_to_channel(channel_id, user["id"])
    bot.reply_in_thread(post, f"Dodano {parts[1]} do kanału.")


bot.start()

Analogicznie:

bot.remove_user_from_channel(channel_id, user["id"])

Coś bardziej kompletnego: prosty bot zapisów

Ten przykład reaguje na hashtag, zapisuje chętnych przez odpowiedź w wątku i potwierdza DM-em.

from razem_mattermost_bot import Post, RazemMattermostBot

bot = RazemMattermostBot(name="zapisy-bot")
events: dict[str, set[str]] = {}


@bot.on_hashtag("#zapisy", include_threads=False)
def create_signup(post: Post) -> None:
    events[post.id] = set()
    bot.reply_in_thread(
        post,
        "Utworzono listę zapisów. Odpowiedz w tym wątku, żeby się zapisać.",
    )


@bot.on_message()
def signup_reply(post: Post) -> None:
    if not post.root_id or post.root_id not in events:
        return
    events[post.root_id].add(post.user_id)
    bot.reply_in_thread(post, "Zapisane.")
    bot.send_dm(post.user_id, "Potwierdzam zapis.")


bot.start()

Reakcje: tablica zapisów emoji

Wzór z emojibot: jedna mapa emoji -> kanał, handler na reakcje i opcjonalny skan istniejących reakcji.

from razem_mattermost_bot import Reaction, RazemMattermostBot

BOARD_POST_ID = "post_z_tablica"
EMOJI_TO_CHANNEL = {
    "art": "kanal-grafiki",
    "newspaper": "kanal-mediow",
}

bot = RazemMattermostBot(name="emoji-board")


@bot.on_reaction_added(*EMOJI_TO_CHANNEL.keys(), post_ids={BOARD_POST_ID})
def join_from_reaction(reaction: Reaction) -> None:
    channel_id = EMOJI_TO_CHANNEL[reaction.emoji_name]
    bot.add_user_to_channel(channel_id, reaction.user_id)


bot.start()

Jeśli bot był wyłączony, można doskanować reakcje na starcie po bot.connect():

bot.connect()
for reaction in bot.get_reactions(BOARD_POST_ID):
    emoji = reaction["emoji_name"]
    if emoji in EMOJI_TO_CHANNEL:
        bot.add_user_to_channel(EMOJI_TO_CHANNEL[emoji], reaction["user_id"])
bot.start()

Edycje postów i mirroring

Wzór z notifier / notifier-sm: bot kopiuje post do innego kanału, zapisuje powiązanie w props, a po edycji oryginału aktualizuje kopię.

from razem_mattermost_bot import Post, RazemMattermostBot

SOURCE = "kanal-zrodlowy"
TARGET = "kanal-docelowy"

bot = RazemMattermostBot(name="mirror-bot")


def format_copy(post: dict, author: dict) -> str:
    username = author.get("username", post.get("user_id", "unknown"))
    return f"**Zlecenie**: {post['message'].strip()}\n**Osoba**: {username}"


@bot.on_hashtag("#zlecenie", channel_ids={SOURCE}, include_threads=False)
def mirror(post: Post) -> None:
    bot.mirror_post(post, target_channel_id=TARGET, message_builder=format_copy)


@bot.on_post_edited(channel_ids={SOURCE}, include_threads=False)
def update_copy(post: Post) -> None:
    bot.update_mirror_for_post(post, target_channel_id=TARGET, message_builder=format_copy)


bot.start()

Pobieranie postów i członków kanałów

Paginacja, filtrowanie root-postów, lista członków.

from razem_mattermost_bot import RazemMattermostBot

bot = RazemMattermostBot()
bot.connect()

posts = bot.get_root_posts_for_channel("kanal", since_ms=1760000000000)
member_ids = bot.get_channel_member_user_ids("kanal")
users = bot.get_users_by_ids(member_ids)

get_root_posts_for_channel() odrzuca odpowiedzi w wątkach, usunięte posty i wiadomości systemowe, a potem sortuje od najstarszych.

Stan JSON

Wzór z bingo, emoji-reminder, onboarding-bot, czescbot, wulgarbot: atomowy zapis state.json bez kopiowania helperów.

from razem_mattermost_bot import JsonStateStore

store = JsonStateStore("state.json", default={"seen_user_ids": []})
state = store.load()
state["seen_user_ids"].append("user_id")
store.save(state)

Prosty łańcuch rozmowy

Do botów, które prowadzą użytkownika przez kilka kroków:

from razem_mattermost_bot import ConversationStore, Post, RazemMattermostBot

bot = RazemMattermostBot()
conversations = ConversationStore()


@bot.on_trigger("!start")
def start(post: Post) -> None:
    conversations.start(post.user_id, "color")
    bot.reply_in_thread(post, "Podaj kolor.")


@bot.on_message()
def continue_flow(post: Post) -> None:
    state = conversations.get(post.user_id)
    if not state:
        return
    if state.step == "color":
        conversations.advance(post.user_id, "number", color=post.message)
        bot.reply_in_thread(post, "Podaj liczbę.")
    elif state.step == "number":
        color = state.data["color"]
        conversations.cancel(post.user_id)
        bot.reply_in_thread(post, f"Gotowe: {color}, {post.message}")


bot.start()

Systemd

Pakiet nie uruchamia sudo i nie instaluje usługi samodzielnie, ale potrafi wyrenderować unit systemd dla wdrożenia przez operatora:

from razem_mattermost_bot import SystemdService

service = SystemdService(
    name="zapisy-bot",
    working_directory="/opt/zapisy-bot",
    module="zapisy_bot.main",
    env_file="/etc/zapisy-bot.env",
    user="mattermost-bot",
)

print(service.render())

ExecStart używa uv run python -m ..., więc wdrożenie może korzystać z lockfile projektu bota.

Logowanie

Każdy RazemMattermostBot konfiguruje spójny format:

[2026-05-21 12:00:00,000] INFO zapisy-bot: event=mattermost_login_ok username='zapisy' user_id='...'

Do własnych zdarzeń:

bot.events.event("signup_created", channel_id=post.channel_id, post_id=post.id)

Dokumentacja API

Docstringi są po angielsku i w stylu zgodnym z mkdocstrings. Dokumentację można wygenerować lokalnie:

uv run --extra docs mkdocs serve

Albo zbudować statycznie:

uv run --extra docs mkdocs build

Pliki konfiguracyjne dokumentacji są w mkdocs.yml i docs/.

Testy

uv run --extra test pytest -q