Уведомления от RouterOS в XMPP

[karton]

Хотел найти какую-то альтернативу уведомлениям в Telegram и пришла идея слать различные уведомления в XMPP. Сразу скажу что я не особо программист и почти всё что касалось кода было сделано руками ChatGPT. Но в любом случае проект как мне кажется очень интересный и может кому-то пригодится, поэтому решил описать.

1. Поднять XMPP-сервер
Теоретически можно найти какой-то публичный и просто использовать аккаунт с него, но я хотел снизить количество точек отказа и использовал self-hosted XMPP-сервер. Как самый простой могу посоветовать Snikket (https://snikket.org/service/quickstart/). Там всё спокойно поднимается через docker-compose.
Далее в нём регаем аккаунт себе и боту, можно хоть для кого создать эти аккаунты.

2. Делаем простой web-api для бота
Тут я и обратился к ChatGPT который написал мне простой WEB-API.
web_api.py

Код: Выделить всё

from fastapi import FastAPI
import subprocess
import uvicorn
import os

SECRET = os.environ['SECRET']

app = FastAPI()

@app.get("/")
async def index():
    return {"status": "ok"}


@app.get("/send")
async def send(secret: str, text: str):
    if secret == SECRET:
        subprocess.run(
            ["python", "sender.py", text],
            check=True
        )
        return {
            "status": "ok",
            "text": text
        }
    else:
        return {
            "status": "error"
        }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

sender.py

Код: Выделить всё

import asyncio
import logging
import sys
import os
from slixmpp import ClientXMPP

JID = os.environ["LOGIN"]
PASSWORD = os.environ["PASSWORD"]
DESTINATION = os.environ["DESTINATION"]
class Bot(ClientXMPP):

    def __init__(self, jid, password, text):
        super().__init__(jid, password)

        self.text = text

        self.add_event_handler("session_start", self.session_start)

    async def session_start(self, event):
        self.send_presence()
        await self.get_roster()

        self.send_message(
            mto=DESTINATION,
            mbody=self.text,
            mtype="chat"
        )

        # Подождать немного, чтобы сообщение успело уйти
        await asyncio.sleep(1)

        self.disconnect()


async def main():

    text = sys.argv[1]

    xmpp = Bot(JID, PASSWORD, text)

    await xmpp.connect()

    # Ждём пока disconnect() завершит соединение
    await xmpp.disconnected


if __name__ == "__main__":

    logging.basicConfig(
        level=logging.INFO,
        format="%(levelname)-8s %(message)s"
    )

    asyncio.run(main())

Там как видно нужно ещё переменные среды использовать (.env-файл), но там всё вроде понятно, если не понятно - спросите ChatGPT или тут :) , тоже самое если есть желание упаковать это в docker.

3. Привязываем RouterOS
Для отправки сообщения нужно сделать скрипт:

Код: Выделить всё

:global notifyText
:put $notifyText
/tool fetch http-method=get url="http://*IP*:*PORT*/send?secret=*SECRET*&text=$notifyText"

Также поставьте галочку Don`t Require Permission (без него у меня не работал fetch)

И допустим если хотим вызвать его из Netwatch то делаем там такое действие на Up/Down:

Код: Выделить всё

:global notifyText ("%F0%9F%9F%A2%20Host%20MikroTik-Srv-1%20is%20UP")
/system script run sender

там набор из спец символов отвечают на красивый зелёный кружок и пробелы, методом тыка и вопросов к ChatGPT можно разобраться)

4. Чем читать
Любым клиентом XMPP, я протестировал современные Gajim (для ПК) и Snikket (для телефона), везде всё нормально.




К одному боту можно спокойно привязать несколько микротиков, и уже в скрипте на микротике добавить какие-нибудь идентификаторы (так и сделано на скринах). P.S. Не уверен что этот API выдержит несколько сообщений в один момент и что-то может "проглотиться", так что если вам нужно больше надёжности - напишите сами или ещё чуть пообщайтесь с ChatGPT или др.

Пишу первый раз подобную статью) Да и прошу прощения за нейрослоп, но это вроде сообщество не совсем программистов, так что прошу отнестись с пониманием. Если есть какие-то вопросы - пишите.