its-matrix-bot/its_matrix_bot/__init__.py

125 lines
4 KiB
Python
Raw Normal View History

import asyncio
2022-03-27 19:17:21 +02:00
import datetime
2022-03-29 19:28:20 +02:00
import logging
2022-03-27 19:17:21 +02:00
from nio.rooms import MatrixRoom
from nio.events.room_events import RoomMessage
import simplematrixbotlib as botlib
from .its_api import ItSyndikatApi
from .config import Config
2022-03-27 19:17:21 +02:00
class ItSyndikatBot:
bot: botlib.Bot
its_api: ItSyndikatApi
config: Config
def __init__(self, config: Config):
self.config = config
2022-03-29 19:29:05 +02:00
self.its_api = ItSyndikatApi(config)
2022-03-27 19:17:21 +02:00
self.current_open_state = None
2022-03-27 19:17:21 +02:00
creds = botlib.Creds(
config.matrix_homeserver,
config.matrix_username,
access_token=config.matrix_access_token,
2022-03-27 19:35:46 +02:00
session_stored_file="",
2022-03-27 19:17:21 +02:00
)
self.bot = botlib.Bot(creds)
self.bot.listener.on_message_event(self.on_message)
async def run(self):
async def poll_for_changes():
while True:
2022-03-29 19:28:20 +02:00
logging.debug("Polling open state")
try:
status = await self.its_api.status()
new_state = status["state"]["open"]
if (
self.current_open_state is not None
and new_state != self.current_open_state
):
await self.announce_open_change(new_state)
self.current_open_state = new_state
except Exception as e:
2022-03-29 19:28:20 +02:00
logging.error(f"Polling for open state failed: {e}")
await asyncio.sleep(60)
asyncio.create_task(poll_for_changes())
await self.bot.main()
2022-03-27 19:17:21 +02:00
async def on_message(self, room, message):
m = botlib.MessageMatch(room, message, self.bot, self.config.command_prefix)
if m.is_not_from_this_bot() and m.prefix():
if m.command("echo"):
await self.echo(room, message, m.args())
elif m.command("isitopen"):
await self.isitopen(room, message)
elif m.command("spaceping"):
await self.spaceping(room, message)
else:
await self.bot.api.send_text_message(
room.room_id, f"Unknown command: {m.command()}"
)
async def announce_open_change(self, now_open: bool):
2022-03-29 19:28:20 +02:00
logging.info("Open state changed: now " + ("open" if now_open else "closed"))
room_ids = self.config.isitopen_announce_rooms
if now_open:
message = "opening IT-Syndikat - Ohai!"
else:
message = "closing IT-Syndikat - nap time!"
for room_id in room_ids:
await self.bot.api.async_client.room_send(
room_id=room_id,
message_type="m.room.message",
content={
"msgtype": "m.notice",
"body": message,
},
)
2022-03-27 19:17:21 +02:00
async def reply(self, room, message, reply):
await self.bot.api.async_client.room_send(
room_id=room.room_id,
message_type="m.room.message",
content={
"msgtype": "m.text",
"body": reply,
"m.relates_to": {"m.in_reply_to": {"event_id": message.event_id}},
},
)
async def echo(self, room, message, args):
await self.bot.api.send_text_message(
room.room_id, " ".join(arg for arg in args)
)
async def isitopen(self, room, message):
try:
status = await self.its_api.status()
is_open = status["state"]["open"]
self.current_open_state = is_open
if is_open:
2022-03-27 19:17:21 +02:00
date = datetime.datetime.fromtimestamp(status["state"]["lastchange"])
text = f"positive! space has been open since {date}"
else:
text = "negative!"
except Exception as e:
text = f"error checking space status: {e}"
await self.reply(room, message, text)
async def spaceping(self, room: MatrixRoom, message: RoomMessage):
await self.its_api.ping()
await self.reply(room, message, "Hello Space!")