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

formatting

Browse Source
This commit is contained in:
William Kray
2025-09-09 21:21:51 -07:00
parent 5f42420619
commit a92759c100
14 changed files with 1380 additions and 967 deletions
Download Patch File Download Diff File Expand all files Collapse all files
community/helpers/room_creation_utils.py
+88 -86
View File
@@ -8,17 +8,15 @@ from mautrix.client import Client
async def validate_room_creation_params(
roomname: str,
config: dict,
evt: Optional[MessageEvent] = None
roomname: str, config: dict, evt: Optional[MessageEvent] = None
) -> Tuple[str, bool, bool, str]:
"""Validate and process room creation parameters.
Args:
roomname: Original room name
config: Bot configuration
evt: Optional MessageEvent for error responses
Returns:
Tuple of (sanitized_name, force_encryption, force_unencryption, error_msg)
"""
@@ -27,23 +25,23 @@ async def validate_room_creation_params(
unencrypted_flag_regex = re.compile(r"(\s+|^)-+unencrypt(ed)?(\s+|$)")
force_encryption = bool(encrypted_flag_regex.search(roomname))
force_unencryption = bool(unencrypted_flag_regex.search(roomname))
# Clean up room name
if force_encryption:
roomname = encrypted_flag_regex.sub("", roomname) # Remove encryption flag
if force_unencryption:
roomname = unencrypted_flag_regex.sub("", roomname) # Remove unencryption flag
# Clean up any extra whitespace
roomname = re.sub(r"\s+", " ", roomname).strip()
sanitized_name = re.sub(r"[^a-zA-Z0-9]", "", roomname).lower()
# Check if community slug is configured
if not config.get("community_slug", ""):
error_msg = "No community slug configured. Please run initialize command first."
return sanitized_name, force_encryption, force_unencryption, error_msg, roomname
return sanitized_name, force_encryption, force_unencryption, "", roomname
@@ -51,27 +49,27 @@ async def prepare_room_creation_data(
sanitized_name: str,
config: dict,
client: Client,
invitees: Optional[List[str]] = None
invitees: Optional[List[str]] = None,
) -> Tuple[str, str, List[str], str]:
"""Prepare data needed for room creation.
Args:
sanitized_name: Sanitized room name
config: Bot configuration
client: Matrix client
invitees: Optional list of users to invite
Returns:
Tuple of (alias_localpart, server, room_invitees, parent_room)
"""
# Create alias with community slug
alias_localpart = f"{sanitized_name}-{config.get('community_slug', '')}"
# Get server and invitees
server = client.parse_user_id(client.mxid)[1]
room_invitees = invitees if invitees is not None else config.get("invitees", [])
parent_room = config.get("parent_room", "")
return alias_localpart, server, room_invitees, parent_room
@@ -79,34 +77,38 @@ async def prepare_power_levels(
client: Client,
config: dict,
parent_room: str,
power_level_override: Optional[PowerLevelStateEventContent] = None
power_level_override: Optional[PowerLevelStateEventContent] = None,
) -> PowerLevelStateEventContent:
"""Prepare power levels for room creation.
Args:
client: Matrix client
config: Bot configuration
parent_room: Parent room ID
power_level_override: Optional existing power level override
Returns:
PowerLevelStateEventContent for room creation
"""
if power_level_override:
return power_level_override
if parent_room:
try:
# Get parent room power levels to extract user power levels
parent_power_levels = await client.get_state_event(
parent_room, EventType.ROOM_POWER_LEVELS
)
# Create new power levels with server defaults, not copying all permissions from space
power_levels = PowerLevelStateEventContent()
# Copy only user power levels from parent space, not the entire permission set
if parent_power_levels and hasattr(parent_power_levels, 'users') and parent_power_levels.users:
if (
parent_power_levels
and hasattr(parent_power_levels, "users")
and parent_power_levels.users
):
try:
user_power_levels = parent_power_levels.users.copy()
# Ensure bot has highest power
@@ -121,10 +123,10 @@ async def prepare_power_levels(
power_levels.users = {
client.mxid: 1000, # Bot gets highest power
}
# Set explicit config values
power_levels.invite = config.get("invite_power_level", 50)
return power_levels
except Exception as e:
# If we can't get parent power levels, create default ones
@@ -150,10 +152,10 @@ def prepare_initial_state(
server: str,
force_encryption: bool,
force_unencryption: bool,
creation_content: Optional[Dict[str, Any]] = None
creation_content: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
"""Prepare initial state events for room creation.
Args:
config: Bot configuration
parent_room: Parent room ID
@@ -161,66 +163,67 @@ def prepare_initial_state(
force_encryption: Whether to force encryption
force_unencryption: Whether to force no encryption
creation_content: Optional creation content
Returns:
List of initial state events
"""
initial_state = []
# Only add space parent state if we have a parent room
if parent_room:
initial_state.extend([
{
"type": str(EventType.SPACE_PARENT),
"state_key": parent_room,
"content": {
"via": [server],
"canonical": True
}
},
{
"type": str(EventType.ROOM_JOIN_RULES),
"content": {
"join_rule": "restricted",
"allow": [{
"type": "m.room_membership",
"room_id": parent_room
}]
}
}
])
initial_state.extend(
[
{
"type": str(EventType.SPACE_PARENT),
"state_key": parent_room,
"content": {"via": [server], "canonical": True},
},
{
"type": str(EventType.ROOM_JOIN_RULES),
"content": {
"join_rule": "restricted",
"allow": [
{"type": "m.room_membership", "room_id": parent_room}
],
},
},
]
)
# Add encryption if needed
if (config.get("encrypt", False) and not force_unencryption) or force_encryption:
initial_state.append({
"type": str(EventType.ROOM_ENCRYPTION),
"content": {
"algorithm": "m.megolm.v1.aes-sha2"
initial_state.append(
{
"type": str(EventType.ROOM_ENCRYPTION),
"content": {"algorithm": "m.megolm.v1.aes-sha2"},
}
})
)
# Add history visibility if specified in creation_content
if creation_content and "m.room.history_visibility" in creation_content:
initial_state.append({
"type": str(EventType.ROOM_HISTORY_VISIBILITY),
"content": {
"history_visibility": creation_content.get("m.room.history_visibility", "joined")
initial_state.append(
{
"type": str(EventType.ROOM_HISTORY_VISIBILITY),
"content": {
"history_visibility": creation_content.get(
"m.room.history_visibility", "joined"
)
},
}
})
)
return initial_state
def adjust_power_levels_for_modern_rooms(
power_levels: PowerLevelStateEventContent,
room_version: str
power_levels: PowerLevelStateEventContent, room_version: str
) -> PowerLevelStateEventContent:
"""Adjust power levels for modern room versions.
Args:
power_levels: Power level state content
room_version: Room version string
Returns:
Adjusted power level state content
"""
@@ -229,20 +232,18 @@ def adjust_power_levels_for_modern_rooms(
if room_version and int(room_version) >= 12 and power_levels:
if power_levels.users:
# Remove bot from users list but keep other important settings
power_levels.users.pop("bot_mxid", None) # Will be replaced with actual bot mxid
power_levels.users.pop(
"bot_mxid", None
) # Will be replaced with actual bot mxid
return power_levels
async def add_room_to_space(
client: Client,
parent_room: str,
room_id: str,
server: str,
sleep_duration: float
client: Client, parent_room: str, room_id: str, server: str, sleep_duration: float
) -> None:
"""Add created room to parent space.
Args:
client: Matrix client
parent_room: Parent room ID
@@ -254,23 +255,17 @@ async def add_room_to_space(
await client.send_state_event(
parent_room,
EventType.SPACE_CHILD,
{
"via": [server],
"suggested": False
},
state_key=room_id
{"via": [server], "suggested": False},
state_key=room_id,
)
await asyncio.sleep(sleep_duration)
async def verify_room_creation(
client: Client,
room_id: str,
expected_version: str,
logger
client: Client, room_id: str, expected_version: str, logger
) -> None:
"""Verify that room was created with correct settings.
Args:
client: Matrix client
room_id: Created room ID
@@ -279,9 +274,16 @@ async def verify_room_creation(
"""
try:
from .room_utils import get_room_version_and_creators
actual_version, actual_creators = await get_room_version_and_creators(client, room_id, logger)
logger.info(f"Room {room_id} created with version {actual_version} (requested: {expected_version})")
actual_version, actual_creators = await get_room_version_and_creators(
client, room_id, logger
)
logger.info(
f"Room {room_id} created with version {actual_version} (requested: {expected_version})"
)
if actual_version != expected_version:
logger.warning(f"Room version mismatch: requested {expected_version}, got {actual_version}")
logger.warning(
f"Room version mismatch: requested {expected_version}, got {actual_version}"
)
except Exception as e:
logger.warning(f"Could not verify room version for {room_id}: {e}")