William Kray
267 lines
8.6 KiB
Python
267 lines
8.6 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
|