Dome
edd3eee178
This change introduces native `matrix:` URI-based rendering for `{user}` and `{room}` placeholders,
replacing previous plaintext and matrix.to-based links. Users and rooms are now rendered as clickable
pills in supporting clients, with a clean display using display names and room names (no @/# prefixes).
Reporting, moderation, and auto-redaction messages have been updated to use the same rendering logic.
Inspect and event links now also use native `matrix:` URIs for direct in-client navigation.
Internally, URI generation and rendering logic have been unified via central helper functions,
ensuring consistent handling of user IDs, room IDs, aliases, and event IDs.
This commit also includes a broader refactor of the codebase:
- decomposed complex flows (e.g. join handling) into smaller helpers
- moved mutable class-level state to instance-level
- reduced duplicate API calls and redundant logic
- improved overall structure and maintainability
Test coverage has been extended for URI helpers and rendering logic to prevent regressions.
No breaking changes to existing template parameters like `{user_link}` or `{room_link}`.
277 lines
8.4 KiB
Python
277 lines
8.4 KiB
Python
"""Base command handler for common command patterns."""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any, Optional
|
|
from mautrix.types import MessageEvent, UserID
|
|
from .decorators import require_permission, require_parent_room, handle_errors
|
|
|
|
|
|
class BaseCommandHandler(ABC):
|
|
"""Base class for command handlers with common patterns."""
|
|
|
|
def __init__(self, bot):
|
|
"""Initialize with bot instance.
|
|
|
|
Args:
|
|
bot: CommunityBot instance
|
|
"""
|
|
self.bot = bot
|
|
self.client = bot.client
|
|
self.config = bot.config
|
|
self.config_manager = bot.config_manager
|
|
self.log = bot.log
|
|
self.database = bot.database
|
|
|
|
@abstractmethod
|
|
async def execute(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute the command logic.
|
|
|
|
Args:
|
|
evt: Message event
|
|
*args: Command arguments
|
|
**kwargs: Additional keyword arguments
|
|
|
|
Returns:
|
|
Command result
|
|
"""
|
|
pass
|
|
|
|
async def check_permissions(
|
|
self, evt: MessageEvent, min_level: int = 50, room_id: str = None
|
|
) -> bool:
|
|
"""Check if user has required permissions.
|
|
|
|
Args:
|
|
evt: Message event
|
|
min_level: Minimum required power level
|
|
room_id: Room ID to check permissions in
|
|
|
|
Returns:
|
|
bool: True if user has permissions
|
|
"""
|
|
return await self.bot.user_permitted(evt.sender, min_level, room_id)
|
|
|
|
async def check_parent_room(self, evt: MessageEvent) -> bool:
|
|
"""Check if parent room is configured.
|
|
|
|
Args:
|
|
evt: Message event
|
|
|
|
Returns:
|
|
bool: True if parent room is configured
|
|
"""
|
|
return await self.bot.check_parent_room(evt)
|
|
|
|
async def reply_error(self, evt: MessageEvent, message: str) -> None:
|
|
"""Reply with an error message.
|
|
|
|
Args:
|
|
evt: Message event
|
|
message: Error message
|
|
"""
|
|
await evt.reply(message)
|
|
|
|
async def reply_success(self, evt: MessageEvent, message: str) -> None:
|
|
"""Reply with a success message.
|
|
|
|
Args:
|
|
evt: Message event
|
|
message: Success message
|
|
"""
|
|
await evt.reply(message)
|
|
|
|
async def respond_html(
|
|
self, evt: MessageEvent, message: str, edits: Optional[MessageEvent] = None
|
|
) -> None:
|
|
"""Respond with HTML content.
|
|
|
|
Args:
|
|
evt: Message event
|
|
message: HTML message
|
|
edits: Optional message to edit
|
|
"""
|
|
await evt.respond(message, allow_html=True, edits=edits)
|
|
|
|
def is_tracking_enabled(self) -> bool:
|
|
"""Check if user tracking is enabled.
|
|
|
|
Returns:
|
|
bool: True if tracking is enabled
|
|
"""
|
|
return self.config_manager.is_tracking_enabled()
|
|
|
|
def is_verification_enabled(self) -> bool:
|
|
"""Check if verification is enabled.
|
|
|
|
Returns:
|
|
bool: True if verification is enabled
|
|
"""
|
|
return self.config_manager.is_verification_enabled()
|
|
|
|
def get_parent_room(self) -> Optional[str]:
|
|
"""Get the parent room ID.
|
|
|
|
Returns:
|
|
str: Parent room ID or None
|
|
"""
|
|
return self.config_manager.get_parent_room()
|
|
|
|
|
|
class TrackingCommandHandler(BaseCommandHandler):
|
|
"""Base handler for commands that require user tracking."""
|
|
|
|
async def execute(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute command with tracking check."""
|
|
if not self.is_tracking_enabled():
|
|
await self.reply_error(evt, "user tracking is disabled")
|
|
return
|
|
return await self.execute_tracking_command(evt, *args, **kwargs)
|
|
|
|
@abstractmethod
|
|
async def execute_tracking_command(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute the tracking command logic.
|
|
|
|
Args:
|
|
evt: Message event
|
|
*args: Command arguments
|
|
**kwargs: Additional keyword arguments
|
|
|
|
Returns:
|
|
Command result
|
|
"""
|
|
pass
|
|
|
|
|
|
class AdminCommandHandler(BaseCommandHandler):
|
|
"""Base handler for admin-only commands."""
|
|
|
|
async def execute(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute command with admin permission check."""
|
|
if not await self.check_permissions(evt, min_level=100):
|
|
await self.reply_error(evt, "You don't have permission to use this command")
|
|
return
|
|
return await self.execute_admin_command(evt, *args, **kwargs)
|
|
|
|
@abstractmethod
|
|
async def execute_admin_command(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute the admin command logic.
|
|
|
|
Args:
|
|
evt: Message event
|
|
*args: Command arguments
|
|
**kwargs: Additional keyword arguments
|
|
|
|
Returns:
|
|
Command result
|
|
"""
|
|
pass
|
|
|
|
|
|
class ModeratorCommandHandler(BaseCommandHandler):
|
|
"""Base handler for moderator commands."""
|
|
|
|
async def execute(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute command with moderator permission check."""
|
|
if not await self.check_permissions(evt, min_level=50):
|
|
await self.reply_error(evt, "You don't have permission to use this command")
|
|
return
|
|
return await self.execute_moderator_command(evt, *args, **kwargs)
|
|
|
|
@abstractmethod
|
|
async def execute_moderator_command(
|
|
self, evt: MessageEvent, *args, **kwargs
|
|
) -> Any:
|
|
"""Execute the moderator command logic.
|
|
|
|
Args:
|
|
evt: Message event
|
|
*args: Command arguments
|
|
**kwargs: Additional keyword arguments
|
|
|
|
Returns:
|
|
Command result
|
|
"""
|
|
pass
|
|
|
|
|
|
class SpaceCommandHandler(BaseCommandHandler):
|
|
"""Base handler for commands that require parent space."""
|
|
|
|
async def execute(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute command with parent space check."""
|
|
if not await self.check_parent_room(evt):
|
|
return
|
|
return await self.execute_space_command(evt, *args, **kwargs)
|
|
|
|
@abstractmethod
|
|
async def execute_space_command(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute the space command logic.
|
|
|
|
Args:
|
|
evt: Message event
|
|
*args: Command arguments
|
|
**kwargs: Additional keyword arguments
|
|
|
|
Returns:
|
|
Command result
|
|
"""
|
|
pass
|
|
|
|
|
|
class SpaceModeratorCommandHandler(SpaceCommandHandler, ModeratorCommandHandler):
|
|
"""Base handler for commands that require both parent space and moderator permissions."""
|
|
|
|
async def execute(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute command with both space and moderator checks."""
|
|
if not await self.check_parent_room(evt):
|
|
return
|
|
if not await self.check_permissions(evt, min_level=50):
|
|
await self.reply_error(evt, "You don't have permission to use this command")
|
|
return
|
|
return await self.execute_space_moderator_command(evt, *args, **kwargs)
|
|
|
|
@abstractmethod
|
|
async def execute_space_moderator_command(
|
|
self, evt: MessageEvent, *args, **kwargs
|
|
) -> Any:
|
|
"""Execute the space moderator command logic.
|
|
|
|
Args:
|
|
evt: Message event
|
|
*args: Command arguments
|
|
**kwargs: Additional keyword arguments
|
|
|
|
Returns:
|
|
Command result
|
|
"""
|
|
pass
|
|
|
|
|
|
class SpaceAdminCommandHandler(SpaceCommandHandler, AdminCommandHandler):
|
|
"""Base handler for commands that require both parent space and admin permissions."""
|
|
|
|
async def execute(self, evt: MessageEvent, *args, **kwargs) -> Any:
|
|
"""Execute command with both space and admin checks."""
|
|
if not await self.check_parent_room(evt):
|
|
return
|
|
if not await self.check_permissions(evt, min_level=100):
|
|
await self.reply_error(evt, "You don't have permission to use this command")
|
|
return
|
|
return await self.execute_space_admin_command(evt, *args, **kwargs)
|
|
|
|
@abstractmethod
|
|
async def execute_space_admin_command(
|
|
self, evt: MessageEvent, *args, **kwargs
|
|
) -> Any:
|
|
"""Execute the space admin command logic.
|
|
|
|
Args:
|
|
evt: Message event
|
|
*args: Command arguments
|
|
**kwargs: Additional keyword arguments
|
|
|
|
Returns:
|
|
Command result
|
|
"""
|
|
pass
|