Dome/Advanced-Community-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

better powerlevel handling, first pass

Browse Source
This commit is contained in:
William Kray
2025-04-04 10:24:43 -07:00
parent d894c36a4c
commit 6d0b178840
2 changed files with 426 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
community/bot.py
+425 -285
View File
@@ -66,6 +66,26 @@ class CommunityBot(Plugin):
self._redaction_tasks.cancel()
await super().stop()
async def user_permitted(self, user_id: UserID, min_level: int = 50) -> bool:
"""Check if a user has sufficient power level in the parent room.
Args:
user_id: The Matrix ID of the user to check
min_level: Minimum required power level (default 50 for moderator)
Returns:
bool: True if user has sufficient power level
"""
try:
power_levels = await self.client.get_state_event(
self.config["parent_room"], EventType.ROOM_POWER_LEVELS
)
user_level = power_levels.get_user_level(user_id)
return user_level >= min_level
except Exception as e:
self.log.error(f"Failed to check user power level: {e}")
return False
async def _redaction_loop(self) -> None:
while True:
try:
@@ -357,6 +377,82 @@ class CommunityBot(Plugin):
except Exception as e:
self.log.error(e)
@event.on(EventType.ROOM_POWER_LEVELS)
async def sync_power_levels(self, evt: StateEvent) -> None:
# Only care about changes in the parent room
if evt.room_id != self.config['parent_room']:
return
# Get the changed user and their new power level
try:
old_levels = evt.prev_content.get("users", {})
new_levels = evt.content.get("users", {})
# Find which user's power level changed
changed_users = {}
for user, new_level in new_levels.items():
if user not in old_levels or old_levels[user] != new_level:
changed_users[user] = new_level
if not changed_users:
return
# Get all rooms in the space
space_rooms = await self.client.get_joined_rooms()
success_rooms = []
failed_rooms = []
# Apply the same power level changes to each room
for room_id in space_rooms:
if room_id == self.config['parent_room']:
continue
try:
roomname = (await self.client.get_state_event(room_id, "m.room.name"))["name"]
except:
self.log.warning(f"Unable to get room name for {room_id}")
# Get current power levels
try:
# Get current power levels
current_pl = await self.client.get_state_event(room_id, EventType.ROOM_POWER_LEVELS)
# Update existing power levels object with new levels
users = current_pl.get("users", {})
for user, level in changed_users.items():
users[user] = level
current_pl["users"] = users
# Send updated power levels
try:
await self.client.send_state_event(room_id, EventType.ROOM_POWER_LEVELS, current_pl)
success_rooms.append(roomname or room_id)
except Exception as e:
self.log.error(f"Failed to send power levels to {roomname or room_id}: {e}")
failed_rooms.append(roomname or room_id)
time.sleep(self.config['sleep'])
except Exception as e:
self.log.warning(f"Failed to update power levels in {room_id}: {e}")
failed_rooms.append(room_id)
# Send notification if configured
if self.config["notification_room"]:
changes = ", ".join([f"{user} → {level}" for user, level in changed_users.items()])
notification = f"Power level changes ({changes}) propagated from parent room:<br>"
notification += f"Succeeded in: <code>{', '.join(success_rooms)}</code><br>"
if failed_rooms:
notification += f"Failed in: <code>{', '.join(failed_rooms)}</code>"
await self.client.send_notice(
self.config["notification_room"],
html=notification
)
except Exception as e:
self.log.error(f"Error syncing power levels: {e}")
@event.on(InternalEventType.JOIN)
async def newjoin(self, evt:StateEvent) -> None:
@@ -476,137 +572,136 @@ class CommunityBot(Plugin):
@community.subcommand("sync", help="update the activity tracker with the current space members \
in case they are missing")
async def sync_space_members(self, evt: MessageEvent) -> None:
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
if not self.config["track_users"]:
await evt.respond("user tracking is disabled")
return
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
results = await self.do_sync()
added_str = "<br />".join(results['added'])
dropped_str = "<br />".join(results['dropped'])
await evt.respond(f"Added: {added_str}<br /><br />Dropped: {dropped_str}", allow_html=True)
# check config values for admins and moderators. if they have a lower PL in the parent room,
# attempt to update the parent room with their appropriate admin/mod status
# we can skip all of this logic if those config values are empty
# this logic helps migrate explicit configuration to the parent-room inheritance model
if not self.config["admins"] and not self.config["moderators"]:
self.log.info("no admins or moderators configured, skipping power level sync")
pass
else:
await evt.reply("lol you don't have permission to do that")
power_levels = await self.client.get_state_event(self.config["parent_room"], EventType.ROOM_POWER_LEVELS)
users = power_levels.get("users", {})
for user in self.config["admins"]:
if user not in users or users.get(user) < 100:
# update the users object in-place
users[user] = 100
for user in self.config["moderators"]:
if user not in users or users.get(user) < 50:
# update the users object in-place
users[user] = 50
try:
# update full powerlevels object with updated user object
power_levels["users"] = users
await self.client.send_state_event(self.config["parent_room"], EventType.ROOM_POWER_LEVELS, power_levels)
# if updating was successful, let's go ahead and clear out the values in the config
self.config["admins"] = []
self.config["moderators"] = []
# and save the config to the file
self.config.save()
self.log.debug("successfully migrated admin/mod config to parent room")
except Exception as e:
self.log.error(f"Failed to send power levels to {self.config["parent_room"]}: {e}")
await evt.respond(f"Failed to send power levels to {self.config["parent_room"]}: {e}")
if not self.config["track_users"]:
await evt.respond("user tracking is disabled")
return
results = await self.do_sync()
added_str = "<br />".join(results['added'])
dropped_str = "<br />".join(results['dropped'])
await evt.respond(f"Added: {added_str}<br /><br />Dropped: {dropped_str}", allow_html=True)
@community.subcommand("ignore", help="exclude a specific matrix ID from inactivity tracking")
@command.argument("mxid", "full matrix ID", required=True)
async def ignore_inactivity(self, evt: MessageEvent, mxid: UserID) -> None:
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
if not self.config["track_users"]:
await evt.reply("user tracking is disabled")
return
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
try:
Client.parse_user_id(mxid)
await self.database.execute("UPDATE user_events SET ignore_inactivity = 1 WHERE \
mxid = $1", mxid)
self.log.info(f"{mxid} set to ignore inactivity")
await evt.react("✅")
except Exception as e:
await evt.respond(f"{e}")
else:
await evt.reply("lol you don't have permission to set that")
if not self.config["track_users"]:
await evt.reply("user tracking is disabled")
return
try:
Client.parse_user_id(mxid)
await self.database.execute("UPDATE user_events SET ignore_inactivity = 1 WHERE \
mxid = $1", mxid)
self.log.info(f"{mxid} set to ignore inactivity")
await evt.react("✅")
except Exception as e:
await evt.respond(f"{e}")
@community.subcommand("unignore", help="re-enable activity tracking for a specific matrix ID")
@command.argument("mxid", "full matrix ID", required=True)
async def unignore_inactivity(self, evt: MessageEvent, mxid: UserID) -> None:
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
if not self.config["track_users"]:
await evt.reply("user tracking is disabled")
return
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
try:
Client.parse_user_id(mxid)
await self.database.execute("UPDATE user_events SET ignore_inactivity = 0 WHERE \
mxid = $1", mxid)
self.log.info(f"{mxid} set to track inactivity")
await evt.react("✅")
except Exception as e:
await evt.respond(f"{e}")
else:
await evt.reply("lol you don't have permission to set that")
if not self.config["track_users"]:
await evt.reply("user tracking is disabled")
return
try:
Client.parse_user_id(mxid)
await self.database.execute("UPDATE user_events SET ignore_inactivity = 0 WHERE \
mxid = $1", mxid)
self.log.info(f"{mxid} set to track inactivity")
await evt.react("✅")
except Exception as e:
await evt.respond(f"{e}")
@community.subcommand("report", help='generate a list of matrix IDs that have been inactive')
async def get_report(self, evt: MessageEvent) -> None:
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
if not self.config["track_users"]:
await evt.reply("user tracking is disabled")
return
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
sync_results = await self.do_sync()
report = await self.generate_report()
await evt.respond(f"<p><b>Users inactive for between {self.config['warn_threshold_days']} and \
{self.config['kick_threshold_days']} days:</b><br /> \
{'<br />'.join(report['warn_inactive'])} <br /></p>\
<p><b>Users inactive for at least {self.config['kick_threshold_days']} days:</b><br /> \
{'<br />'.join(report['kick_inactive'])} <br /></p> \
<p><b>Ignored users:</b><br /> \
{'<br />'.join(report['ignored'])}</p>", \
allow_html=True)
if not self.config["track_users"]:
await evt.reply("user tracking is disabled")
return
sync_results = await self.do_sync()
report = await self.generate_report()
await evt.respond(f"<p><b>Users inactive for between {self.config['warn_threshold_days']} and \
{self.config['kick_threshold_days']} days:</b><br /> \
{'<br />'.join(report['warn_inactive'])} <br /></p>\
<p><b>Users inactive for at least {self.config['kick_threshold_days']} days:</b><br /> \
{'<br />'.join(report['kick_inactive'])} <br /></p> \
<p><b>Ignored users:</b><br /> \
{'<br />'.join(report['ignored'])}</p>", \
allow_html=True)
@community.subcommand("purge", help='kick users for excessive inactivity')
async def kick_users(self, evt: MessageEvent) -> None:
await evt.mark_read()
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
msg = await evt.respond("starting the purge...")
report = await self.generate_report()
purgeable = report['kick_inactive']
roomlist = await self.get_space_roomlist()
# don't forget to kick from the space itself
roomlist.append(self.config["parent_room"])
purge_list = {}
error_list = {}
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
for user in purgeable:
purge_list[user] = []
for room in roomlist:
try:
roomname = None
roomnamestate = await self.client.get_state_event(room, 'm.room.name')
roomname = roomnamestate['name']
await self.client.get_state_event(room, EventType.ROOM_MEMBER, user)
await self.client.kick_user(room, user, reason='inactivity')
if roomname:
purge_list[user].append(roomname)
else:
purge_list[user].append(room)
time.sleep('sleep')
except MNotFound:
pass
except Exception as e:
self.log.warning(e)
error_list[user] = []
error_list[user].append(roomname or room)
results = "the following users were purged:<p><code>{purge_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(purge_list=purge_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
# sync our database after we've made changes to room memberships
await self.do_sync()
else:
await evt.reply("lol you don't have permission to do that")
@community.subcommand("kick", help='kick a specific user from the community and all rooms')
@command.argument("mxid", "full matrix ID", required=True)
async def kick_user(self, evt: MessageEvent, mxid: UserID) -> None:
await evt.mark_read()
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
user = mxid
msg = await evt.respond("starting the purge...")
roomlist = await self.get_space_roomlist()
# don't forget to kick from the space itself
roomlist.append(self.config["parent_room"])
purge_list = {}
error_list = {}
msg = await evt.respond("starting the purge...")
report = await self.generate_report()
purgeable = report['kick_inactive']
roomlist = await self.get_space_roomlist()
# don't forget to kick from the space itself
roomlist.append(self.config["parent_room"])
purge_list = {}
error_list = {}
for user in purgeable:
purge_list[user] = []
for room in roomlist:
try:
@@ -615,12 +710,12 @@ class CommunityBot(Plugin):
roomname = roomnamestate['name']
await self.client.get_state_event(room, EventType.ROOM_MEMBER, user)
await self.client.kick_user(room, user, reason='kicked')
await self.client.kick_user(room, user, reason='inactivity')
if roomname:
purge_list[user].append(roomname)
else:
purge_list[user].append(room)
time.sleep(self.config['sleep'])
time.sleep('sleep')
except MNotFound:
pass
except Exception as e:
@@ -629,110 +724,154 @@ class CommunityBot(Plugin):
error_list[user].append(roomname or room)
results = "the following users were kicked:<p><code>{purge_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(purge_list=purge_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
results = "the following users were purged:<p><code>{purge_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(purge_list=purge_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
# sync our database after we've made changes to room memberships
await self.do_sync()
# sync our database after we've made changes to room memberships
await self.do_sync()
else:
await evt.reply("lol you don't have permission to do that")
@community.subcommand("kick", help='kick a specific user from the community and all rooms')
@command.argument("mxid", "full matrix ID", required=True)
async def kick_user(self, evt: MessageEvent, mxid: UserID) -> None:
await evt.mark_read()
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
user = mxid
msg = await evt.respond("starting the purge...")
roomlist = await self.get_space_roomlist()
# don't forget to kick from the space itself
roomlist.append(self.config["parent_room"])
purge_list = {}
error_list = {}
purge_list[user] = []
for room in roomlist:
try:
roomname = None
roomnamestate = await self.client.get_state_event(room, 'm.room.name')
roomname = roomnamestate['name']
await self.client.get_state_event(room, EventType.ROOM_MEMBER, user)
await self.client.kick_user(room, user, reason='kicked')
if roomname:
purge_list[user].append(roomname)
else:
purge_list[user].append(room)
time.sleep(self.config['sleep'])
except MNotFound:
pass
except Exception as e:
self.log.warning(e)
error_list[user] = []
error_list[user].append(roomname or room)
results = "the following users were kicked:<p><code>{purge_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(purge_list=purge_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
# sync our database after we've made changes to room memberships
await self.do_sync()
@community.subcommand("ban", help='kick and ban a specific user from the community and all rooms')
@command.argument("mxid", "full matrix ID", required=True)
async def ban_user(self, evt: MessageEvent, mxid: UserID) -> None:
await evt.mark_read()
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
user = mxid
msg = await evt.respond("starting the ban...")
results_map = await self.ban_this_user(user, all_rooms=True)
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
user = mxid
msg = await evt.respond("starting the ban...")
results_map = await self.ban_this_user(user, all_rooms=True)
results = "the following users were kicked and banned:<p><code>{ban_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(ban_list=results_map['ban_list'],
error_list=results_map['error_list'])
await evt.respond(results, allow_html=True, edits=msg)
results = "the following users were kicked and banned:<p><code>{ban_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(ban_list=results_map['ban_list'],
error_list=results_map['error_list'])
await evt.respond(results, allow_html=True, edits=msg)
# sync our database after we've made changes to room memberships
await self.do_sync()
else:
await evt.reply("lol you don't have permission to do that")
# sync our database after we've made changes to room memberships
await self.do_sync()
@community.subcommand("unban", help='unban a specific user from the community and all rooms')
@command.argument("mxid", "full matrix ID", required=True)
async def unban_user(self, evt: MessageEvent, mxid: UserID) -> None:
await evt.mark_read()
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
user = mxid
msg = await evt.respond("starting the unban...")
roomlist = await self.get_space_roomlist()
# don't forget to kick from the space itself
roomlist.append(self.config["parent_room"])
unban_list = {}
error_list = {}
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
unban_list[user] = []
for room in roomlist:
try:
roomname = None
roomnamestate = await self.client.get_state_event(room, 'm.room.name')
roomname = roomnamestate['name']
user = mxid
msg = await evt.respond("starting the unban...")
roomlist = await self.get_space_roomlist()
# don't forget to kick from the space itself
roomlist.append(self.config["parent_room"])
unban_list = {}
error_list = {}
await self.client.get_state_event(room, EventType.ROOM_MEMBER, user)
await self.client.unban_user(room, user, reason='unbanned')
if roomname:
unban_list[user].append(roomname)
else:
unban_list[user].append(room)
time.sleep(self.config['sleep'])
except MNotFound:
pass
except Exception as e:
self.log.warning(e)
error_list[user] = []
error_list[user].append(roomname or room)
unban_list[user] = []
for room in roomlist:
try:
roomname = None
roomnamestate = await self.client.get_state_event(room, 'm.room.name')
roomname = roomnamestate['name']
await self.client.get_state_event(room, EventType.ROOM_MEMBER, user)
await self.client.unban_user(room, user, reason='unbanned')
if roomname:
unban_list[user].append(roomname)
else:
unban_list[user].append(room)
time.sleep(self.config['sleep'])
except MNotFound:
pass
except Exception as e:
self.log.warning(e)
error_list[user] = []
error_list[user].append(roomname or room)
results = "the following users were unbanned:<p><code>{unban_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(unban_list=unban_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
results = "the following users were unbanned:<p><code>{unban_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(unban_list=unban_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
# sync our database after we've made changes to room memberships
await self.do_sync()
else:
await evt.reply("lol you don't have permission to do that")
# sync our database after we've made changes to room memberships
await self.do_sync()
@community.subcommand("redact", help="redact messages from a specific user (optionally in a specific room)")
@command.argument("mxid", "full matrix ID", required=True)
@command.argument("room", "room ID", required=False)
async def mark_for_redaction(self, evt: MessageEvent, mxid: UserID, room: str) -> None:
await evt.mark_read()
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
if room:
if room.startswith('#'):
room_id = await self.client.resolve_room_alias(room)
room_id = room_id["room_id"]
else:
room_id = room
else:
room_id = evt.room_id
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
# get list of messages to redact in this room
messages = await self.get_messages_to_redact(room_id, mxid)
for msg in messages:
await self.database.execute(
"INSERT INTO redaction_tasks (event_id, room_id) VALUES ($1, $2)",
msg.event_id,
room_id
)
await evt.respond(f"Queued {len(messages)} messages for redaction in {room_id}")
if room:
if room.startswith('#'):
room_id = await self.client.resolve_room_alias(room)
room_id = room_id["room_id"]
else:
room_id = room
else:
await evt.reply("lol you don't have permission to do that")
room_id = evt.room_id
# get list of messages to redact in this room
messages = await self.get_messages_to_redact(room_id, mxid)
for msg in messages:
await self.database.execute(
"INSERT INTO redaction_tasks (event_id, room_id) VALUES ($1, $2)",
msg.event_id,
room_id
)
await evt.respond(f"Queued {len(messages)} messages for redaction in {room_id}")
@community.subcommand("createroom", help="create a new room titled <roomname> and add it to the parent space. \
optionally include `--encrypt` to encrypt it regardless of the default settings.")
@@ -743,59 +882,60 @@ class CommunityBot(Plugin):
use `--encrypt` to ensure it is encrypted at creation time even if that isnt my default \
setting.')
else:
if evt.sender in self.config["admins"] or evt.sender in self.config["moderators"]:
encrypted_flag_regex = re.compile(r'(\s+|^)-+encrypt(ed)?\s?')
force_encryption = bool(encrypted_flag_regex.search(roomname))
try:
if force_encryption:
roomname = encrypted_flag_regex.sub('', roomname)
sanitized_name = re.sub(r"[^a-zA-Z0-9]", '', roomname).lower()
invitees = self.config['invitees']
parent_room = self.config['parent_room']
## homeserver is derived from maubot's client instance since this is the user that will create the room
server = self.client.parse_user_id(self.client.mxid)[1]
# set bot PL higher than admin so we can kick old admins if needed
pl_override = {"users": {self.client.mxid: 1000}}
for u in self.config['admins']:
pl_override["users"][u] = 100
for u in self.config['moderators']:
pl_override["users"][u] = 50
pl_json = json.dumps(pl_override)
if not await self.user_permitted(evt.sender):
await evt.reply("You don't have permission to use this command")
return
mymsg = await evt.respond(f"creating {sanitized_name}, give me a minute...")
#self.log.info(mymsg)
room_id = await self.client.create_room(alias_localpart=sanitized_name, name=roomname,
invitees=invitees, power_level_override=pl_override)
encrypted_flag_regex = re.compile(r'(\s+|^)-+encrypt(ed)?\s?')
force_encryption = bool(encrypted_flag_regex.search(roomname))
try:
if force_encryption:
roomname = encrypted_flag_regex.sub('', roomname)
sanitized_name = re.sub(r"[^a-zA-Z0-9]", '', roomname).lower()
invitees = self.config['invitees']
parent_room = self.config['parent_room']
## homeserver is derived from maubot's client instance since this is the user that will create the room
server = self.client.parse_user_id(self.client.mxid)[1]
# set bot PL higher than admin so we can kick old admins if needed
pl_override = {"users": {self.client.mxid: 1000}}
for u in self.config['admins']:
pl_override["users"][u] = 100
for u in self.config['moderators']:
pl_override["users"][u] = 50
pl_json = json.dumps(pl_override)
mymsg = await evt.respond(f"creating {sanitized_name}, give me a minute...")
#self.log.info(mymsg)
room_id = await self.client.create_room(alias_localpart=sanitized_name, name=roomname,
invitees=invitees, power_level_override=pl_override)
time.sleep(self.config['sleep'])
await evt.respond(f"updating room states...", edits=mymsg)
parent_event_content = json.dumps({'auto_join': False, 'suggested': False, 'via': [server]})
child_event_content = json.dumps({'canonical': True, 'via': [server]})
join_rules_content = json.dumps({'join_rule': 'restricted', 'allow': [{'type': 'm.room_membership',
'room_id': parent_room}]})
await self.client.send_state_event(parent_room, 'm.space.child', parent_event_content, state_key=room_id)
time.sleep(self.config['sleep'])
await self.client.send_state_event(room_id, 'm.space.parent', child_event_content, state_key=parent_room)
time.sleep(self.config['sleep'])
await self.client.send_state_event(room_id, 'm.room.join_rules', join_rules_content, state_key="")
time.sleep(self.config['sleep'])
if self.config["encrypt"] or force_encryption:
encryption_content = json.dumps({"algorithm": "m.megolm.v1.aes-sha2"})
await self.client.send_state_event(room_id, 'm.room.encryption', encryption_content,
state_key="")
await evt.respond(f"encrypting room...", edits=mymsg)
time.sleep(self.config['sleep'])
await evt.respond(f"updating room states...", edits=mymsg)
parent_event_content = json.dumps({'auto_join': False, 'suggested': False, 'via': [server]})
child_event_content = json.dumps({'canonical': True, 'via': [server]})
join_rules_content = json.dumps({'join_rule': 'restricted', 'allow': [{'type': 'm.room_membership',
'room_id': parent_room}]})
await self.client.send_state_event(parent_room, 'm.space.child', parent_event_content, state_key=room_id)
time.sleep(self.config['sleep'])
await self.client.send_state_event(room_id, 'm.space.parent', child_event_content, state_key=parent_room)
time.sleep(self.config['sleep'])
await self.client.send_state_event(room_id, 'm.room.join_rules', join_rules_content, state_key="")
time.sleep(self.config['sleep'])
if self.config["encrypt"] or force_encryption:
encryption_content = json.dumps({"algorithm": "m.megolm.v1.aes-sha2"})
await self.client.send_state_event(room_id, 'm.room.encryption', encryption_content,
state_key="")
await evt.respond(f"encrypting room...", edits=mymsg)
time.sleep(self.config['sleep'])
await evt.respond(f"room created and updated, alias is #{sanitized_name}:{server}", edits=mymsg)
await evt.respond(f"room created and updated, alias is #{sanitized_name}:{server}", edits=mymsg)
except Exception as e:
await evt.respond(f"i tried, but something went wrong: \"{e}\"", edits=mymsg)
else:
await evt.reply("you're not the boss of me!")
except Exception as e:
await evt.respond(f"i tried, but something went wrong: \"{e}\"", edits=mymsg)
#need to somehow regularly fetch and update the list of room ids that are associated with a given space
@@ -857,68 +997,68 @@ class CommunityBot(Plugin):
@community.subcommand("setpower", help="set power levels according to the community configuration")
async def set_powerlevels(self, evt: MessageEvent,) -> None:
await evt.mark_read()
if evt.sender in self.config["admins"]:
msg = await evt.respond("truing up power levels, this could take a minute...")
admins = self.config['admins']
moderators = self.config['moderators']
roomlist = await self.get_space_roomlist()
# don't forget to include the space itself
roomlist.append(self.config["parent_room"])
success_list = []
error_list = []
adminpl = 100
modpl = 50
defaultpl = 0
if not await self.user_permitted(evt.sender, min_level=100):
await evt.reply("You don't have permission to use this command")
return
for room in roomlist:
# need to get and evaluate the current state that contains powerlevels first
current_pl = await self.client.get_state_event(room, 'm.room.power_levels')
users = current_pl['users'].serialize()
updated_user_map = dict(users)
try:
roomname = None
roomnamestate = await self.client.get_state_event(room, 'm.room.name')
roomname = roomnamestate['name']
except Exception as e:
self.log.warning(e)
msg = await evt.respond("truing up power levels, this could take a minute...")
admins = self.config['admins']
moderators = self.config['moderators']
roomlist = await self.get_space_roomlist()
# don't forget to include the space itself
roomlist.append(self.config["parent_room"])
success_list = []
error_list = []
adminpl = 100
modpl = 50
defaultpl = 0
# update our powerlevel map values
for user in admins:
updated_user_map[user] = adminpl
for user in moderators:
updated_user_map[user] = modpl
for room in roomlist:
# need to get and evaluate the current state that contains powerlevels first
current_pl = await self.client.get_state_event(room, 'm.room.power_levels')
users = current_pl['users'].serialize()
updated_user_map = dict(users)
try:
roomname = None
roomnamestate = await self.client.get_state_event(room, 'm.room.name')
roomname = roomnamestate['name']
except Exception as e:
self.log.warning(e)
# revoke values for people no longer in the config
for user in users.keys():
if ( user not in admins and
user not in moderators and
updated_user_map[user] > defaultpl and
user != self.client.mxid ):
del updated_user_map[user]
# update our powerlevel map values
for user in admins:
updated_user_map[user] = adminpl
for user in moderators:
updated_user_map[user] = modpl
# revoke values for people no longer in the config
for user in users.keys():
if ( user not in admins and
user not in moderators and
updated_user_map[user] > defaultpl and
user != self.client.mxid ):
del updated_user_map[user]
# and send the new state event back to the room
new_pl = current_pl
new_pl['users'] = updated_user_map
try:
#self.log.debug(f"DEBUG sending finalized PL map to room {room}: {updated_user_map}")
await self.client.send_state_event(room, 'm.room.power_levels', new_pl)
success_list.append(roomname or room)
except Exception as e:
self.log.warning(e)
error_list.append(roomname or room)
# and send the new state event back to the room
new_pl = current_pl
new_pl['users'] = updated_user_map
try:
#self.log.debug(f"DEBUG sending finalized PL map to room {room}: {updated_user_map}")
await self.client.send_state_event(room, 'm.room.power_levels', new_pl)
success_list.append(roomname or room)
except Exception as e:
self.log.warning(e)
error_list.append(roomname or room)
time.sleep(self.config['sleep'])
time.sleep(self.config['sleep'])
results = "the following rooms were updated:<p><code>{success_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(success_list=success_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
results = "the following rooms were updated:<p><code>{success_list}</code></p>the following errors were \
recorded:<p><code>{error_list}</code></p>".format(success_list=success_list, error_list=error_list)
await evt.respond(results, allow_html=True, edits=msg)
# sync our database after we've made changes to room memberships
await self.do_sync()
else:
await evt.reply("lol you don't have permission to do that")
# sync our database after we've made changes to room memberships
await self.do_sync()
maubot.yaml
+1 -1
View File
@@ -1,6 +1,6 @@
maubot: 0.1.0
id: org.jobmachine.communitybot
version: 0.1.18
version: 0.1.19
license: MIT
modules:
- community