Basic ext.commands impl

twitchio/ext/commands/__init__.py

@@ -23,3 +23,7 @@
 """
 
 from .bot import Bot as Bot
+from .cogs import *
+from .context import *
+from .core import *
+from .exceptions import *

twitchio/ext/commands/bot.py

@@ -22,19 +22,36 @@
 SOFTWARE.
 """
 
-from typing import Unpack
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Unpack
 
 from twitchio.client import Client
-from twitchio.types_.options import ClientOptions
+
+from .context import Context
+from .core import CommandErrorPayload, Mixin
+from .exceptions import *
+
+
+if TYPE_CHECKING:
+    from models.eventsub_ import ChatMessage
+    from types_.options import Prefix_T
+
+    from twitchio.types_.options import ClientOptions
 
 
-class Bot(Client):
+logger: logging.Logger = logging.getLogger(__name__)
+
+
+class Bot(Mixin[None], Client):
     def __init__(
         self,
         *,
         client_id: str,
         client_secret: str,
         bot_id: str,
+        prefix: Prefix_T,
         **options: Unpack[ClientOptions],
     ) -> None:
         super().__init__(
@@ -44,7 +61,38 @@ def __init__(
             **options,
         )
 
+        self._get_prefix: Prefix_T = prefix
+
     @property
     def bot_id(self) -> str:
         assert self._bot_id
         return self._bot_id
+
+    async def _process_commands(self, message: ChatMessage) -> None:
+        ctx: Context = Context(message, bot=self)
+
+        try:
+            await self.invoke(ctx)
+        except CommandError as e:
+            payload = CommandErrorPayload(context=ctx, exception=e)
+            self.dispatch("command_error", payload=payload)
+
+    async def process_commands(self, message: ChatMessage) -> None:
+        await self._process_commands(message)
+
+    async def invoke(self, ctx: Context) -> None:
+        await ctx.invoke()
+
+    async def event_channel_chat_message(self, payload: ChatMessage) -> None:
+        if payload.chatter.id == self.bot_id:
+            return
+
+        await self.process_commands(payload)
+
+    async def event_command_error(self, payload: CommandErrorPayload) -> None:
+        msg = f'Ignoring exception in command "{payload.context.command}":\n'
+        logger.error(msg, exc_info=payload.exception)
+
+    async def before_invoke_hook(self, ctx: Context) -> None: ...
+
+    async def after_invoke_hook(self, ctx: Context) -> None: ...

twitchio/ext/commands/cogs.py

@@ -0,0 +1,32 @@
+"""
+MIT License
+
+Copyright (c) 2017 - Present PythonistaGuild
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+"""
+
+from .core import CommandErrorPayload
+
+
+__all__ = ("Cog",)
+
+
+class Cog:
+    async def cog_command_error(self, payload: CommandErrorPayload) -> None: ...

twitchio/ext/commands/context.py

@@ -0,0 +1,187 @@
+"""
+MIT License
+
+Copyright (c) 2017 - Present PythonistaGuild
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, Any
+
+from .core import Command, Group
+from .exceptions import *
+from .view import StringView
+
+
+__all__ = ("Context",)
+
+
+if TYPE_CHECKING:
+    from models.eventsub_ import ChatMessage
+    from types_.options import Prefix_T
+    from user import PartialUser
+
+    from .bot import Bot
+
+
+class Context:
+    def __init__(self, message: ChatMessage, bot: Bot) -> None:
+        self._message: ChatMessage = message
+        self._bot: Bot = bot
+        self._prefix: str | None = None
+
+        self._raw_content: str = self._message.text
+        self._command: Command[Any, ...] | None = None
+        self._invoked_with: str | None = None
+        self._command_failed: bool = False
+
+        self._view: StringView = StringView(self._raw_content)
+
+    @property
+    def message(self) -> ChatMessage:
+        return self._message
+
+    @property
+    def command(self) -> Command[Any, ...] | None:
+        return self._command
+
+    @property
+    def chatter(self) -> PartialUser:
+        return self._message.chatter
+
+    @property
+    def broadcaster(self) -> PartialUser:
+        return self._message.broadcaster
+
+    @property
+    def channel(self) -> PartialUser:
+        return self.broadcaster
+
+    @property
+    def bot(self) -> Bot:
+        return self._bot
+
+    @property
+    def prefix(self) -> str | None:
+        return self._prefix
+
+    @property
+    def content(self) -> str:
+        return self._raw_content
+
+    def is_valid(self) -> bool:
+        return self._prefix is not None
+
+    def _validate_prefix(self, potential: str | Iterable[str]) -> None:
+        text: str = self._message.text
+
+        if isinstance(potential, str):
+            if text.startswith(potential):
+                self._prefix = potential
+
+            return
+
+        for prefix in tuple(potential):
+            if not isinstance(prefix, str):  # type: ignore
+                msg = f'Command prefix in iterable or iterable returned from coroutine must be "str", not: {type(prefix)}'
+                raise PrefixError(msg)
+
+            if text.startswith(prefix):
+                self._prefix = prefix
+                return
+
+    async def _get_prefix(self) -> None:
+        assigned: Prefix_T = self._bot._get_prefix
+        potential: str | Iterable[str]
+
+        if callable(assigned):
+            potential = await assigned(self._bot, self._message)
+        else:
+            potential = assigned
+
+        if not isinstance(potential, Iterable):  # type: ignore
+            msg = f'Command prefix must be a "str", "Iterable[str]" or a coroutine returning either. Not: {type(potential)}'
+            raise PrefixError(msg)
+
+        self._validate_prefix(potential)
+
+    def _get_command(self) -> None:
+        if not self.prefix:
+            return
+
+        commands = self._bot._commands
+        self._view.skip_string(self.prefix)
+
+        next_ = self._view.get_word()
+        self._invoked_with = next_
+        command = commands.get(next_)
+
+        if not command:
+            return
+
+        if isinstance(command, Group):
+            ...
+
+        else:
+            self._command = command
+            return
+
+    async def _prepare(self) -> None:
+        await self._get_prefix()
+        self._get_command()
+
+    async def prepare(self) -> None:
+        await self._prepare()
+
+    async def invoke(self) -> None:
+        await self.prepare()
+
+        if not self.is_valid():
+            return
+
+        if not self._command:
+            raise CommandNotFound(f'The command "{self._invoked_with}" was not found.')
+
+        try:
+            await self._bot.before_invoke_hook(self)
+        except Exception as e:
+            raise CommandHookError(str(e), e) from e
+
+        # TODO: Payload...
+        self.bot.dispatch("command_invoked")
+
+        try:
+            await self._command._invoke(self)
+        except CommandError as e:
+            await self._command._dispatch_error(self, e)
+            return
+
+        try:
+            await self._bot.after_invoke_hook(self)
+        except Exception as e:
+            raise CommandHookError(str(e), e) from e
+
+        # TODO: Payload...
+        self.bot.dispatch("command_completed")
+
+    async def send(self, content: str) -> None:
+        await self.channel.send_message(sender_id=self.bot.bot_id, message=content)