cqrenet/synapse-module-auto-invite
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

First working code

Browse Source
This commit is contained in:
Tomas Kracmar
2024-04-25 11:46:00 +02:00
parent 8c4775e882
commit 801e6b527e
1 changed files with 26 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
module_auto_invite.py
View File

@@ -1,55 +1,51 @@
import logging
from synapse.module_api import ModuleApi
from synapse.module_api import ModuleApi, errors
logger = logging.getLogger(__name__)
class AutoInviteBotModule:
def __init__(self, config, api: ModuleApi):
self.api = api
self.bot_user_id = config.get("bot_user_id", "@bot.id:blackmesaresearch.net")
logger.info("Module initialized with bot_user_id: %s", self.bot_user_id)
# Register the new event callback only
# Fetch the bot user ID from the config file
self.auto_invite_user_id = config.get("auto_invite_user_id")
if not self.auto_invite_user_id:
raise ValueError("auto_invite_user_id must be configured")
self.api.register_third_party_rules_callbacks(
on_new_event=self.on_new_event
)
@staticmethod
def parse_config(config):
# Validate that the auto_invite_user_id is in the config
if "auto_invite_user_id" not in config:
logger.error("auto_invite_user_id not provided in config")
raise Exception("auto_invite_user_id must be provided in config")
return config
async def on_new_event(self, event, state_events):
# Respond only to membership changes, not all messages
if event.type == "m.room.member":
room_id = event.room_id
target_user_id = event.state_key
membership = event.content.get("membership")
if event['type'] == "m.room.member" and event['content'].get('membership') == "invite":
room_id = event['room_id']
inviter = event['sender']
invitee = event['state_key']
logger.debug("Membership event in room %s for user %s: %s", room_id, target_user_id, membership)
# Fetch the current state to check the bot's membership status
current_state = await self.api.get_room_state_by_type(room_id, "m.room.member", self.bot_user_id)
bot_membership = current_state.get("content", {}).get("membership", "leave")
# Log current membership state
logger.debug("Bot's current membership status in %s: %s", room_id, bot_membership)
# Invite bot if it is not already a member
if bot_membership != "join":
logger.info("Bot is not a member of the room. Sending invite...")
# Check if the bot is already in the room
try:
state = await self.api.get_room_state(room_id)
bot_membership = state.get(("m.room.member", self.auto_invite_user_id), {}).get("content", {}).get("membership")
if bot_membership != "join":
await self.api.update_room_membership(
sender=event.sender,
target=self.bot_user_id,
sender=inviter,
target=self.auto_invite_user_id,
room_id=room_id,
membership="invite"
new_membership="invite"
)
logger.info("Bot invited to room %s upon new event.", room_id)
except Exception as e:
logger.error("Failed to invite bot to room %s: %s", room_id, e)
else:
logger.info("Bot is already a member of the room. No action needed.")
logger.info(f"Invited bot to room {room_id} as {invitee} was invited by {inviter}")
except errors.SynapseError as e:
logger.error(f"Failed to invite bot to room {room_id}: {str(e)}")
def setup(config, api):
# Parse and validate config when setting up the module
module_config = AutoInviteBotModule.parse_config(config)
return AutoInviteBotModule(module_config, api)