123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
"""Manage sending matrix messages to matrix rooms."""
|
|
|
|
import datetime
|
|
import logging
|
|
from typing import List
|
|
|
|
import nio
|
|
|
|
from .config import Config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MatrixBot:
|
|
"""A class for managing the matrix bot that handles messaging users."""
|
|
|
|
def __init__(self, config: Config):
|
|
"""Initialize some variables for the bot."""
|
|
# Basic setup
|
|
self.client = None
|
|
self.initial_sync_done = None
|
|
|
|
# Matrix configuration
|
|
self.homeserver = config.matrix_homeserver
|
|
self.bot_id = config.matrix_user
|
|
self.bot_password_file = config.credentials_dir / "bot-password"
|
|
|
|
async def _on_error(self, response):
|
|
if self.client:
|
|
await self.client.close()
|
|
raise Exception(response)
|
|
|
|
async def _on_sync(self, response):
|
|
if not self.initial_sync_done:
|
|
self.initial_sync_done = True
|
|
for room_id in self.client.rooms:
|
|
logger.info(f"joined room {room_id}")
|
|
|
|
if self.client.rooms[room_id].member_count < 2:
|
|
await self.client.room_leave(room_id)
|
|
logger.debug(f"left room {room_id} since all other users left")
|
|
|
|
logger.info("initial sync done, ready for work")
|
|
await self._send_greeting()
|
|
|
|
async def _on_invite(self, room, event):
|
|
logger.info(f"invited by {event.sender} to {room.room_id}")
|
|
|
|
if event.sender in ["@tlater:matrix.tlater.net", "@yuanyuan:matrix.tlater.net"]:
|
|
await self.client.join(room.room_id)
|
|
await self.client.room_send(
|
|
room_id=room.room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": "m.text",
|
|
"body": "Hi! I'll inform you if new slots become available.",
|
|
},
|
|
)
|
|
else:
|
|
logger.info(f"rejected invite by {event.sender}")
|
|
await self.client.room_leave(room.room_id)
|
|
|
|
async def send_appointments(self, slots: List[datetime.datetime]):
|
|
"""Send a message with spotted appointments to all joined rooms."""
|
|
assert self.client
|
|
|
|
for room_id in self.client.rooms:
|
|
logger.info(f"notifying room {room_id} about new slots")
|
|
|
|
message = "Appointment time slots spotted:\n\n" + "\n".join(
|
|
slot.strftime("%Y-%m-%d %H:%M") for slot in slots
|
|
)
|
|
|
|
await self.client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": "m.text",
|
|
"body": message,
|
|
},
|
|
)
|
|
|
|
async def send_warning(self):
|
|
"""Send a message telling everyone that we're no longer running."""
|
|
assert self.client
|
|
|
|
for room_id in self.client.rooms:
|
|
await self.client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": "m.text",
|
|
"body": "Something went wrong, tell Tristan, I might be broken!",
|
|
},
|
|
)
|
|
|
|
async def _send_greeting(self):
|
|
"""Send a greeting to make sure the bot is working."""
|
|
assert self.client
|
|
|
|
for room_id in self.client.rooms:
|
|
await self.client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": "m.text",
|
|
"body": "Whaaaaaaa, good monning. I'll check slots for you.",
|
|
},
|
|
)
|
|
|
|
async def run(self):
|
|
"""Start the bot."""
|
|
logger.info(f"Connecting to {self.homeserver}")
|
|
self.client = nio.AsyncClient(self.homeserver, self.bot_id)
|
|
self.client.device_id = "TLSContactAppointmentBot"
|
|
self.client.add_response_callback(self._on_error, nio.SyncError)
|
|
self.client.add_response_callback(self._on_sync, nio.SyncResponse)
|
|
self.client.add_event_callback(self._on_invite, nio.InviteMemberEvent)
|
|
|
|
logger.info(await self.client.login(self.bot_password_file.read_text()))
|
|
await self.client.sync_forever(timeout=30000, loop_sleep_time=200)
|
|
await self.client.close()
|