matrix-awarenessbot/bot.py
Kumi 0f58772c26
refactor: rename InclusiveLanguageBot to AwarenessBot
Renamed the InclusiveLanguageBot class and its instances to AwarenessBot
for better alignment with the bot’s broader focus on awareness beyond
just language inclusivity.
2024-08-25 13:53:41 +02:00

133 lines
5.1 KiB
Python

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()