username/derp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add hot-reload, shorthand commands, and plugin help

Browse Source 
- Plugin registry: add unload_plugin(), reload_plugin(), path tracking
- Bot: add load_plugin(), reload_plugin(), unload_plugin() public API
- Core plugin: add !load, !reload, !unload, !plugins commands
- Command dispatch: support unambiguous prefix matching (!h -> !help)
- Help: support !help <plugin> to show plugin description and commands
- Tests: 17 new tests covering hot-reload, prefix matching

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-15 01:15:59 +01:00
parent ad18a902dd
commit 77f9a364e6
6 changed files with 508 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
src/derp/bot.py
View File

@@ -7,11 +7,12 @@ import logging
from pathlib import Path
from derp.irc import IRCConnection, Message, format_msg, parse
from derp.plugin import PluginRegistry
from derp.plugin import Handler, PluginRegistry
log = logging.getLogger(__name__)
RECONNECT_DELAY = 30
_AMBIGUOUS = object() # sentinel for ambiguous prefix matches
class Bot:
@@ -108,15 +109,39 @@ class Bot:
parts = text[len(self.prefix):].split(None, 1)
cmd_name = parts[0].lower() if parts else ""
handler = self.registry.commands.get(cmd_name)
handler = self._resolve_command(cmd_name)
if handler is None:
return
if handler is _AMBIGUOUS:
matches = [k for k in self.registry.commands if k.startswith(cmd_name)]
names = ", ".join(self.prefix + m for m in sorted(matches))
await self.reply(msg, f"Ambiguous command '{self.prefix}{cmd_name}': {names}")
return
try:
await handler.callback(self, msg)
except Exception:
log.exception("error in command handler '%s'", cmd_name)
def _resolve_command(self, name: str) -> Handler | None:
"""Resolve a command name, supporting unambiguous prefix matching.
Returns the Handler on exact or unique prefix match, the sentinel
``_AMBIGUOUS`` if multiple commands match, or None if nothing matches.
"""
# Exact match takes priority
handler = self.registry.commands.get(name)
if handler is not None:
return handler
# Prefix match
matches = [v for k, v in self.registry.commands.items() if k.startswith(name)]
if len(matches) == 1:
return matches[0]
if len(matches) > 1:
return _AMBIGUOUS # type: ignore[return-value]
return None
# -- Public API for plugins --
async def send(self, target: str, text: str) -> None:
@@ -158,3 +183,32 @@ class Bot:
plugins_dir = self.config["bot"].get("plugins_dir", "plugins")
path = Path(plugins_dir)
self.registry.load_directory(path)
@property
def plugins_dir(self) -> Path:
"""Resolved path to the plugins directory."""
return Path(self.config["bot"].get("plugins_dir", "plugins"))
def load_plugin(self, name: str) -> tuple[bool, str]:
"""Hot-load a new plugin by name from the plugins directory."""
if name in self.registry._modules:
return False, f"plugin already loaded: {name}"
path = self.plugins_dir / f"{name}.py"
if not path.is_file():
return False, f"{name}.py not found"
count = self.registry.load_plugin(path)
if count < 0:
return False, f"failed to load {name}"
return True, f"{count} handlers"
def reload_plugin(self, name: str) -> tuple[bool, str]:
"""Reload a plugin, picking up any file changes."""
return self.registry.reload_plugin(name)
def unload_plugin(self, name: str) -> tuple[bool, str]:
"""Unload a plugin, removing all its handlers."""
if self.registry.unload_plugin(name):
return True, ""
if name == "core":
return False, "cannot unload core"
return False, f"plugin not loaded: {name}"

99
src/derp/plugin.py
View File

@@ -2,9 +2,10 @@
from __future__ import annotations
import importlib.util
import inspect
import logging
import sys
import types
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
@@ -64,6 +65,7 @@ class PluginRegistry:
self.commands: dict[str, Handler] = {}
self.events: dict[str, list[Handler]] = {}
self._modules: dict[str, Any] = {}
self._paths: dict[str, Path] = {}
def register_command(self, name: str, callback: Callable, help: str = "",
plugin: str = "") -> None:
@@ -98,23 +100,30 @@ class PluginRegistry:
count += 1
return count
def load_plugin(self, path: Path) -> None:
"""Load a single plugin from a .py file."""
def load_plugin(self, path: Path) -> int:
"""Load a single plugin from a .py file.
Returns the number of handlers registered, or -1 on failure.
"""
plugin_name = path.stem
if plugin_name.startswith("_"):
return
return -1
try:
spec = importlib.util.spec_from_file_location(f"derp.plugins.{plugin_name}", path)
if spec is None or spec.loader is None:
log.error("failed to create spec for %s", path)
return
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
mod_name = f"derp.plugins.{plugin_name}"
source = path.read_text(encoding="utf-8")
code = compile(source, str(path), "exec")
module = types.ModuleType(mod_name)
module.__file__ = str(path)
sys.modules[mod_name] = module
exec(code, module.__dict__) # noqa: S102
count = self._scan_module(module, plugin_name)
self._modules[plugin_name] = module
self._paths[plugin_name] = path.resolve()
log.info("loaded plugin: %s (%d handlers)", plugin_name, count)
return count
except Exception:
log.exception("failed to load plugin: %s", path)
return -1
def load_directory(self, dir_path: Path) -> None:
"""Load all .py plugin files from a directory."""
@@ -123,3 +132,73 @@ class PluginRegistry:
return
for path in sorted(dir_path.glob("*.py")):
self.load_plugin(path)
def unload_plugin(self, name: str) -> bool:
"""Unload a plugin, removing all its handlers.
Returns True if the plugin was found and unloaded, False otherwise.
Refuses to unload the ``core`` plugin.
"""
if name == "core":
log.warning("refusing to unload core plugin")
return False
if name not in self._modules:
return False
# Remove commands belonging to this plugin
self.commands = {
k: v for k, v in self.commands.items() if v.plugin != name
}
# Remove event handlers belonging to this plugin
for event_type in list(self.events):
self.events[event_type] = [
h for h in self.events[event_type] if h.plugin != name
]
if not self.events[event_type]:
del self.events[event_type]
# Clean up module references
mod_name = f"derp.plugins.{name}"
sys.modules.pop(mod_name, None)
del self._modules[name]
del self._paths[name]
log.info("unloaded plugin: %s", name)
return True
def reload_plugin(self, name: str) -> tuple[bool, str]:
"""Reload a plugin from its original path.
Returns ``(True, "")`` on success, ``(False, reason)`` on failure.
"""
if name not in self._paths:
return False, f"plugin not loaded: {name}"
path = self._paths[name]
# Core can be reloaded (unload guard is bypassed)
if name == "core":
self._force_unload(name)
else:
self.unload_plugin(name)
count = self.load_plugin(path)
if count < 0:
return False, f"failed to load {name}"
return True, ""
def _force_unload(self, name: str) -> None:
"""Unload a plugin without the core guard (used by reload)."""
self.commands = {
k: v for k, v in self.commands.items() if v.plugin != name
}
for event_type in list(self.events):
self.events[event_type] = [
h for h in self.events[event_type] if h.plugin != name
]
if not self.events[event_type]:
del self.events[event_type]
mod_name = f"derp.plugins.{name}"
sys.modules.pop(mod_name, None)
self._modules.pop(name, None)
self._paths.pop(name, None)