refactor: Better plugin management in groupbot.

Author: iphydf

tools/groupbot/api.py

@@ -109,6 +109,7 @@ def save(self) -> None:
 @dataclass
 class Reply:
     text: str
+    message_type: core.Tox_Message_Type = core.TOX_MESSAGE_TYPE_NORMAL
 
 
 T = TypeVar("T")
@@ -135,3 +136,30 @@ def wrapper(
         return func(self, bot, friend_pk, message_type, params)
 
     return wrapper
+
+
+@dataclass
+class HandlerData:
+    """Serializable handler data."""
+
+    data: dict[str, str]
+
+    def __init__(self, **data: str) -> None:
+        self.data = data
+
+
+class Handler:
+
+    def handle(
+        self,
+        bot: GroupBot,
+        friend_pk: bytes,
+        message_type: core.Tox_Message_Type,
+        message: tuple[str, ...],
+    ) -> Optional[Reply]:
+        """Handle a Tox message."""
+        raise NotImplementedError
+
+    def data(self) -> HandlerData:
+        """Get the handler data usable to clone a handler."""
+        return HandlerData()

tools/groupbot/commands.py

@@ -3,31 +3,53 @@
 import importlib
 import shlex
 import sys
+from dataclasses import dataclass
+from typing import Callable
 from typing import Optional
 
 from py_toxcore_c.tools.groupbot import api
+from py_toxcore_c.tools.groupbot.plugins import echo
 from py_toxcore_c.tools.groupbot.plugins import github
 
 import pytox.toxcore.tox as core
 
 
-def _reload_modules() -> None:
+@dataclass
+class Module:
+    """A groupbot module."""
+
+    prefix: str
+    new: Callable[[api.Config], api.Handler]
+    clone: Callable[[api.HandlerData], api.Handler]
+
+
+def _reload_modules() -> tuple[Module, ...]:
     """Reload the extension modules."""
     for name in tuple(sys.modules.keys()):
         if name.startswith("py_toxcore_c.tools.groupbot.plugins."):
             importlib.reload(sys.modules[name])
 
+    return (
+        Module("echo", echo.Echo.new, echo.Echo.clone),
+        Module("gh", github.GitHub.new, github.GitHub.clone),
+    )
+
 
 class Commands:
-    _gh: github.GitHub
+    _handlers: dict[str, api.Handler]
 
     def __init__(self, config: api.Config) -> None:
-        self._gh = github.GitHub(config.github_path)
+        self._handlers = {
+            mod.prefix: mod.new(config)
+            for mod in _reload_modules()
+        }
 
     def _reload(self) -> None:
         """Reload the extension modules."""
-        _reload_modules()
-        self._gh = github.GitHub(self._gh.path)
+        self._handlers = {
+            mod.prefix: mod.clone(self._handlers[mod.prefix].data())
+            for mod in _reload_modules()
+        }
 
     def handle(
         self,
@@ -44,19 +66,14 @@ def handle(
         if not command:
             return None
         try:
-            fun = self._dispatch(command[0])
-            if fun:
-                return fun(bot, friend_pk, message_type, tuple(command[1:]))
+            return self._dispatch(command[0])(bot, friend_pk, message_type,
+                                              tuple(command[1:]))
         except Exception as e:
             return api.Reply(f"Error: {e}")
         return None
 
-    def _dispatch(self, command: str) -> Optional[api.CommandFunction]:
+    def _dispatch(self, command: str) -> api.CommandFunction:
         """Dispatch a command."""
-        if command == "echo":
-            return self.echo
-        if command == "gh":
-            return self.gh
         if command == "leave":
             return self.leave
         if command == "nick":
@@ -65,27 +82,19 @@ def _dispatch(self, command: str) -> Optional[api.CommandFunction]:
             return self.reload
         if command == "save":
             return self.save
-        return None
-
-    def echo(
-        self,
-        bot: api.GroupBot,
-        friend_pk: bytes,
-        message_type: core.Tox_Message_Type,
-        message: tuple[str, ...],
-    ) -> Optional[api.Reply]:
-        """Echo a message."""
-        return api.Reply(str(list(message)))
+        if command in self._handlers:
+            return self._handlers[command].handle
+        return self.null
 
-    def gh(
+    def null(
         self,
         bot: api.GroupBot,
         friend_pk: bytes,
         message_type: core.Tox_Message_Type,
         params: tuple[str, ...],
     ) -> Optional[api.Reply]:
-        """Provide access to GitHub issue/PR information."""
-        return self._gh.handle(bot, friend_pk, message_type, params)
+        """Null command."""
+        return None
 
     @api.admin
     def leave(

tools/groupbot/groupbot.py

@@ -110,7 +110,7 @@ def handle_group_message(
     ) -> None:
         print("Group message:", group_number, peer_id, message_type.name,
               message.decode())
-        peer = self.group_peer_get_name(group_number, peer_id)
+        peer = self.group_peer_get_name(group_number, peer_id).decode()
         if not self.conference_chatlist:
             return
         if message.startswith(b"~"):
@@ -125,13 +125,12 @@ def handle_group_message(
             if reply:
                 self.group_send_message(
                     group_number,
-                    core.TOX_MESSAGE_TYPE_NORMAL,
-                    f"{peer.decode()}: {reply.text}".encode(),
+                    reply.message_type,
+                    f"{peer}: {reply.text}".encode(),
                 )
         else:
             self.conference_send_message(
-                0, message_type,
-                f"<{peer.decode()}> {message.decode()}".encode())
+                0, message_type, f"<{peer}> {message.decode()}".encode())
 
     def handle_conference_message(
         self,
@@ -164,7 +163,7 @@ def handle_conference_message(
             if reply:
                 self.conference_send_message(
                     conference_number,
-                    core.TOX_MESSAGE_TYPE_NORMAL,
+                    reply.message_type,
                     f"{peer}: {reply.text}".encode(),
                 )
         else:

tools/groupbot/groupbot.tox

@@ -0,0 +1,30 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+# Copyright © 2025 The TokTok team
+from typing import Optional
+
+from py_toxcore_c.tools.groupbot import api
+
+import pytox.toxcore.tox as core
+
+
+class Echo(api.Handler):
+
+    @staticmethod
+    def new(config: api.Config) -> "Echo":
+        """Create a Echo module instance from a configuration."""
+        return Echo()
+
+    @staticmethod
+    def clone(data: api.HandlerData) -> "Echo":
+        """Clone a module instance."""
+        return Echo()
+
+    def handle(
+        self,
+        bot: api.GroupBot,
+        friend_pk: bytes,
+        message_type: core.Tox_Message_Type,
+        message: tuple[str, ...],
+    ) -> Optional[api.Reply]:
+        """Handle a Tox message."""
+        return api.Reply(str(list(message)), message_type)

tools/groupbot/plugins/echo.py

@@ -77,12 +77,20 @@ class Issue:
     title: str
     state: str
     user: User
-    is_pull_request: bool
 
     @property
     def repo(self) -> str:
         return self.path.repo.name
 
+    @property
+    def is_pull_request(self) -> bool:
+        return self.path.is_pull_request
+
+    @property
+    def emoji(self) -> str:
+        """Get an emoji to distinguish between issue and PR."""
+        return "🎁" if self.is_pull_request else "🐛"
+
     @staticmethod
     def fromJSON(path: IssuePath, issue: dict[str, Any]) -> "Issue":
         return Issue(
@@ -91,14 +99,29 @@ def fromJSON(path: IssuePath, issue: dict[str, Any]) -> "Issue":
             title=str(issue["title"]),
             state=str(issue["state"]),
             user=User.fromJSON(issue["user"]),
-            is_pull_request=bool(issue.get("pull_request")),
         )
 
 
 @dataclass
-class GitHub:
+class GitHub(api.Handler):
     path: str
 
+    @staticmethod
+    def new(config: api.Config) -> "GitHub":
+        """Create a GitHub instance from a configuration."""
+        return GitHub(config.github_path)
+
+    @staticmethod
+    def clone(data: api.HandlerData) -> "GitHub":
+        """Clone a GitHub instance."""
+        if "path" not in data.data:
+            raise ValueError("Missing path in HandlerData")
+        return GitHub(data.data["path"])
+
+    def data(self) -> api.HandlerData:
+        """Get the module data."""
+        return api.HandlerData(path=self.path)
+
     def __hash__(self) -> int:
         return hash(self.path)
 
@@ -127,43 +150,55 @@ def handle_cli(self, message: tuple[str, ...]) -> Optional[api.Reply]:
 
         return None
 
-    def handle_issue(self, repo_name: Optional[str],
-                     issue_id: str) -> api.Reply:
-        """Handle an issue/PR number."""
-        try:
-            issue_number = int(issue_id)
-        except ValueError:
-            return api.Reply(f"Error: {issue_id} is not a valid issue number")
-
-        issues: list[Issue] = []
+    def _find_issue(
+            self, repo_name: Optional[str],
+            issue_number: int) -> tuple[Optional[str], Optional[Issue]]:
+        """Find an issue/PR by number."""
+        candidates: list[Issue] = []
         repos = (self.repos() if repo_name is None else [
             repo for repo in self.repos()
-            if repo.name.lower() == repo_name.lower()
+            if repo.name.lower().startswith(repo_name.lower())
         ])
         if not repos:
-            return api.Reply(f"Error: Repository {repo_name} not found")
+            return None, None
         for repo in repos:
             for issue_path in self.issues(repo):
                 if issue_path.number == issue_number:
-                    issues.append(self.load_issue(issue_path))
+                    candidates.append(self.load_issue(issue_path))
 
+        if len(repos) == 1:
+            repo_name = repos[0].name
         repo_name = repo_name or "any repository"
 
         # If any issues were found, return the first one that's open. If none
         # are open, return the first one.
         issue: Optional[Issue] = None
-        for candidate in issues:
+        for candidate in candidates:
             if candidate.state == "open":
+                issue = candidate
                 break
         else:
-            issue = issues[0] if issues else None
-        if issue:
+            issue = candidates[0] if candidates else None
+
+        return repo_name, issue
+
+    def handle_issue(self, repo_name: Optional[str],
+                     issue_id: str) -> api.Reply:
+        """Handle an issue/PR number."""
+        try:
+            issue_number = int(issue_id)
+        except ValueError:
+            return api.Reply(f"Error: {issue_id} is not a valid issue number")
+
+        found_repo, issue = self._find_issue(repo_name, issue_number)
+        if not found_repo:
+            return api.Reply(f"Error: Repository {repo_name} not found")
+        if not issue:
             return api.Reply(
-                f"{issue.title} by {issue.user.login} ({issue.repo}#{issue.number}, {issue.state})"
-            )
+                f"Error: Issue {issue_number} not found in {found_repo}")
 
-        return api.Reply(
-            f"Error: Issue {issue_number} not found in {repo_name}")
+        return api.Reply(f"{issue.emoji} {issue.title} by {issue.user.login} "
+                         f"({issue.repo}#{issue.number}, {issue.state})")
 
     @memoize
     def load_issue(self, issue: IssuePath) -> Issue: