matrix-awarenessbot/bot.py

import asyncio
import pathlib
import yaml
from nio import (
    AsyncClient,
    MatrixRoom,
    RoomMessageText,
    RoomCreateResponse,
    RoomGetStateResponse,
)


class AwarenessBot:
    def __init__(self, config_path):
        # Load configuration from YAML file
        with open(config_path, "r") as config_file:
            self.config = yaml.safe_load(config_file)

        # Load dictionary from YAML file specified in the config
        with open(
            self.config.get(
                "dictionary_path",
                pathlib.Path(__file__).parent / "dictionary" / "default.yaml",
            ),
            "r",
        ) as dictionary_file:
            dictionary = yaml.safe_load(dictionary_file)
            self.non_inclusive_terms = dictionary["non_inclusive_terms"]
            self.unacceptable_terms = dictionary["unacceptable_terms"]

        # Initialize the client with server URL and credentials from config
        self.client = AsyncClient(self.config["server_url"], self.config["username"])

    def check_inclusive_language(self, message):
        suggestions = []
        for term, details in self.non_inclusive_terms.items():
            if term in message:
                suggestions.append(
                    {
                        "term": term,
                        "replacement": details["replacement"],
                        "reason": details["reason"],
                    }
                )
        return suggestions

    def check_unacceptable_language(self, message):
        for entry in self.unacceptable_terms:
            term = entry["term"]
            if term in message:
                return entry
        return None

    async def get_dm_room(self, user_id):
        # Check if a DM room already exists
        for room_id, room in self.client.rooms.items():
            if user_id in room.users and len(room.users) == 2:
                return room_id

        # Create a new DM room with the user
        response = await self.client.room_create(invite=[user_id], is_direct=True)

        if isinstance(response, RoomCreateResponse):
            return response.room_id
        else:
            return None  # Failed to create or invite to a room

    async def send_private_message(self, user_id, message):
        room_id = await self.get_dm_room(user_id)
        if room_id:
            await self.client.room_send(
                room_id,
                message_type="m.room.message",
                content={"msgtype": "m.text", "body": message},
            )

    async def get_room_admins(self, room_id):
        response = await self.client.room_get_state(room_id)
        if isinstance(response, RoomGetStateResponse):
            admins = []
            for event in response.events:
                if event.type == "m.room.power_levels":
                    power_levels = event.content
                    for user, level in power_levels.get("users", {}).items():
                        if level >= 100:  # TODO: Make this configurable or something
                            admins.append(user)
            return admins
        return []

    async def message_callback(self, room: MatrixRoom, event: RoomMessageText):
        if event.sender == self.client.user:
            return  # Ignore messages sent by the bot

        # Check for unacceptable language
        unacceptable_entry = self.check_unacceptable_language(event.body)
        if unacceptable_entry:
            admin_response = f"Attention: The term '{unacceptable_entry['term']}' used by {event.sender} in {room.room_id} might be unacceptable ({unacceptable_entry['reason']})."

            # Get room admins and send them a message
            admins = await self.get_room_admins(room.room_id)
            for admin in admins:
                await self.send_private_message(admin, admin_response)

            response = f"Your message in {room.room_id} contains the term '{unacceptable_entry['term']}' which might be unacceptable ({unacceptable_entry['reason']}). Please consider revising it."

            # Send private message to the user
            await self.send_private_message(event.sender, response)

        # Check for non-inclusive language
        suggestions = self.check_inclusive_language(event.body)
        if suggestions:
            response = f"I've noticed some potentially non-inclusive language in your message in {room.room_id}:\n"
            for suggestion in suggestions:
                response += f"- '{suggestion['term']}' could potentially be replaced with '{suggestion['replacement']}' ({suggestion['reason']}).\n"

            response += "Thank you for using inclusive language and making our community a better place! 🌈🦄"

            # Send private message to the user
            await self.send_private_message(event.sender, response)

    async def run(self):
        await self.client.login(self.config["password"])
        self.client.add_event_callback(self.message_callback, RoomMessageText)
        await self.client.sync_forever(timeout=30000)


def main():
    bot = AwarenessBot("config.yaml")
    asyncio.get_event_loop().run_until_complete(bot.run())


if __name__ == "__main__":
    main()