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

feat: initial implementation

Browse Source 
Asyncio IRC bot with decorator-based plugin system.
Zero external dependencies, Python 3.11+.

- IRC protocol: message parsing, formatting, async TCP/TLS connection
- Plugin system: @command and @event decorators, file-based loading
- Bot orchestrator: connect, dispatch, reconnect, nick recovery
- CLI: argparse entry point with TOML config
- Built-in plugins: ping, help, version, echo
- 28 unit tests for parser and plugin system
This commit is contained in:
user
2026-02-15 00:37:31 +01:00
commit bf45abcbad
23 changed files with 1398 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
Makefile Normal file
View File

@@ -0,0 +1,39 @@
.PHONY: install dev test lint clean help
VENV := .venv
PIP := $(VENV)/bin/pip
PYTHON := $(VENV)/bin/python
PYTEST := $(VENV)/bin/pytest
RUFF := $(VENV)/bin/ruff
help: ## Show this help
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | \
awk 'BEGIN {FS = ":.*?## "}; {printf " \033[38;5;110m%-12s\033[0m %s\n", $$1, $$2}'
venv: ## Create virtual environment
python3 -m venv $(VENV)
$(PIP) install --upgrade pip
install: venv ## Install package (editable)
$(PIP) install -e .
$(PIP) install pytest ruff
test: ## Run tests
$(PYTEST) -v
lint: ## Run linter
$(RUFF) check src/ tests/ plugins/
fmt: ## Format code
$(RUFF) format src/ tests/ plugins/
run: ## Run the bot
$(PYTHON) -m derp
clean: ## Remove build artifacts
rm -rf build/ dist/ *.egg-info src/*.egg-info
find . -type d -name __pycache__ -exec rm -rf {} + 2>/dev/null || true
link: install ## Symlink to ~/.local/bin
ln -sf $$(pwd)/$(VENV)/bin/derp ~/.local/bin/derp
@echo "Installed: $$(which derp)"

37
PROJECT.md Normal file
View File

@@ -0,0 +1,37 @@
# derp - Project
## Purpose
A lightweight, zero-dependency asyncio IRC bot with a clean plugin system for Python 3.11+.
## Architecture
```
CLI (argparse) -> Config (TOML) -> Bot (orchestrator)
|-> IRCConnection (async TCP/TLS)
|-> PluginRegistry (decorators, loader)
|-> plugins/*.py
```
### Modules
| Module | Role |
|--------|------|
| `cli.py` | Argument parsing, logging setup, entry point |
| `config.py` | TOML loader with defaults merging |
| `irc.py` | IRC protocol: message parsing, formatting, async connection |
| `plugin.py` | Decorator-based plugin system with file loader |
| `bot.py` | Orchestrator: connect, dispatch, reconnect |
### Key Design Decisions
- **Zero dependencies**: stdlib only (`asyncio`, `ssl`, `tomllib`, `importlib`)
- **Decorator-based plugins**: `@command` and `@event` for clean registration
- **File-based plugin loading**: drop `.py` files in `plugins/` directory
- **Async throughout**: all handlers are `async def`
## Dependencies
- Python 3.11+ (for `tomllib`)
- No external packages required at runtime
- Dev: `pytest`, `ruff`

75
README.md Normal file
View File

@@ -0,0 +1,75 @@
# derp
Asyncio IRC bot for Python 3.11+ with a decorator-based plugin system. Zero external dependencies.
## Quick Start
```bash
git clone <repo> ~/git/derp && cd ~/git/derp
make install
# Edit config/derp.toml with your server details
make run
```
## Features
- Async IRC over plain TCP or TLS
- Plugin system with `@command` and `@event` decorators
- TOML configuration with sensible defaults
- Auto reconnect, nick recovery, PING/PONG handling
- Built-in commands: `!ping`, `!help`, `!version`
## Configuration
Edit `config/derp.toml`:
```toml
[server]
host = "irc.libera.chat"
port = 6697
tls = true
nick = "derp"
[bot]
prefix = "!"
channels = ["#test"]
plugins_dir = "plugins"
```
## Writing Plugins
Create a `.py` file in `plugins/`:
```python
from derp.plugin import command, event
@command("greet", help="Say hello")
async def cmd_greet(bot, message):
await bot.reply(message, f"Hello, {message.nick}!")
@event("JOIN")
async def on_join(bot, message):
if message.nick != bot.nick:
await bot.send(message.target, f"Welcome, {message.nick}")
```
## Commands
| Command | Description |
|---------|-------------|
| `make install` | Create venv and install |
| `make test` | Run test suite |
| `make lint` | Lint with ruff |
| `make run` | Start the bot |
| `make link` | Symlink to `~/.local/bin/` |
## Documentation
- [Installation](docs/INSTALL.md)
- [Usage Guide](docs/USAGE.md)
- [Cheatsheet](docs/CHEATSHEET.md)
- [Debugging](docs/DEBUG.md)
## License
MIT

33
ROADMAP.md Normal file
View File

@@ -0,0 +1,33 @@
# derp - Roadmap
## v0.1.0 (current)
- [x] IRC protocol: connect, parse, send
- [x] TLS support
- [x] Plugin system with `@command` and `@event`
- [x] TOML configuration
- [x] Built-in plugins: ping, help, version
- [x] Auto PING/PONG, nick recovery, reconnect
- [x] CLI entry point
## v0.2.0
- [ ] Plugin hot-reload (`!reload` command)
- [ ] Per-channel plugin enable/disable
- [ ] SASL authentication
- [ ] Rate limiting (anti-flood)
- [ ] CTCP VERSION/TIME/PING responses
## v0.3.0
- [ ] Admin system (owner/admin nicks in config)
- [ ] Permissions per command
- [ ] Channel management commands (kick, ban, topic)
- [ ] Plugin state persistence (sqlite)
## v1.0.0
- [ ] Multi-server support
- [ ] IRCv3 capability negotiation
- [ ] Message tags support
- [ ] Stable plugin API

15
TASKS.md Normal file
View File

@@ -0,0 +1,15 @@
# derp - Tasks
## Current (2026-02-15)
| Pri | Status | Task |
|-----|--------|------|
| P0 | [x] | Core IRC protocol implementation |
| P0 | [x] | Plugin system with decorators |
| P0 | [x] | Bot orchestrator with reconnect |
| P0 | [x] | CLI entry point |
| P0 | [x] | Built-in plugins (core, example) |
| P0 | [x] | Unit tests for parser and plugins |
| P0 | [x] | Documentation |
| P1 | [ ] | Test against live IRC server |
| P2 | [ ] | SASL authentication |

23
TODO.md Normal file
View File

@@ -0,0 +1,23 @@
# derp - Backlog
## Features
- [ ] Plugin hot-reload command
- [ ] SASL PLAIN authentication
- [ ] Admin/owner permission system
- [ ] Rate limiting for outgoing messages
- [ ] CTCP responses (VERSION, TIME, PING)
- [ ] Multi-server support
## Improvements
- [ ] Structured logging (JSON option)
- [ ] Plugin state persistence
- [ ] Channel-specific plugin config
- [ ] Configurable reconnect backoff
## Testing
- [ ] Integration tests with mock IRC server
- [ ] Bot orchestrator tests
- [ ] Config merge edge case tests

16
config/derp.toml Normal file
View File

@@ -0,0 +1,16 @@
[server]
host = "irc.libera.chat"
port = 6697
tls = true
nick = "derp"
user = "derp"
realname = "derp IRC bot"
password = ""
[bot]
prefix = "!"
channels = ["#test"]
plugins_dir = "plugins"
[logging]
level = "info"

59
docs/CHEATSHEET.md Normal file
View File

@@ -0,0 +1,59 @@
# Cheatsheet
## Quick Commands
```bash
make install # Setup venv + install
make test # Run tests
make lint # Lint with ruff
make run # Start bot
make link # Symlink to ~/.local/bin
derp -c config.toml # Run with custom config
derp -v # Verbose/debug mode
```
## Bot Commands
```
!ping # Pong
!help # List commands
!help <cmd> # Command help
!version # Bot version
!echo <text> # Echo text back
```
## Plugin Template
```python
from derp.plugin import command, event
@command("name", help="Description")
async def cmd_name(bot, message):
text = message.text.split(None, 1)
await bot.reply(message, "response")
@event("JOIN")
async def on_join(bot, message):
await bot.send(message.target, f"Hi {message.nick}")
```
## Message Object
```
msg.nick # Sender nick
msg.target # Channel or nick
msg.text # Message body
msg.is_channel # True if channel
msg.prefix # nick!user@host
msg.command # PRIVMSG, JOIN, etc.
msg.params # All params list
```
## Config Locations
```
1. --config PATH # CLI flag
2. ./config/derp.toml # Project dir
3. ~/.config/derp/derp.toml # User config
4. Built-in defaults # Fallback
```

74
docs/DEBUG.md Normal file
View File

@@ -0,0 +1,74 @@
# Debugging
## Verbose Mode
```bash
derp --verbose
```
Shows all IRC traffic:
```
>>> NICK derp
>>> USER derp 0 * :derp IRC bot
<<< :server 001 derp :Welcome
>>> JOIN #test
```
## Log Levels
Set in `config/derp.toml`:
```toml
[logging]
level = "debug" # debug, info, warning, error
```
## Common Issues
### Connection refused
```
ERROR derp.irc connection lost: [Errno 111] Connection refused
```
- Check `host` and `port` in config
- Verify TLS setting matches port (6697 = TLS, 6667 = plain)
- Test connectivity: `nc -zv <host> <port>`
### Nickname in use
The bot appends `_` to the nick and retries automatically. Check logs for:
```
<<< :server 433 * derp :Nickname is already in use
>>> NICK derp_
```
### TLS certificate errors
If the server uses a self-signed certificate, you may need to adjust the SSL context. Currently uses system default CA bundle.
### Plugin load failures
```
ERROR derp.plugin failed to load plugin: plugins/broken.py
```
- Check plugin file for syntax errors: `python -c "import plugins.broken"`
- Ensure handlers are `async def`
- Check imports (`from derp.plugin import command, event`)
### No response to commands
- Verify `prefix` in config matches what you type
- Check that the plugin is loaded (look for "loaded plugin" in verbose output)
- Ensure the bot has joined the channel
## Testing IRC Connection
```bash
# Test raw IRC (without the bot)
openssl s_client -connect irc.libera.chat:6697
# Then type: NICK testbot / USER testbot 0 * :test
```

52
docs/INSTALL.md Normal file
View File

@@ -0,0 +1,52 @@
# Installation
## Prerequisites
- Python 3.11+
- git
## Setup
```bash
cd ~/git/derp
make install
```
This creates a `.venv`, installs derp in editable mode, and adds dev tools.
## Symlink
```bash
make link
```
Installs `derp` to `~/.local/bin/`. Verify:
```bash
which derp
derp --version
```
## Manual Install
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -e .
```
## Configuration
Copy and edit the default config:
```bash
cp config/derp.toml ~/.config/derp/derp.toml
# Edit server, nick, channels
```
Config search order:
1. Path given via `--config`
2. `./config/derp.toml`
3. `~/.config/derp/derp.toml`
4. Built-in defaults

97
docs/USAGE.md Normal file
View File

@@ -0,0 +1,97 @@
# Usage Guide
## Running
```bash
# From project directory
derp
# With options
derp --config /path/to/derp.toml --verbose
```
## CLI Flags
| Flag | Description |
|------|-------------|
| `-c, --config PATH` | Config file path |
| `-v, --verbose` | Debug logging |
| `-V, --version` | Print version |
| `-h, --help` | Show help |
## Configuration
All settings in `config/derp.toml`:
```toml
[server]
host = "irc.libera.chat" # IRC server hostname
port = 6697 # Port (6697 = TLS, 6667 = plain)
tls = true # Enable TLS encryption
nick = "derp" # Bot nickname
user = "derp" # Username (ident)
realname = "derp IRC bot" # Real name field
password = "" # Server password (optional)
[bot]
prefix = "!" # Command prefix character
channels = ["#test"] # Channels to join on connect
plugins_dir = "plugins" # Plugin directory path
[logging]
level = "info" # Logging level: debug, info, warning, error
```
## Built-in Commands
| Command | Description |
|---------|-------------|
| `!ping` | Bot responds with "pong" |
| `!help` | List all available commands |
| `!help <cmd>` | Show help for a specific command |
| `!version` | Show bot version |
| `!echo <text>` | Echo back text (example plugin) |
## Writing Plugins
Create a `.py` file in the `plugins/` directory:
```python
from derp.plugin import command, event
@command("hello", help="Greet the user")
async def cmd_hello(bot, message):
"""Handler receives bot instance and parsed Message."""
await bot.reply(message, f"Hello, {message.nick}!")
@event("JOIN")
async def on_join(bot, message):
"""Event handlers fire on IRC events (JOIN, PART, QUIT, etc.)."""
if message.nick != bot.nick:
await bot.send(message.target, f"Welcome, {message.nick}")
```
### Plugin API
The `bot` object provides:
| Method | Description |
|--------|-------------|
| `bot.send(target, text)` | Send message to channel or nick |
| `bot.reply(msg, text)` | Reply to source (channel or PM) |
| `bot.action(target, text)` | Send `/me` action |
| `bot.join(channel)` | Join a channel |
| `bot.part(channel [, reason])` | Leave a channel |
| `bot.quit([reason])` | Disconnect from server |
The `message` object provides:
| Attribute | Description |
|-----------|-------------|
| `message.nick` | Sender's nickname |
| `message.prefix` | Full `nick!user@host` prefix |
| `message.command` | IRC command (PRIVMSG, JOIN, etc.) |
| `message.target` | First param (channel or nick) |
| `message.text` | Trailing text content |
| `message.is_channel` | Whether target is a channel |
| `message.params` | All message parameters |

39
plugins/core.py Normal file
View File

@@ -0,0 +1,39 @@
"""Core plugin: ping, help, version."""
from derp import __version__
from derp.plugin import command
@command("ping", help="Check if the bot is alive")
async def cmd_ping(bot, message):
"""Respond with pong."""
await bot.reply(message, "pong")
@command("help", help="List commands or show command help")
async def cmd_help(bot, message):
"""Show available commands or help for a specific command.
Usage: !help [command]
"""
parts = message.text.split(None, 2)
if len(parts) > 1:
# Help for a specific command
cmd_name = parts[1].lower().lstrip(bot.prefix)
handler = bot.registry.commands.get(cmd_name)
if handler:
help_text = handler.help or "No help available."
await bot.reply(message, f"{bot.prefix}{cmd_name} -- {help_text}")
else:
await bot.reply(message, f"Unknown command: {cmd_name}")
return
# List all commands
names = sorted(bot.registry.commands.keys())
await bot.reply(message, f"Commands: {', '.join(bot.prefix + n for n in names)}")
@command("version", help="Show bot version")
async def cmd_version(bot, message):
"""Report the running version."""
await bot.reply(message, f"derp {__version__}")

23
plugins/example.py Normal file
View File

@@ -0,0 +1,23 @@
"""Example plugin demonstrating the derp plugin API."""
from derp.plugin import command, event
@command("echo", help="Echo back text")
async def cmd_echo(bot, message):
"""Repeat everything after the command.
Usage: !echo <text>
"""
parts = message.text.split(None, 1)
if len(parts) > 1:
await bot.reply(message, parts[1])
else:
await bot.reply(message, "Usage: !echo <text>")
@event("JOIN")
async def on_join(bot, message):
"""Greet users joining a channel (skip self)."""
if message.nick and message.nick != bot.nick:
await bot.send(message.target, f"Hey {message.nick}")

26
pyproject.toml Normal file
View File

@@ -0,0 +1,26 @@
[build-system]
requires = ["setuptools>=68.0"]
build-backend = "setuptools.build_meta"
[project]
name = "derp"
version = "0.1.0"
description = "Asyncio IRC bot with plugin system"
requires-python = ">=3.11"
license = "MIT"
[project.scripts]
derp = "derp.cli:main"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.ruff]
line-length = 99
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "W", "I"]

3
src/derp/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""derp - asyncio IRC bot with plugin system."""
__version__ = "0.1.0"

5
src/derp/__main__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Allow running as `python -m derp`."""
from derp.cli import main
raise SystemExit(main())

159
src/derp/bot.py Normal file
View File

@@ -0,0 +1,159 @@
"""Bot orchestrator: connect, run loop, dispatch events."""
from __future__ import annotations
import asyncio
import logging
from pathlib import Path
from derp.irc import IRCConnection, Message, format_msg, parse
from derp.plugin import PluginRegistry
log = logging.getLogger(__name__)
RECONNECT_DELAY = 30
class Bot:
"""IRC bot: ties connection, config, and plugins together."""
def __init__(self, config: dict, registry: PluginRegistry) -> None:
self.config = config
self.registry = registry
self.conn = IRCConnection(
host=config["server"]["host"],
port=config["server"]["port"],
tls=config["server"]["tls"],
)
self.nick: str = config["server"]["nick"]
self.prefix: str = config["bot"]["prefix"]
self._running = False
async def start(self) -> None:
"""Connect, register, join channels, and enter the main loop."""
self._running = True
while self._running:
try:
await self._connect_and_run()
except (OSError, ConnectionError) as exc:
log.error("connection lost: %s", exc)
if self._running:
log.info("reconnecting in %ds...", RECONNECT_DELAY)
await asyncio.sleep(RECONNECT_DELAY)
async def _connect_and_run(self) -> None:
"""Single connection lifecycle."""
await self.conn.connect()
try:
await self._register()
await self._loop()
finally:
await self.conn.close()
async def _register(self) -> None:
"""Send NICK/USER registration to the server."""
srv = self.config["server"]
if srv.get("password"):
await self.conn.send(format_msg("PASS", srv["password"]))
await self.conn.send(format_msg("NICK", self.nick))
await self.conn.send(format_msg("USER", srv["user"], "0", "*", srv["realname"]))
async def _loop(self) -> None:
"""Read and dispatch messages until disconnect."""
while self._running:
line = await self.conn.readline()
if line is None:
log.warning("server closed connection")
return
msg = parse(line)
await self._handle(msg)
async def _handle(self, msg: Message) -> None:
"""Dispatch a parsed message to the appropriate handlers."""
# Protocol-level PING/PONG
if msg.command == "PING":
await self.conn.send(format_msg("PONG", msg.params[0] if msg.params else ""))
return
# RPL_WELCOME (001) — join channels
if msg.command == "001":
self.nick = msg.params[0] if msg.params else self.nick
for channel in self.config["bot"]["channels"]:
await self.join(channel)
# Nick already in use (433) — append underscore
if msg.command == "433":
self.nick = self.nick + "_"
await self.conn.send(format_msg("NICK", self.nick))
return
# Dispatch to event handlers
event_type = msg.command
for handler in self.registry.events.get(event_type, []):
try:
await handler.callback(self, msg)
except Exception:
log.exception("error in event handler %s", handler.name)
# Dispatch to command handlers (PRIVMSG only)
if msg.command == "PRIVMSG" and msg.text:
await self._dispatch_command(msg)
async def _dispatch_command(self, msg: Message) -> None:
"""Check if a PRIVMSG is a bot command and dispatch it."""
text = msg.text
if not text or not text.startswith(self.prefix):
return
parts = text[len(self.prefix):].split(None, 1)
cmd_name = parts[0].lower() if parts else ""
handler = self.registry.commands.get(cmd_name)
if handler is None:
return
try:
await handler.callback(self, msg)
except Exception:
log.exception("error in command handler '%s'", cmd_name)
# -- Public API for plugins --
async def send(self, target: str, text: str) -> None:
"""Send a PRIVMSG to a target (channel or nick)."""
for line in text.split("\n"):
await self.conn.send(format_msg("PRIVMSG", target, line))
async def reply(self, msg: Message, text: str) -> None:
"""Reply to the source of a message (channel or PM)."""
target = msg.target if msg.is_channel else msg.nick
if target:
await self.send(target, text)
async def action(self, target: str, text: str) -> None:
"""Send a CTCP ACTION (/me) to a target."""
await self.send(target, f"\x01ACTION {text}\x01")
async def join(self, channel: str) -> None:
"""Join an IRC channel."""
await self.conn.send(format_msg("JOIN", channel))
log.info("joining %s", channel)
async def part(self, channel: str, reason: str = "") -> None:
"""Part an IRC channel."""
if reason:
await self.conn.send(format_msg("PART", channel, reason))
else:
await self.conn.send(format_msg("PART", channel))
async def quit(self, reason: str = "bye") -> None:
"""Quit the IRC server and stop the bot."""
self._running = False
await self.conn.send(format_msg("QUIT", reason))
await self.conn.close()
def load_plugins(self, plugins_dir: str | Path | None = None) -> None:
"""Load plugins from the configured directory."""
if plugins_dir is None:
plugins_dir = self.config["bot"].get("plugins_dir", "plugins")
path = Path(plugins_dir)
self.registry.load_directory(path)

69
src/derp/cli.py Normal file
View File

@@ -0,0 +1,69 @@
"""CLI entry point for derp."""
from __future__ import annotations
import argparse
import asyncio
import logging
import sys
from derp import __version__
from derp.bot import Bot
from derp.config import resolve_config
from derp.plugin import PluginRegistry
LOG_FORMAT = "%(asctime)s %(levelname)-5s %(name)s %(message)s"
LOG_DATE = "%H:%M:%S"
def build_parser() -> argparse.ArgumentParser:
"""Build the argument parser."""
p = argparse.ArgumentParser(
prog="derp",
description="Asyncio IRC bot with plugin system",
)
p.add_argument(
"-c", "--config",
metavar="PATH",
help="path to config file [config/derp.toml]",
)
p.add_argument(
"-v", "--verbose",
action="store_true",
help="enable debug logging",
)
p.add_argument(
"-V", "--version",
action="version",
version=f"derp {__version__}",
)
return p
def main(argv: list[str] | None = None) -> int:
"""Main entry point."""
parser = build_parser()
args = parser.parse_args(argv)
level = logging.DEBUG if args.verbose else logging.INFO
logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE, level=level)
config = resolve_config(args.config)
log = logging.getLogger("derp")
log.info("derp %s starting", __version__)
registry = PluginRegistry()
bot = Bot(config, registry)
bot.load_plugins()
try:
asyncio.run(bot.start())
except KeyboardInterrupt:
log.info("interrupted, shutting down")
return 0
if __name__ == "__main__":
sys.exit(main())

57
src/derp/config.py Normal file
View File

@@ -0,0 +1,57 @@
"""TOML configuration loader."""
from __future__ import annotations
import tomllib
from pathlib import Path
DEFAULTS: dict = {
"server": {
"host": "irc.libera.chat",
"port": 6697,
"tls": True,
"nick": "derp",
"user": "derp",
"realname": "derp IRC bot",
"password": "",
},
"bot": {
"prefix": "!",
"channels": ["#test"],
"plugins_dir": "plugins",
},
"logging": {
"level": "info",
},
}
def _merge(base: dict, override: dict) -> dict:
"""Deep-merge override into base, returning a new dict."""
merged = base.copy()
for key, val in override.items():
if key in merged and isinstance(merged[key], dict) and isinstance(val, dict):
merged[key] = _merge(merged[key], val)
else:
merged[key] = val
return merged
def load(path: Path) -> dict:
"""Load TOML config from path, merged with defaults."""
with open(path, "rb") as f:
user_config = tomllib.load(f)
return _merge(DEFAULTS, user_config)
def resolve_config(path: str | None) -> dict:
"""Resolve config path and load. Falls back to defaults if no file found."""
search_paths = [
Path(path) if path else None,
Path("config/derp.toml"),
Path.home() / ".config" / "derp" / "derp.toml",
]
for p in search_paths:
if p and p.is_file():
return load(p)
return DEFAULTS.copy()

144
src/derp/irc.py Normal file
View File

@@ -0,0 +1,144 @@
"""IRC protocol: message parsing, formatting, and async connection."""
from __future__ import annotations
import asyncio
import logging
import ssl
from dataclasses import dataclass
log = logging.getLogger(__name__)
@dataclass(slots=True)
class Message:
"""Parsed IRC message (RFC 1459)."""
raw: str
prefix: str | None
nick: str | None
command: str
params: list[str]
@property
def target(self) -> str | None:
"""First param — typically channel or nick."""
return self.params[0] if self.params else None
@property
def text(self) -> str | None:
"""Trailing text (last param if prefixed with ':' in raw)."""
return self.params[-1] if self.params else None
@property
def is_channel(self) -> bool:
"""Whether the target is a channel."""
return bool(self.target and self.target.startswith(("#", "&")))
def parse(line: str) -> Message:
"""Parse a raw IRC line into a Message.
Format: [:prefix] command [params...] [:trailing]
"""
raw = line
prefix = None
nick = None
if line.startswith(":"):
prefix, line = line[1:].split(" ", 1)
if "!" in prefix:
nick = prefix.split("!")[0]
else:
nick = None
trailing = None
if " :" in line:
line, trailing = line.split(" :", 1)
parts = line.split()
command = parts[0].upper() if parts else ""
params = parts[1:]
if trailing is not None:
params.append(trailing)
return Message(raw=raw, prefix=prefix, nick=nick, command=command, params=params)
def format_msg(command: str, *params: str) -> str:
"""Format an IRC command into a raw line.
The last param is automatically prefixed with ':' if it contains spaces
or is the trailing argument.
"""
if not params:
return command
*head, tail = params
if " " in tail or tail.startswith(":") or not head:
prefix = f"{command} {' '.join(head)}" if head else command
return f"{prefix} :{tail}"
return f"{command} {' '.join(params)}"
class IRCConnection:
"""Async TCP/TLS connection to an IRC server."""
def __init__(self, host: str, port: int, tls: bool = True) -> None:
self.host = host
self.port = port
self.tls = tls
self._reader: asyncio.StreamReader | None = None
self._writer: asyncio.StreamWriter | None = None
async def connect(self) -> None:
"""Open connection to the IRC server."""
ssl_ctx = None
if self.tls:
ssl_ctx = ssl.create_default_context()
log.info("connecting to %s:%d (tls=%s)", self.host, self.port, self.tls)
self._reader, self._writer = await asyncio.open_connection(
self.host, self.port, ssl=ssl_ctx
)
log.info("connected")
async def send(self, line: str) -> None:
"""Send a raw IRC line (appends CRLF)."""
if self._writer is None:
raise RuntimeError("not connected")
data = f"{line}\r\n".encode("utf-8")
self._writer.write(data)
await self._writer.drain()
log.debug(">>> %s", line)
async def readline(self) -> str | None:
"""Read one IRC line. Returns None on EOF."""
if self._reader is None:
raise RuntimeError("not connected")
try:
data = await self._reader.readline()
except (ConnectionResetError, asyncio.IncompleteReadError):
return None
if not data:
return None
line = data.decode("utf-8", errors="replace").strip()
log.debug("<<< %s", line)
return line
async def close(self) -> None:
"""Close the connection."""
if self._writer:
self._writer.close()
try:
await self._writer.wait_closed()
except Exception:
pass
self._writer = None
self._reader = None
log.info("connection closed")
@property
def connected(self) -> bool:
"""Whether the connection is open."""
return self._writer is not None and not self._writer.is_closing()

125
src/derp/plugin.py Normal file
View File

@@ -0,0 +1,125 @@
"""Plugin system: decorators, registry, and loader."""
from __future__ import annotations
import importlib.util
import inspect
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
log = logging.getLogger(__name__)
@dataclass(slots=True)
class Handler:
"""A registered command or event handler."""
name: str
callback: Callable
help: str = ""
plugin: str = ""
def command(name: str, help: str = "") -> Callable:
"""Decorator to register an async function as a bot command.
Usage::
@command("ping", help="Check if the bot is alive")
async def cmd_ping(bot, message):
await bot.reply(message, "pong")
"""
def decorator(func: Callable) -> Callable:
func._derp_command = name # type: ignore[attr-defined]
func._derp_help = help # type: ignore[attr-defined]
return func
return decorator
def event(event_type: str) -> Callable:
"""Decorator to register an async function as an event handler.
Usage::
@event("join")
async def on_join(bot, message):
await bot.send(message.target, f"Welcome, {message.nick}!")
"""
def decorator(func: Callable) -> Callable:
func._derp_event = event_type.upper() # type: ignore[attr-defined]
return func
return decorator
class PluginRegistry:
"""Collects and manages command/event handlers from plugin modules."""
def __init__(self) -> None:
self.commands: dict[str, Handler] = {}
self.events: dict[str, list[Handler]] = {}
self._modules: dict[str, Any] = {}
def register_command(self, name: str, callback: Callable, help: str = "",
plugin: str = "") -> None:
"""Register a command handler."""
if name in self.commands:
log.warning("command '%s' already registered, overwriting", name)
self.commands[name] = Handler(name=name, callback=callback, help=help, plugin=plugin)
log.debug("registered command: %s (%s)", name, plugin)
def register_event(self, event_type: str, callback: Callable, plugin: str = "") -> None:
"""Register an event handler."""
event_type = event_type.upper()
if event_type not in self.events:
self.events[event_type] = []
handler = Handler(name=event_type, callback=callback, plugin=plugin)
self.events[event_type].append(handler)
log.debug("registered event: %s (%s)", event_type, plugin)
def _scan_module(self, module: Any, plugin_name: str) -> int:
"""Scan a module for decorated handlers. Returns count of handlers found."""
count = 0
for _name, obj in inspect.getmembers(module, inspect.isfunction):
if hasattr(obj, "_derp_command"):
self.register_command(
obj._derp_command, obj,
help=getattr(obj, "_derp_help", ""),
plugin=plugin_name,
)
count += 1
if hasattr(obj, "_derp_event"):
self.register_event(obj._derp_event, obj, plugin=plugin_name)
count += 1
return count
def load_plugin(self, path: Path) -> None:
"""Load a single plugin from a .py file."""
plugin_name = path.stem
if plugin_name.startswith("_"):
return
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)
count = self._scan_module(module, plugin_name)
self._modules[plugin_name] = module
log.info("loaded plugin: %s (%d handlers)", plugin_name, count)
except Exception:
log.exception("failed to load plugin: %s", path)
def load_directory(self, dir_path: Path) -> None:
"""Load all .py plugin files from a directory."""
if not dir_path.is_dir():
log.warning("plugins directory not found: %s", dir_path)
return
for path in sorted(dir_path.glob("*.py")):
self.load_plugin(path)

100
tests/test_irc.py Normal file
View File

@@ -0,0 +1,100 @@
"""Tests for IRC message parsing and formatting."""
from derp.irc import format_msg, parse
class TestParse:
"""IRC message parser tests."""
def test_privmsg(self):
msg = parse(":nick!user@host PRIVMSG #channel :hello world")
assert msg.prefix == "nick!user@host"
assert msg.nick == "nick"
assert msg.command == "PRIVMSG"
assert msg.params == ["#channel", "hello world"]
assert msg.target == "#channel"
assert msg.text == "hello world"
assert msg.is_channel is True
def test_privmsg_pm(self):
msg = parse(":nick!user@host PRIVMSG bot :hello")
assert msg.target == "bot"
assert msg.is_channel is False
def test_ping(self):
msg = parse("PING :server.example.com")
assert msg.prefix is None
assert msg.nick is None
assert msg.command == "PING"
assert msg.params == ["server.example.com"]
def test_join(self):
msg = parse(":nick!user@host JOIN #channel")
assert msg.command == "JOIN"
assert msg.nick == "nick"
assert msg.target == "#channel"
def test_numeric(self):
msg = parse(":server 001 bot :Welcome to the IRC Network")
assert msg.command == "001"
assert msg.prefix == "server"
assert msg.nick is None
assert msg.params == ["bot", "Welcome to the IRC Network"]
def test_nick_in_use(self):
msg = parse(":server 433 * derp :Nickname is already in use")
assert msg.command == "433"
assert msg.params[1] == "derp"
def test_empty_trailing(self):
msg = parse(":nick!user@host PRIVMSG #channel :")
assert msg.text == ""
def test_no_prefix(self):
msg = parse("NOTICE AUTH :*** Looking up your hostname")
assert msg.prefix is None
assert msg.command == "NOTICE"
assert msg.params == ["AUTH", "*** Looking up your hostname"]
def test_server_prefix_no_nick(self):
msg = parse(":irc.server.net 372 bot :- Message of the day")
assert msg.prefix == "irc.server.net"
assert msg.nick is None
def test_command_case(self):
msg = parse(":nick!u@h privmsg #ch :test")
assert msg.command == "PRIVMSG"
def test_multiple_colons_in_trailing(self):
msg = parse(":nick!u@h PRIVMSG #ch :url: https://example.com")
assert msg.text == "url: https://example.com"
def test_part_with_reason(self):
msg = parse(":nick!u@h PART #channel :leaving")
assert msg.command == "PART"
assert msg.target == "#channel"
assert msg.text == "leaving"
class TestFormat:
"""IRC message formatting tests."""
def test_simple_command(self):
assert format_msg("NICK", "derp") == "NICK :derp"
def test_command_with_trailing(self):
assert format_msg("PRIVMSG", "#channel", "hello world") == \
"PRIVMSG #channel :hello world"
def test_join(self):
assert format_msg("JOIN", "#channel") == "JOIN :#channel"
def test_no_params(self):
assert format_msg("QUIT") == "QUIT"
def test_user_registration(self):
result = format_msg("USER", "derp", "0", "*", "derp IRC bot")
assert result == "USER derp 0 * :derp IRC bot"
def test_pong(self):
assert format_msg("PONG", "server.example.com") == "PONG :server.example.com"

128
tests/test_plugin.py Normal file
View File

@@ -0,0 +1,128 @@
"""Tests for the plugin system."""
import textwrap
from pathlib import Path
from derp.plugin import PluginRegistry, command, event
class TestDecorators:
"""Test that decorators mark functions correctly."""
def test_command_decorator(self):
@command("test", help="a test command")
async def handler(bot, msg):
pass
assert handler._derp_command == "test"
assert handler._derp_help == "a test command"
def test_event_decorator(self):
@event("join")
async def handler(bot, msg):
pass
assert handler._derp_event == "JOIN"
def test_event_decorator_case(self):
@event("PRIVMSG")
async def handler(bot, msg):
pass
assert handler._derp_event == "PRIVMSG"
class TestRegistry:
"""Test the plugin registry."""
def test_register_command(self):
registry = PluginRegistry()
async def handler(bot, msg):
pass
registry.register_command("test", handler, help="test help")
assert "test" in registry.commands
assert registry.commands["test"].help == "test help"
def test_register_event(self):
registry = PluginRegistry()
async def handler(bot, msg):
pass
registry.register_event("PRIVMSG", handler)
assert "PRIVMSG" in registry.events
assert len(registry.events["PRIVMSG"]) == 1
def test_multiple_event_handlers(self):
registry = PluginRegistry()
async def handler_a(bot, msg):
pass
async def handler_b(bot, msg):
pass
registry.register_event("JOIN", handler_a)
registry.register_event("JOIN", handler_b)
assert len(registry.events["JOIN"]) == 2
def test_load_plugin_file(self, tmp_path: Path):
plugin_code = textwrap.dedent("""\
from derp.plugin import command, event
@command("greet", help="Say hello")
async def cmd_greet(bot, msg):
pass
@event("JOIN")
async def on_join(bot, msg):
pass
""")
plugin_file = tmp_path / "greet.py"
plugin_file.write_text(plugin_code)
registry = PluginRegistry()
registry.load_plugin(plugin_file)
assert "greet" in registry.commands
assert "JOIN" in registry.events
def test_load_directory(self, tmp_path: Path):
for name in ("a.py", "b.py"):
(tmp_path / name).write_text(textwrap.dedent(f"""\
from derp.plugin import command
@command("{name[0]}", help="{name}")
async def cmd(bot, msg):
pass
"""))
registry = PluginRegistry()
registry.load_directory(tmp_path)
assert "a" in registry.commands
assert "b" in registry.commands
def test_skip_underscore_files(self, tmp_path: Path):
(tmp_path / "__init__.py").write_text("")
(tmp_path / "_private.py").write_text("x = 1\n")
(tmp_path / "valid.py").write_text(textwrap.dedent("""\
from derp.plugin import command
@command("valid")
async def cmd(bot, msg):
pass
"""))
registry = PluginRegistry()
registry.load_directory(tmp_path)
assert "valid" in registry.commands
assert len(registry.commands) == 1
def test_load_missing_directory(self, tmp_path: Path):
registry = PluginRegistry()
registry.load_directory(tmp_path / "nonexistent")
assert len(registry.commands) == 0