Compare commits

..

6 Commits

Author SHA1 Message Date
user 6083de13f9 feat: playlist shuffle, lazy resolution, TTS ducking, kept repair
CI / gitleaks (push) Failing after 3s
CI / lint (push) Successful in 22s
CI / test (3.11) (push) Failing after 2m47s
CI / test (3.13) (push) Failing after 2m52s
CI / test (3.12) (push) Failing after 2m54s
CI / build (push) Has been skipped
Music:
- #random URL fragment shuffles playlist tracks before enqueuing
- Lazy playlist resolution: first 10 tracks resolve immediately,
  remaining are fetched in a background task
- !kept repair re-downloads kept tracks with missing local files
- !kept shows [MISSING] marker for tracks without local files
- TTS ducking: music ducks when merlin speaks via voice peer,
  smooth restore after TTS finishes

Performance (from profiling):
- Connection pool: preload_content=True for SOCKS connection reuse
- Pool tuning: 30 pools / 8 connections (up from 20/4)
- _PooledResponse wrapper for stdlib-compatible read interface
- Iterative _extract_videos (replace 51K-deep recursion with stack)
- proxy=False for local SearXNG

Voice + multi-bot:
- Per-bot voice config lookup ([<username>.voice] in TOML)
- Mute detection: skip duck silence when all users muted
- Autoplay shuffle deck (no repeats until full cycle)
- Seek clamp to track duration (prevent seek-past-end stall)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-22 16:21:47 +01:00
user 6d6b957557 docs: update roadmap, tasks, and backlog for v2.3.0
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-22 12:10:28 +01:00
user f72f55148b fix: ignore bot audio in sound callback, self-mute support
- _on_sound_received skips audio from our own bots entirely,
  preventing self-ducking and STT transcribing bot TTS/music
- New self_mute config: mute on connect, unmute before stream_audio,
  re-mute 3s after audio finishes

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-22 12:09:30 +01:00
user e9d17e8b00 feat: voice profiles, rubberband FX, per-bot plugin filtering
- Add rubberband package to container for pitch-shifting FX
- Split FX chain: rubberband CLI for pitch, ffmpeg for filters
- Configurable voice profile (voice, fx, piper params) in [voice]
- Extra bots inherit voice config (minus trigger) for own TTS
- Greeting is voice-only, spoken directly by the greeting bot
- Per-bot only_plugins/except_plugins filtering on Mumble
- Alias plugin, core plugin tests

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-22 11:41:00 +01:00
user 3afeace6e7 feat: container management tools in tools/
Shell scripts for build, start, stop, restart, nuke, logs, status.
Shared helpers in _common.sh (colours, compose detection, project root).
Updated CHEATSHEET.md with new tool references.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-22 11:40:53 +01:00
user b88a459142 feat: music library management, smooth fades, clickable URLs
- Audio-only downloads (-x), resume (-c), skip existing (--no-overwrites)
- Title-based filenames (e.g. never-gonna-give-you-up.opus)
- Separate cache (data/music/cache/) from kept tracks (data/music/)
- Kept track IDs: !keep assigns #id, !play #id, !kept shows IDs
- Linear fade-in (5s) and fade-out (3s) with volume-proportional step
- Fix ramp click: threshold-based convergence instead of float equality
- Clean up cache files for skipped/stopped tracks
- Auto-linkify URLs in Mumble text chat (clickable <a> tags)
- FlaskPaste links use /raw endpoint for direct content access
- Metadata fetch uses --no-playlist for reliable results

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-22 08:11:29 +01:00
34 changed files with 3558 additions and 269 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
FROM python:3.13-alpine
RUN apk add --no-cache opus ffmpeg yt-dlp && \
RUN apk add --no-cache opus ffmpeg yt-dlp rubberband && \
ln -s /usr/lib/libopus.so.0 /usr/lib/libopus.so
WORKDIR /app
+41
View File
@@ -159,3 +159,44 @@
- [ ] Slack adapter via Socket Mode WebSocket
- [ ] Mattermost adapter via WebSocket API
- [ ] Bluesky adapter via AT Protocol firehose + REST API
## v2.3.0 -- Mumble Voice + Multi-Bot (done)
- [x] pymumble transport rewrite (voice + text)
- [x] Music playback: play/stop/skip/prev/queue/np/volume/seek/resume
- [x] Voice ducking (auto-lower music on voice activity)
- [x] Kept track library with metadata (!keep, !kept, !play #N)
- [x] Smooth fade-out on skip/stop/prev, fade-in on resume
- [x] In-stream seek with pipeline swap (no task cancellation)
- [x] Multi-bot Mumble: extra bots via `[[mumble.extra]]`
- [x] Per-bot plugin filtering (only_plugins / except_plugins)
- [x] Voice STT (Whisper) + TTS (Piper) plugin
- [x] Configurable voice profiles (voice, FX, piper params)
- [x] Rubberband pitch-shifting via CLI (Alpine ffmpeg lacks librubberband)
- [x] Bot audio ignored in sound callback (no self-ducking, no STT of bots)
- [x] Self-mute support (mute on connect, unmute for audio, re-mute after)
- [x] Autoplay shuffled kept tracks on reconnect (silence detection)
- [x] Alias plugin (!alias add/del/list)
- [x] Container management tools (tools/build, start, stop, restart, nuke, logs, status)
## v2.4.0 -- Music Discovery + Performance
- [ ] Last.fm integration (artist.getSimilar, artist.getTopTags, track.getSimilar)
- [ ] `!similar` command (find similar artists, optionally queue via YouTube)
- [ ] `!tags` command (genre/style tags for current track)
- [x] Pause/unpause (`!pause` toggle, position tracking, stale re-download)
- [x] Autoplay continuous radio (random kept, silence-aware, cooldown between tracks)
- [x] Periodic resume persistence (10s interval, survives hard kills)
- [x] Track duration in `!np` (elapsed/total via ffprobe)
- [x] `!announce` toggle (optional track announcements)
- [x] Direct bot addressing (`merlin: say <text>`, TTS via voice peer)
- [x] Self-deafen on connect
- [x] Fade-out click fix (conditional buffer clear, post-fade drain)
- [x] cProfile analysis tool (`tools/profile`)
- [x] Mute detection: skip duck silence when all users muted
- [x] Autoplay shuffle deck (no repeats until full cycle)
- [x] Seek clamp to track duration (prevent seek-past-end stall)
- [x] Iterative `_extract_videos` (replace 51K-deep recursion with stack)
- [x] Bypass SOCKS5 for local SearXNG (`proxy=False`)
- [x] Connection pool: `preload_content=True` for SOCKS connection reuse
- [x] Pool tuning: 30 pools / 8 connections (up from 20/4)
+42 -1
View File
@@ -1,6 +1,45 @@
# derp - Tasks
## Current Sprint -- v2.3.0 Mumble Music Playback (2026-02-21)
## Current Sprint -- Performance: HTTP + Parsing (2026-02-22)
| Pri | Status | Task |
|-----|--------|------|
| P0 | [x] | Rewrite `_extract_videos` as iterative stack-based (51K recursive calls from 4 invocations) |
| P0 | [x] | `plugins/searx.py` -- route through `derp.http.urlopen(proxy=False)` |
| P1 | [x] | Connection pool: `preload_content=True` + `_PooledResponse` wrapper for connection reuse |
| P1 | [x] | Pool tuning: `num_pools=30, maxsize=8` (was 20/4) |
| P2 | [ ] | Audit remaining plugins for unnecessary proxy routing |
## Previous Sprint -- Music Discovery via Last.fm (2026-02-22)
| Pri | Status | Task |
|-----|--------|------|
| P0 | [x] | `plugins/lastfm.py` -- Last.fm API client (artist.getSimilar, artist.getTopTags, track.getSimilar) |
| P0 | [x] | `!similar` command -- show similar artists for current or named track/artist |
| P0 | [x] | `!similar play` -- queue a similar track via YouTube search |
| P1 | [x] | `!tags` command -- show genre/style tags for current or named track |
| P1 | [x] | Config: `[lastfm] api_key` or `LASTFM_API_KEY` env var |
| P2 | [ ] | Tests: `test_lastfm.py` (API response mocking, command dispatch) |
| P2 | [ ] | Documentation update (USAGE.md, CHEATSHEET.md) |
## Previous Sprint -- v2.3.0 Mumble Voice + Multi-Bot (2026-02-22)
| Pri | Status | Task |
|-----|--------|------|
| P0 | [x] | `src/derp/mumble.py` -- rewrite to pymumble transport (voice + text) |
| P0 | [x] | `plugins/music.py` -- play/stop/skip/queue/np/volume/seek/resume |
| P0 | [x] | `plugins/voice.py` -- STT (Whisper) + TTS (Piper), voice profiles |
| P0 | [x] | Container patches for pymumble ssl + opuslib musl |
| P0 | [x] | Multi-bot Mumble (`[[mumble.extra]]`), per-bot plugin filtering |
| P0 | [x] | Rubberband pitch-shifting via CLI (Containerfile + FX chain split) |
| P0 | [x] | Bot audio ignored in sound callback (no self-ducking/STT of bots) |
| P0 | [x] | Self-mute support (mute on join, unmute for audio, re-mute after) |
| P1 | [x] | `plugins/alias.py` -- command aliases (add/del/list) |
| P1 | [x] | Container management tools (`tools/build,start,stop,restart,nuke,logs,status`) |
| P1 | [x] | Tests: `test_mumble.py`, `test_music.py`, `test_alias.py`, `test_core.py` |
| P2 | [x] | Documentation update (USAGE.md, CHEATSHEET.md, ROADMAP.md) |
## Previous Sprint -- v2.3.0 Mumble Music Playback (2026-02-21)
| Pri | Status | Task |
|-----|--------|------|
@@ -242,6 +281,8 @@
| Date | Task |
|------|------|
| 2026-02-22 | v2.3.0 (voice profiles, rubberband FX, multi-bot, self-mute, container tools) |
| 2026-02-21 | v2.3.0 (pymumble rewrite, music playback, fades, seek, kept library) |
| 2026-02-17 | v1.2.3 (paste overflow with FlaskPaste integration) |
| 2026-02-17 | v1.2.1 (HTTP opener cache, alert perf, concurrent multi-instance, tracemalloc) |
| 2026-02-16 | v1.2.0 (subscriptions, alerts, proxy, reminders) |
+34
View File
@@ -130,6 +130,17 @@ is preserved in git history for reference.
- [ ] SASL authentication
- [ ] TLS/STARTTLS connection
## Performance
- [ ] Iterative `_extract_videos` in alert.py (51K recursive calls, 6.7s CPU)
- [ ] Bypass SOCKS5 for local services (FlaskPaste, SearXNG)
- [ ] Connection pool tuning (529 SOCKS connections per 25min session)
- [ ] Async HTTP client (aiohttp + aiohttp-socks) to avoid blocking executors
- [x] Connection pooling via urllib3 SOCKSProxyManager
- [x] Batch OG fetch via ThreadPoolExecutor
- [x] HTTP opener caching at module level
- [x] Per-backend error tracking with exponential backoff
## Mumble
- [x] Mumble adapter via TCP/TLS + protobuf control channel (no SDK)
@@ -137,6 +148,29 @@ is preserved in git history for reference.
- [x] Text chat only (no voice)
- [x] Channel-based messaging
- [x] Minimal protobuf encoder/decoder (no protobuf dep)
- [x] pymumble transport rewrite (voice + text)
- [x] Music playback (yt-dlp + ffmpeg + Opus)
- [x] Voice STT/TTS (Whisper + Piper)
- [x] Multi-bot with per-bot plugin filtering
- [x] Configurable voice profiles (voice, FX chain)
- [x] Self-mute support (auto mute/unmute around audio)
- [x] Bot audio isolation (ignore own bots in sound callback)
- [x] Pause/unpause with position tracking, stale stream re-download, rewind + fade-in
- [x] Autoplay continuous radio (random kept track, silence-aware, configurable cooldown)
- [x] Periodic resume state persistence (survives hard kills)
- [x] Track duration in `!np` (ffprobe), optional `!announce` toggle
- [x] Direct bot addressing (`merlin: say <text>`)
- [x] Self-deafen on connect
- [ ] Per-channel voice settings (different voice per channel)
- [ ] Voice activity log (who spoke, duration, transcript)
## Music Discovery
- [ ] Last.fm integration (API key, free tier)
- [ ] `!similar` command -- find similar artists/tracks via Last.fm
- [ ] `!tags` command -- show genre/style tags for current track
- [ ] Auto-queue similar tracks when autoplay has no kept tracks
- [ ] MusicBrainz fallback (no API key, 1 req/sec rate limit)
## Slack
+333
View File
@@ -0,0 +1,333 @@
# Audio Engine -- Issues, Fixes, and Consolidation Notes
Technical reference for the Mumble audio pipeline: known issues,
applied fixes, architectural decisions, and areas for future work.
## Architecture Overview
```
yt-dlp -> ffmpeg (decode to s16le 48kHz mono) -> PCM frames (20ms)
-> volume ramp/scale -> pymumble sound_output -> Opus encode -> Mumble
```
Key components:
| File | Role |
|------|------|
| `src/derp/mumble.py` | `stream_audio()` -- PCM feed loop, volume ramp, seek |
| `plugins/music.py` | Queue, play loop, fade orchestration, duck monitor |
### Volume control layers (evaluated per-frame, highest priority first)
1. **fade_vol** -- active during fade-out (skip/stop/pause); set to 0 as target
2. **duck_vol** -- voice-activated ducking; snap to floor, linear restore
3. **volume** -- user-set level (0-100)
The play loop passes a lambda to `stream_audio`:
```python
volume=lambda: (
ps["fade_vol"] if ps["fade_vol"] is not None else
ps["duck_vol"] if ps["duck_vol"] is not None else
ps["volume"]
) / 100.0
```
### Per-frame volume ramping
`stream_audio` never jumps to the target volume. Each 20ms frame is
ramped from `_cur_vol` toward `target` by at most `step`:
- **_max_step** = 0.005 (~4s full ramp) -- ceiling for normal changes
- **fade_in_step** -- computed from fade-in duration (default 5s)
- **fade_step** -- override from plugin (fade-out on skip/stop/pause)
When `abs(diff) < 0.0001`, flat scaling is used (avoids ramp artifacts
on steady-state frames). Otherwise, `_scale_pcm_ramp()` linearly
interpolates across all 960 samples in the frame.
---
## Issues and Fixes
### 1. Alpine ffmpeg lacks librubberband
**Symptom:** 13/15 voice audition samples failed. `rubberband` audio
filter unavailable in ffmpeg.
**Root cause:** Alpine's ffmpeg package is compiled without
`--enable-librubberband`.
**Fix:** Added `rubberband` CLI package to `Containerfile`. Created
`_split_fx()` in `plugins/voice.py` to parse FX chains: pitch-shifting
goes through the `rubberband` CLI binary, remaining filters (bass, echo)
through ffmpeg. Two-stage pipeline.
**Files:** `Containerfile`, `plugins/voice.py`
---
### 2. Self-ducking between bots
**Symptom:** derp's music volume dropped when merlin spoke (TTS).
**Root cause:** merlin's TTS output triggered `_on_sound_received`,
which updated the shared `registry._voice_ts` timestamp. derp's duck
monitor saw recent voice activity and ducked.
**Fix:** `_on_sound_received` checks `registry._bots` and returns early
for any bot username -- no timestamp update, no listener dispatch.
```python
def _on_sound_received(self, user, sound_chunk) -> None:
name = user["name"] if isinstance(user, dict) else None
bots = getattr(self.registry, "_bots", {})
if name and name in bots:
return # ignore audio from bots entirely
```
**Files:** `src/derp/mumble.py`
---
### 3. Click/pop on skip/stop (fade-out cancellation)
**Symptom:** Audible glitch at the end of fade-out when skipping or
stopping a track.
**Root cause:** `_fade_and_cancel()` fades volume to 0 over ~3s, then
calls `task.cancel()`. In `stream_audio`, `CancelledError` triggers
`clear_buffer()`, which drops any frames still queued in pymumble's
output -- including frames that were encoded at non-zero amplitude a
few frames earlier. The sudden buffer wipe produces a click.
**Fix (two-part):**
1. **Plugin side** (`music.py`): Added 150ms post-fade drain before
cancel, giving pymumble time to flush remaining silent frames.
2. **Engine side** (`mumble.py`): `CancelledError` handler only calls
`clear_buffer()` if `_cur_vol > 0.01`. When a fade-out has already
driven volume to ~0, the remaining buffer frames are silent and
clearing them is unnecessary.
```python
# mumble.py -- CancelledError handler
if _cur_vol > 0.01:
self._mumble.sound_output.clear_buffer()
```
```python
# music.py -- _fade_and_cancel()
await asyncio.sleep(duration)
await asyncio.sleep(0.15) # drain window
task.cancel()
```
**Files:** `src/derp/mumble.py`, `plugins/music.py`
---
### 4. Fade-out math
**How it works:** `_fade_and_cancel(duration=3.0)` computes the
per-frame step from the current effective volume:
```python
cur_vol = (duck_vol or volume) / 100.0
n_frames = duration / 0.02 # 150 frames for 3s
step = cur_vol / n_frames
```
The play loop sets `ps["fade_vol"] = 0` (the target) and
`ps["fade_step"] = step` (the rate). `stream_audio` ramps `_cur_vol`
toward 0 at `step` per frame. At 50% volume: step = 0.0033, reaching
zero in exactly 150 frames (3.0s).
**Note:** `fade_vol` is set to 0 immediately, making the volume lambda
return 0 as the target. The ramp code smoothly transitions -- there is
no abrupt jump because `_cur_vol` tracks actual output level, not the
target.
---
### 5. Self-mute lifecycle
**Requirement:** merlin mutes on connect, unmutes only when emitting
audio (TTS), re-mutes after a delay.
**Implementation:**
```
connect -> mute()
stream_audio start -> cancel pending mute task, unmute()
stream_audio finally -> spawn _delayed_mute(3.0)
```
The 3-second delay prevents rapid mute/unmute flicker on back-to-back
TTS. The mute task is cancelled if new audio starts before it fires.
**Config:** `self_mute = true` in `[[mumble.extra]]`
**Files:** `src/derp/mumble.py`
---
### 6. Self-deafen on connect
**Requirement:** merlin deafens on connect (no audio reception needed).
**Implementation:** `self_deaf = true` config flag, calls
`self._mumble.users.myself.deafen()` in `_on_connected`.
**Files:** `src/derp/mumble.py`, `config/derp.toml`
---
## Pause/Resume
### Design
`!pause` toggles between paused and playing states:
**Pause:** Captures current track + elapsed position + monotonic
timestamp. Fades out, cancels play loop. Queue is preserved.
**Unpause:** Re-inserts track at queue front, starts play loop with
seek. Two special behaviors:
1. **Rewind:** 3s rewind on unpause for continuity (only if paused >= 3s
to prevent anti-flood: rapid toggle doesn't compound the rewind).
2. **Stale stream:** If paused > 45s, cached stream files (in
`data/music/cache/`) are deleted so the play loop re-downloads.
Kept files (`data/music/`) are never deleted. Stream URLs from
YouTube et al. expire within minutes.
3. **Fade-in:** Unpause always uses `fade_in=True` (5s ramp from 0).
**State cleanup:** `!stop` clears `ps["paused"]`. The play loop's
`finally` block skips `_cleanup_track` when paused (preserves the file).
---
## Autoplay
### Design
When `autoplay = true` (config), the play loop stays alive after the
queue empties:
1. Waits for silence (duck_silence threshold, default 15s)
2. Picks one random kept track
3. Plays it
4. On completion, loops back to step 1
This replaces the previous bulk-queue approach (shuffle all kept tracks
at once). Benefits: no large upfront queue, silence-aware gaps between
tracks, indefinite looping.
### Resume persistence
A background task saves track URL + elapsed position to the state DB
every 10 seconds during playback:
```python
async def _periodic_save():
while True:
await asyncio.sleep(10)
el = cur_seek + progress[0] * 0.02
if el > 1.0:
_save_resume(bot, track, el)
```
On hard kill: resumes from at most ~10s behind. On normal track
completion: `_clear_resume()` wipes the state.
---
## Voice Ducking
### Flow
```
voice detected -> duck_vol = floor (instant)
silence > duck_silence -> linear restore over duck_restore seconds
```
The duck monitor runs as a background task alongside the play loop.
It updates `ps["duck_vol"]` which the volume lambda reads per-frame.
### Restore ramp
Restoration is linear from floor to user volume. The per-frame ramp in
`stream_audio` further smooths each 1-second update from the monitor,
eliminating audible steps.
### Bot audio isolation
Bot usernames (from `registry._bots`) are excluded from
`_on_sound_received` entirely -- no timestamp update, no listener
dispatch. This prevents self-ducking between derp and merlin.
---
## Seek (in-stream pipeline swap)
### Design
Seek rebuilds the ffmpeg pipeline at the new position without cancelling
the play loop task. This avoids the overhead of re-downloading.
1. Set `_seek_fading = True`, `_seek_fade_out = 10` (0.2s ramp-down)
2. Continue reading frames, scaling by decreasing ratio
3. At fade-out = 0: kill ffmpeg, clear buffer, spawn new pipeline
4. 0.5s fade-in on the new pipeline
### Consolidation note
Seek fade-out (10 frames / 0.2s) is much shorter than skip/stop
fade-out (3s). This is intentional -- seek should feel responsive.
The mechanisms are separate: seek uses frame-counting in
`stream_audio`, skip/stop uses `_fade_and_cancel` in the plugin.
---
## Consolidation Opportunities
### Volume control unification
Three volume layers (fade_vol, duck_vol, volume) evaluated in a lambda
per-frame. Works but the priority logic is implicit. A future refactor
could use a single `effective_volume()` method that explicitly resolves
priority and makes the per-frame cost clearer.
### Fade-out ownership
Skip/stop/pause all route through `_fade_and_cancel()` -- good. But the
fade target is communicated indirectly via `ps["fade_vol"] = 0` and
`ps["fade_step"]`, read by a lambda in the play loop, evaluated in
`stream_audio`. A more explicit signal (e.g. an asyncio.Event or a
dedicated fade state machine in `stream_audio`) could simplify reasoning
about timing.
### Buffer drain timing
The 150ms post-fade drain is empirical. A more robust approach would be
to query `sound_output.get_buffer_size()` and wait for it to drop below
a threshold before cancelling. This would adapt to varying network
conditions and pymumble buffer sizes.
### Track duration
Duration is probed via `ffprobe` after download (blocking, run in
executor). For kept tracks, it's stored in state metadata. This is
duplicated -- kept track metadata already has duration from
`_fetch_metadata` (yt-dlp). The `ffprobe` path is the fallback for
non-kept tracks. Could unify by always probing locally.
### Periodic resume save interval
Currently 10s fixed. Could be adaptive -- save more frequently near
the start of a track (where losing position is more noticeable) and
less frequently later. Marginal benefit vs. complexity though.
+25 -4
View File
@@ -53,14 +53,34 @@ format = "json" # JSONL output (default: "text")
## Container
```bash
make build # Build image (only for dep changes)
make up # Start (podman-compose)
make down # Stop
make logs # Follow logs
tools/build # Build image
tools/build --no-cache # Rebuild from scratch
tools/start # Start (builds if no image)
tools/stop # Stop and remove container
tools/restart # Stop + rebuild + start
tools/restart --no-cache # Full clean restart
tools/logs # Tail logs (default 30 lines)
tools/logs 100 # Tail last 100 lines
tools/status # Container, image, mount state
tools/nuke # Full teardown (container + image)
```
Code, plugins, config, and data are bind-mounted. No rebuild needed for
code changes -- restart the container or use `!reload` for plugins.
Rebuild only when `requirements.txt` or `Containerfile` change.
## Profiling
```bash
tools/profile # Top 30 by cumulative time
tools/profile -s tottime -n 20 # Top 20 by total time
tools/profile -f mumble # Filter to mumble functions
tools/profile -c -f stream_audio # Who calls stream_audio
tools/profile data/old.prof # Analyze a specific file
```
Sort keys: `cumtime`, `tottime`, `calls`, `name`.
Profile data written on graceful shutdown when bot runs with `--cprofile`.
## Bot Commands
@@ -562,6 +582,7 @@ HTML stripped on receive, escaped on send. IRC-only commands are no-ops.
!keep # Keep current file + save metadata
!kept # List kept files with metadata
!kept clear # Delete all kept files + metadata
!kept repair # Re-download missing kept files
!duck # Show ducking status
!duck on # Enable voice ducking
!duck off # Disable voice ducking
+3 -1
View File
@@ -1628,7 +1628,7 @@ and voice transmission.
!np Now playing
!volume [0-100] Get/set volume (persisted across restarts)
!keep Keep current track's audio file (with metadata)
!kept [clear] List kept files with metadata, or clear all
!kept [clear|repair] List kept files, clear all, or re-download missing
!testtone Play 3-second 440Hz test tone
```
@@ -1751,6 +1751,8 @@ file (natural dedup).
- Use `!kept` to list preserved files with metadata (title, artist, duration,
file size)
- Use `!kept clear` to delete all preserved files and their metadata
- Use `!kept repair` to re-download any kept tracks whose local files are
missing (e.g. after a cleanup or volume mount issue)
- On cancel/error, files are not deleted (needed for `!resume`)
### Extra Mumble Bots
+46 -35
View File
@@ -368,45 +368,56 @@ def _fetch_og_batch(urls: list[str]) -> dict[str, tuple[str, str, str]]:
# -- YouTube InnerTube search (blocking) ------------------------------------
def _extract_videos(obj: object, depth: int = 0) -> list[dict]:
"""Recursively walk YouTube JSON to find video results.
"""Walk YouTube JSON to find video results (iterative).
Finds all objects containing both 'videoId' and 'title' keys.
Resilient to YouTube rearranging wrapper layers.
Uses an explicit stack instead of recursion to avoid 50K+ call
overhead on deeply nested InnerTube responses.
"""
if depth > 20:
return []
results = []
if isinstance(obj, dict):
video_id = obj.get("videoId")
title_obj = obj.get("title")
if isinstance(video_id, str) and video_id and title_obj is not None:
if isinstance(title_obj, dict):
runs = title_obj.get("runs", [])
title = "".join(r.get("text", "") for r in runs if isinstance(r, dict))
elif isinstance(title_obj, str):
title = title_obj
else:
title = ""
if title:
# Extract relative publish time (e.g. "2 days ago")
pub_obj = obj.get("publishedTimeText")
date = ""
if isinstance(pub_obj, dict):
date = pub_obj.get("simpleText", "")
elif isinstance(pub_obj, str):
date = pub_obj
results.append({
"id": video_id,
"title": title,
"url": f"https://www.youtube.com/watch?v={video_id}",
"date": date,
"extra": "",
})
for val in obj.values():
results.extend(_extract_videos(val, depth + 1))
elif isinstance(obj, list):
for item in obj:
results.extend(_extract_videos(item, depth + 1))
_MAX_DEPTH = 20
results: list[dict] = []
# Stack of (node, depth) tuples
stack: list[tuple[object, int]] = [(obj, 0)]
while stack:
node, d = stack.pop()
if d > _MAX_DEPTH:
continue
if isinstance(node, dict):
video_id = node.get("videoId")
title_obj = node.get("title")
if isinstance(video_id, str) and video_id and title_obj is not None:
if isinstance(title_obj, dict):
runs = title_obj.get("runs", [])
title = "".join(
r.get("text", "") for r in runs if isinstance(r, dict)
)
elif isinstance(title_obj, str):
title = title_obj
else:
title = ""
if title:
pub_obj = node.get("publishedTimeText")
date = ""
if isinstance(pub_obj, dict):
date = pub_obj.get("simpleText", "")
elif isinstance(pub_obj, str):
date = pub_obj
results.append({
"id": video_id,
"title": title,
"url": f"https://www.youtube.com/watch?v={video_id}",
"date": date,
"extra": "",
})
# Reverse to preserve original traversal order (stack is LIFO)
children = [v for v in node.values() if isinstance(v, (dict, list))]
for val in reversed(children):
stack.append((val, d + 1))
elif isinstance(node, list):
for item in reversed(node):
if isinstance(item, (dict, list)):
stack.append((item, d + 1))
return results
+85
View File
@@ -0,0 +1,85 @@
"""Plugin: user-defined command aliases (persistent)."""
from __future__ import annotations
import logging
from derp.plugin import command
log = logging.getLogger(__name__)
_NS = "alias"
@command("alias", help="Aliases: !alias add|del|list|clear")
async def cmd_alias(bot, message):
"""Create short aliases for existing bot commands.
Usage:
!alias add <name> <target> Create alias (e.g. !alias add s skip)
!alias del <name> Remove alias
!alias list Show all aliases
!alias clear Remove all aliases (admin only)
"""
parts = message.text.split(None, 3)
if len(parts) < 2:
await bot.reply(message, "Usage: !alias <add|del|list|clear> [args]")
return
sub = parts[1].lower()
if sub == "add":
if len(parts) < 4:
await bot.reply(message, "Usage: !alias add <name> <target>")
return
name = parts[2].lower()
target = parts[3].lower()
# Cannot shadow an existing registered command
if name in bot.registry.commands:
await bot.reply(message, f"'{name}' is already a registered command")
return
# Cannot alias to another alias (single-level only)
if bot.state.get(_NS, target) is not None:
await bot.reply(message, f"'{target}' is itself an alias; no chaining")
return
# Target must resolve to a real command
if target not in bot.registry.commands:
await bot.reply(message, f"unknown command: {target}")
return
bot.state.set(_NS, name, target)
await bot.reply(message, f"alias: {name} -> {target}")
elif sub == "del":
if len(parts) < 3:
await bot.reply(message, "Usage: !alias del <name>")
return
name = parts[2].lower()
if bot.state.delete(_NS, name):
await bot.reply(message, f"alias removed: {name}")
else:
await bot.reply(message, f"no alias: {name}")
elif sub == "list":
keys = bot.state.keys(_NS)
if not keys:
await bot.reply(message, "No aliases defined")
return
entries = []
for key in sorted(keys):
target = bot.state.get(_NS, key)
entries.append(f"{key} -> {target}")
await bot.reply(message, "Aliases: " + ", ".join(entries))
elif sub == "clear":
if not bot._is_admin(message):
await bot.reply(message, "Permission denied: clear requires admin")
return
count = bot.state.clear(_NS)
await bot.reply(message, f"Cleared {count} alias(es)")
else:
await bot.reply(message, "Usage: !alias <add|del|list|clear> [args]")
+28
View File
@@ -174,6 +174,34 @@ async def cmd_admins(bot, message):
await bot.reply(message, " | ".join(parts))
@command("deaf", help="Toggle voice listener deaf on Mumble")
async def cmd_deaf(bot, message):
"""Toggle the voice listener's deaf state on Mumble.
Targets the bot with ``receive_sound = true`` (merlin) so that
deafening stops ducking without affecting the music bot's playback.
"""
# Find the listener bot (receive_sound=true) among registered peers
listener = None
bots = getattr(bot.registry, "_bots", {})
for peer in bots.values():
if getattr(peer, "_receive_sound", False):
listener = peer
break
mumble = getattr(listener or bot, "_mumble", None)
if mumble is None:
return
myself = mumble.users.myself
name = getattr(listener, "nick", "bot")
if myself.get("self_deaf", False):
myself.undeafen()
myself.unmute()
await bot.reply(message, f"{name}: undeafened")
else:
myself.deafen()
await bot.reply(message, f"{name}: deafened")
@command("state", help="Inspect plugin state: !state <list|get|del|clear> ...", admin=True)
async def cmd_state(bot, message):
"""Manage the plugin state store.
+1 -1
View File
@@ -114,7 +114,7 @@ def _create_paste(base_url: str, content: str) -> str:
body = json.loads(resp.read())
paste_id = body.get("id", "")
if paste_id:
return f"{base_url}/{paste_id}"
return f"{base_url}/{paste_id}/raw"
return body.get("url", "")
+272
View File
@@ -0,0 +1,272 @@
"""Plugin: music discovery via Last.fm API."""
from __future__ import annotations
import json
import logging
import os
import random
from urllib.parse import urlencode
from derp.plugin import command
log = logging.getLogger(__name__)
_BASE = "https://ws.audioscrobbler.com/2.0/"
# -- Config ------------------------------------------------------------------
def _get_api_key(bot) -> str:
"""Resolve Last.fm API key from env or config."""
return (os.environ.get("LASTFM_API_KEY", "")
or bot.config.get("lastfm", {}).get("api_key", ""))
# -- API helpers -------------------------------------------------------------
def _api_call(api_key: str, method: str, **params) -> dict:
"""Blocking Last.fm API call. Run in executor."""
from derp.http import urlopen
qs = urlencode({
"method": method,
"api_key": api_key,
"format": "json",
**params,
})
url = f"{_BASE}?{qs}"
try:
resp = urlopen(url, timeout=10)
return json.loads(resp.read().decode())
except Exception:
log.exception("lastfm: API call failed: %s", method)
return {}
def _get_similar_artists(api_key: str, artist: str,
limit: int = 10) -> list[dict]:
"""Fetch similar artists for a given artist name."""
data = _api_call(api_key, "artist.getSimilar",
artist=artist, limit=str(limit))
artists = data.get("similarartists", {}).get("artist", [])
if isinstance(artists, dict):
artists = [artists]
return artists
def _get_top_tags(api_key: str, artist: str) -> list[dict]:
"""Fetch top tags for an artist."""
data = _api_call(api_key, "artist.getTopTags", artist=artist)
tags = data.get("toptags", {}).get("tag", [])
if isinstance(tags, dict):
tags = [tags]
return tags
def _get_similar_tracks(api_key: str, artist: str, track: str,
limit: int = 10) -> list[dict]:
"""Fetch similar tracks for a given artist + track."""
data = _api_call(api_key, "track.getSimilar",
artist=artist, track=track, limit=str(limit))
tracks = data.get("similartracks", {}).get("track", [])
if isinstance(tracks, dict):
tracks = [tracks]
return tracks
def _search_track(api_key: str, query: str,
limit: int = 5) -> list[dict]:
"""Search Last.fm for tracks matching a query."""
data = _api_call(api_key, "track.search",
track=query, limit=str(limit))
results = data.get("results", {}).get("trackmatches", {}).get("track", [])
if isinstance(results, dict):
results = [results]
return results
# -- Metadata extraction -----------------------------------------------------
def _current_meta(bot) -> tuple[str, str]:
"""Extract artist and title from the currently playing track.
Returns (artist, title). Either or both may be empty.
Tries the music plugin's current track metadata, falling back to
splitting the title on common separators.
"""
music_ps = bot._pstate.get("music", {})
current = music_ps.get("current")
if current is None:
return ("", "")
raw_title = current.title or ""
# Try common "Artist - Title" patterns
for sep in (" - ", " -- ", " | ", " ~ "):
if sep in raw_title:
parts = raw_title.split(sep, 1)
return (parts[0].strip(), parts[1].strip())
# No separator -- treat whole thing as a search query
return ("", raw_title)
# -- Formatting --------------------------------------------------------------
def _fmt_match(m: float | str) -> str:
"""Format a Last.fm match score as a percentage."""
try:
return f"{float(m) * 100:.0f}%"
except (ValueError, TypeError):
return ""
# -- Commands ----------------------------------------------------------------
@command("similar", help="Music: !similar [artist|play] -- find similar music")
async def cmd_similar(bot, message):
"""Find similar artists or tracks.
Usage:
!similar Similar to currently playing track
!similar <artist> Similar artists to named artist
!similar play Queue a random similar track
!similar play <artist> Queue a similar track for named artist
"""
api_key = _get_api_key(bot)
if not api_key:
await bot.reply(message, "Last.fm API key not configured")
return
parts = message.text.split(None, 2)
# !similar play [artist]
play_mode = len(parts) >= 2 and parts[1].lower() == "play"
if play_mode:
query = parts[2].strip() if len(parts) > 2 else ""
else:
query = parts[1].strip() if len(parts) > 1 else ""
import asyncio
loop = asyncio.get_running_loop()
# Resolve artist from query or current track
if query:
artist = query
title = ""
else:
artist, title = _current_meta(bot)
if not artist and not title:
await bot.reply(message, "Nothing playing and no artist given")
return
# Try track-level similarity first if we have both artist + title
similar = []
if artist and title:
similar = await loop.run_in_executor(
None, _get_similar_tracks, api_key, artist, title,
)
# Fall back to artist-level similarity
if not similar:
search_artist = artist or title
similar_artists = await loop.run_in_executor(
None, _get_similar_artists, api_key, search_artist,
)
if not similar_artists:
await bot.reply(message, f"No similar artists found for '{search_artist}'")
return
if play_mode:
# Pick a random similar artist and search YouTube
pick = random.choice(similar_artists[:10])
pick_name = pick.get("name", "")
if not pick_name:
await bot.reply(message, "No playable result found")
return
# Inject a !play command with a YouTube search
message.text = f"!play {pick_name}"
music_mod = bot.registry._modules.get("music")
if music_mod:
await music_mod.cmd_play(bot, message)
return
# Display similar artists
lines = [f"Similar to {search_artist}:"]
for a in similar_artists[:8]:
name = a.get("name", "?")
match = _fmt_match(a.get("match", ""))
suffix = f" ({match})" if match else ""
lines.append(f" {name}{suffix}")
await bot.long_reply(message, lines, label="similar artists")
return
# Track-level results
if play_mode:
pick = random.choice(similar[:10])
pick_artist = pick.get("artist", {}).get("name", "")
pick_title = pick.get("name", "")
search = f"{pick_artist} {pick_title}".strip()
if not search:
await bot.reply(message, "No playable result found")
return
message.text = f"!play {search}"
music_mod = bot.registry._modules.get("music")
if music_mod:
await music_mod.cmd_play(bot, message)
return
# Display similar tracks
lines = [f"Similar to {artist} - {title}:"]
for t in similar[:8]:
t_artist = t.get("artist", {}).get("name", "")
t_name = t.get("name", "?")
match = _fmt_match(t.get("match", ""))
suffix = f" ({match})" if match else ""
lines.append(f" {t_artist} - {t_name}{suffix}")
await bot.long_reply(message, lines, label="similar tracks")
@command("tags", help="Music: !tags [artist] -- show genre tags")
async def cmd_tags(bot, message):
"""Show genre/style tags for an artist.
Usage:
!tags Tags for currently playing artist
!tags <artist> Tags for named artist
"""
api_key = _get_api_key(bot)
if not api_key:
await bot.reply(message, "Last.fm API key not configured")
return
parts = message.text.split(None, 1)
query = parts[1].strip() if len(parts) > 1 else ""
import asyncio
loop = asyncio.get_running_loop()
if query:
artist = query
else:
artist, title = _current_meta(bot)
artist = artist or title
if not artist:
await bot.reply(message, "Nothing playing and no artist given")
return
tags = await loop.run_in_executor(
None, _get_top_tags, api_key, artist,
)
if not tags:
await bot.reply(message, f"No tags found for '{artist}'")
return
# Show top tags with counts
tag_names = [t.get("name", "?") for t in tags[:10] if t.get("name")]
await bot.reply(message, f"{artist}: {', '.join(tag_names)}")
+651 -92
View File
File diff suppressed because it is too large Load Diff
+2 -1
View File
@@ -6,6 +6,7 @@ import json
import urllib.parse
import urllib.request
from derp.http import urlopen as _urlopen
from derp.plugin import command
# -- Constants ---------------------------------------------------------------
@@ -38,7 +39,7 @@ def _search(query: str) -> list[dict]:
url = f"{_SEARX_URL}?{params}"
req = urllib.request.Request(url, method="GET")
resp = urllib.request.urlopen(req, timeout=_FETCH_TIMEOUT)
resp = _urlopen(req, timeout=_FETCH_TIMEOUT, proxy=False)
raw = resp.read()
resp.close()
+259 -19
View File
@@ -38,6 +38,18 @@ _MAX_SAY_LEN = 500 # max characters for !say
_WHISPER_URL = "http://192.168.129.9:8080/inference"
_PIPER_URL = "http://192.168.129.9:5100/"
def _find_voice_peer(bot):
"""Find the voice-capable peer (the bot with 'voice' in only_plugins)."""
bots = getattr(bot.registry, "_bots", {})
for name, b in bots.items():
if name == bot._username:
continue
if getattr(b, "_only_plugins", None) and "voice" in b._only_plugins:
return b
return None
# -- Per-bot state -----------------------------------------------------------
@@ -54,6 +66,11 @@ def _ps(bot):
"silence_gap": cfg.get("silence_gap", _SILENCE_GAP),
"whisper_url": cfg.get("whisper_url", _WHISPER_URL),
"piper_url": cfg.get("piper_url", _PIPER_URL),
"voice": cfg.get("voice", ""),
"length_scale": cfg.get("length_scale", 1.0),
"noise_scale": cfg.get("noise_scale", 0.667),
"noise_w": cfg.get("noise_w", 0.8),
"fx": cfg.get("fx", ""),
"_listener_registered": False,
})
@@ -167,8 +184,10 @@ async def _flush_monitor(bot):
remainder = text[len(trigger):].strip()
if remainder:
log.info("voice: trigger from %s: %s", name, remainder)
bot._spawn(
_tts_play(bot, remainder), name="voice-tts",
# Route TTS through voice-capable peer if available
speaker = _find_voice_peer(bot) or bot
speaker._spawn(
_tts_play(speaker, remainder), name="voice-tts",
)
continue
@@ -210,21 +229,40 @@ def _fetch_tts(piper_url: str, text: str) -> str | None:
async def _tts_play(bot, text: str):
"""Fetch TTS audio and play it via stream_audio."""
"""Fetch TTS audio and play it via stream_audio.
Uses the configured voice profile (voice, fx, piper params) when set,
otherwise falls back to Piper's default voice.
"""
from pathlib import Path
ps = _ps(bot)
loop = asyncio.get_running_loop()
wav_path = await loop.run_in_executor(
None, _fetch_tts, ps["piper_url"], text,
)
if ps["voice"] or ps["fx"]:
wav_path = await loop.run_in_executor(
None, lambda: _fetch_tts_voice(
ps["piper_url"], text,
voice=ps["voice"],
length_scale=ps["length_scale"],
noise_scale=ps["noise_scale"],
noise_w=ps["noise_w"],
fx=ps["fx"],
),
)
else:
wav_path = await loop.run_in_executor(
None, _fetch_tts, ps["piper_url"], text,
)
if wav_path is None:
return
try:
# Signal music plugin to duck while TTS is playing
bot.registry._tts_active = True
done = asyncio.Event()
await bot.stream_audio(str(wav_path), volume=1.0, on_done=done)
await done.wait()
finally:
bot.registry._tts_active = False
Path(wav_path).unlink(missing_ok=True)
@@ -322,26 +360,228 @@ async def cmd_say(bot, message):
bot._spawn(_tts_play(bot, text), name="voice-tts")
def _split_fx(fx: str) -> tuple[list[str], str]:
"""Split FX chain into rubberband CLI args and ffmpeg filter string.
Alpine's ffmpeg lacks librubberband, so pitch shifting is handled by
the ``rubberband`` CLI tool and remaining filters by ffmpeg.
"""
import math
parts = fx.split(",")
rb_args: list[str] = []
ff_parts: list[str] = []
for part in parts:
if part.startswith("rubberband="):
opts: dict[str, str] = {}
for kv in part[len("rubberband="):].split(":"):
k, _, v = kv.partition("=")
opts[k] = v
if "pitch" in opts:
semitones = 12 * math.log2(float(opts["pitch"]))
rb_args += ["--pitch", f"{semitones:.2f}"]
if opts.get("formant") == "1":
rb_args.append("--formant")
else:
ff_parts.append(part)
return rb_args, ",".join(ff_parts)
def _fetch_tts_voice(piper_url: str, text: str, *, voice: str = "",
speaker_id: int = 0, length_scale: float = 1.0,
noise_scale: float = 0.667, noise_w: float = 0.8,
fx: str = "") -> str | None:
"""Fetch TTS with explicit voice params and optional FX. Blocking.
Pitch shifting uses the ``rubberband`` CLI (Alpine ffmpeg has no
librubberband); remaining audio filters go through ffmpeg.
"""
import os
import subprocess
import tempfile
payload = {"text": text}
if voice:
payload["voice"] = voice
if speaker_id:
payload["speaker_id"] = speaker_id
payload["length_scale"] = length_scale
payload["noise_scale"] = noise_scale
payload["noise_w"] = noise_w
data = json.dumps(payload).encode()
req = urllib.request.Request(piper_url, data=data, method="POST")
req.add_header("Content-Type", "application/json")
resp = _urlopen(req, timeout=30, proxy=False)
wav_data = resp.read()
resp.close()
if not wav_data:
return None
tmp = tempfile.NamedTemporaryFile(suffix=".wav", prefix="derp_aud_", delete=False)
tmp.write(wav_data)
tmp.close()
if not fx:
return tmp.name
rb_args, ff_filters = _split_fx(fx)
current = tmp.name
# Pitch shift via rubberband CLI
if rb_args:
rb_out = tempfile.NamedTemporaryFile(
suffix=".wav", prefix="derp_aud_", delete=False,
)
rb_out.close()
r = subprocess.run(
["rubberband"] + rb_args + [current, rb_out.name],
capture_output=True, timeout=15,
)
os.unlink(current)
if r.returncode != 0:
log.warning("voice: rubberband failed: %s", r.stderr[:200])
os.unlink(rb_out.name)
return None
current = rb_out.name
# Remaining filters via ffmpeg
if ff_filters:
ff_out = tempfile.NamedTemporaryFile(
suffix=".wav", prefix="derp_aud_", delete=False,
)
ff_out.close()
r = subprocess.run(
["ffmpeg", "-y", "-i", current, "-af", ff_filters, ff_out.name],
capture_output=True, timeout=15,
)
os.unlink(current)
if r.returncode != 0:
log.warning("voice: ffmpeg failed: %s", r.stderr[:200])
os.unlink(ff_out.name)
return None
current = ff_out.name
return current
@command("audition", help="Voice: !audition -- play voice samples", tier="admin")
async def cmd_audition(bot, message):
"""Play voice samples through Mumble for comparison."""
if not _is_mumble(bot):
return
ps = _ps(bot)
piper_url = ps["piper_url"]
phrase = "The sorcerer has arrived. I have seen things beyond your understanding."
# FX building blocks
_deep = "rubberband=pitch=0.87:formant=1"
_bass = "bass=g=6:f=110:w=0.6"
_bass_heavy = "equalizer=f=80:t=h:w=150:g=8"
_echo_subtle = "aecho=0.8:0.6:25|40:0.25|0.15"
_echo_chamber = "aecho=0.8:0.88:60:0.35"
_echo_cave = "aecho=0.8:0.7:40|70|100:0.3|0.2|0.1"
samples = [
# -- Base voices (no FX) for reference
("ryan-high raw", "en_US-ryan-high", 0, ""),
("lessac-high raw", "en_US-lessac-high", 0, ""),
# -- Deep pitch only
("ryan deep", "en_US-ryan-high", 0,
_deep),
("lessac deep", "en_US-lessac-high", 0,
_deep),
# -- Deep + bass boost
("ryan deep+bass", "en_US-ryan-high", 0,
f"{_deep},{_bass}"),
("lessac deep+bass", "en_US-lessac-high", 0,
f"{_deep},{_bass}"),
# -- Deep + heavy bass
("ryan deep+heavy bass", "en_US-ryan-high", 0,
f"{_deep},{_bass_heavy}"),
# -- Deep + bass + subtle echo
("ryan deep+bass+echo", "en_US-ryan-high", 0,
f"{_deep},{_bass},{_echo_subtle}"),
("lessac deep+bass+echo", "en_US-lessac-high", 0,
f"{_deep},{_bass},{_echo_subtle}"),
# -- Deep + bass + chamber reverb
("ryan deep+bass+chamber", "en_US-ryan-high", 0,
f"{_deep},{_bass},{_echo_chamber}"),
("lessac deep+bass+chamber", "en_US-lessac-high", 0,
f"{_deep},{_bass},{_echo_chamber}"),
# -- Deep + heavy bass + cave reverb
("ryan deep+heavybass+cave", "en_US-ryan-high", 0,
f"{_deep},{_bass_heavy},{_echo_cave}"),
# -- Libritts best candidates with full sorcerer chain
("libritts #20 deep+bass+echo", "en_US-libritts_r-medium", 20,
f"{_deep},{_bass},{_echo_subtle}"),
("libritts #22 deep+bass+echo", "en_US-libritts_r-medium", 22,
f"{_deep},{_bass},{_echo_subtle}"),
("libritts #79 deep+bass+chamber", "en_US-libritts_r-medium", 79,
f"{_deep},{_bass},{_echo_chamber}"),
]
# Find merlin (the listener bot) -- plays the audition samples
merlin = None
for peer in getattr(bot.registry, "_bots", {}).values():
if getattr(peer, "_receive_sound", False):
merlin = peer
break
await bot.reply(message, f"Auditioning {len(samples)} voice samples...")
loop = asyncio.get_running_loop()
from pathlib import Path
# Pre-generate derp's default voice (same phrase, no FX)
derp_wav = await loop.run_in_executor(
None, lambda: _fetch_tts_voice(piper_url, phrase),
)
for i, (label, voice, sid, fx) in enumerate(samples, 1):
announcer = merlin or bot
await announcer.send("0", f"[{i}/{len(samples)}] {label}")
await asyncio.sleep(1)
# Generate the audition sample (merlin's candidate voice)
sample_wav = await loop.run_in_executor(
None, lambda v=voice, s=sid, f=fx: _fetch_tts_voice(
piper_url, phrase, voice=v, speaker_id=s,
length_scale=1.15, noise_scale=0.4, noise_w=0.5, fx=f,
),
)
if sample_wav is None:
await bot.send("0", " (failed)")
continue
try:
# Both bots speak simultaneously:
# merlin plays the audition sample, derp plays its default voice
merlin_done = asyncio.Event()
derp_done = asyncio.Event()
if merlin:
merlin_task = asyncio.create_task(
merlin.stream_audio(sample_wav, volume=1.0,
on_done=merlin_done))
derp_task = asyncio.create_task(
bot.stream_audio(derp_wav, volume=1.0,
on_done=derp_done))
await asyncio.gather(merlin_task, derp_task)
else:
await bot.stream_audio(sample_wav, volume=1.0,
on_done=merlin_done)
await merlin_done.wait()
finally:
Path(sample_wav).unlink(missing_ok=True)
await asyncio.sleep(2)
if derp_wav:
Path(derp_wav).unlink(missing_ok=True)
announcer = merlin or bot
await announcer.send("0", "Audition complete.")
# -- Plugin lifecycle --------------------------------------------------------
async def on_connected(bot) -> None:
"""Re-register listener after reconnect; play TTS greeting on first join."""
"""Re-register listener after reconnect."""
if not _is_mumble(bot):
return
ps = _ps(bot)
# TTS greeting on first connect
greet = bot.config.get("mumble", {}).get("greet")
if greet and not ps.get("_greeted"):
ps["_greeted"] = True
# Wait for audio subsystem to be ready
for _ in range(20):
if bot._is_audio_ready():
break
await asyncio.sleep(0.5)
bot._spawn(_tts_play(bot, greet), name="voice-greet")
if ps["listen"] or ps["trigger"]:
_ensure_listener(bot)
_ensure_flush_task(bot)
+6
View File
@@ -405,6 +405,12 @@ class Bot:
parts = text[len(self.prefix):].split(None, 1)
cmd_name = parts[0].lower() if parts else ""
handler = self._resolve_command(cmd_name)
if handler is None:
# Check user-defined aliases
target = self.state.get("alias", cmd_name) if hasattr(self, "state") else None
if target:
cmd_name = target
handler = self._resolve_command(cmd_name)
if handler is None:
return
if handler is _AMBIGUOUS:
+16 -3
View File
@@ -161,11 +161,24 @@ def main(argv: list[str] | None = None) -> int:
merged_mu = dict(config["mumble"])
merged_mu.update(extra)
merged_mu.pop("extra", None)
# Plugin filters are exclusive; don't inherit the parent's
if "only_plugins" in extra:
merged_mu.pop("except_plugins", None)
elif "except_plugins" in extra:
merged_mu.pop("only_plugins", None)
extra_cfg["mumble"] = merged_mu
# Extra bots don't run voice trigger by default
if "voice" not in extra:
extra_cfg["voice"] = {}
username = extra.get("username", f"mumble-{len(bots)}")
# Voice config: per-bot [<username>.voice] overrides global [voice]
per_bot_voice = config.get(username, {}).get("voice")
if per_bot_voice:
voice_cfg = dict(config.get("voice", {}))
voice_cfg.update(per_bot_voice)
extra_cfg["voice"] = voice_cfg
elif "voice" not in extra:
extra_cfg["voice"] = {
k: v for k, v in config.get("voice", {}).items()
if k != "trigger"
}
bot = MumbleBot(username, extra_cfg, registry)
bots.append(bot)
+41 -8
View File
@@ -40,8 +40,8 @@ def _get_pool() -> SOCKSProxyManager:
if _pool is None:
_pool = SOCKSProxyManager(
f"socks5h://{_PROXY_ADDR}:{_PROXY_PORT}/",
num_pools=20,
maxsize=4,
num_pools=30,
maxsize=8,
retries=_POOL_RETRIES,
)
return _pool
@@ -85,10 +85,46 @@ class _ProxyHandler(SocksiPyHandler, urllib.request.HTTPSHandler):
# -- Public HTTP interface ---------------------------------------------------
class _PooledResponse:
"""Thin wrapper around a preloaded urllib3 response.
Provides a ``read()`` that behaves like stdlib (returns full data
on first call, empty bytes on subsequent calls), plus ``close()``
as a no-op. Preloading ensures the underlying connection returns
to the pool immediately.
"""
__slots__ = ("status", "headers", "reason", "_data", "_pos")
def __init__(self, resp):
self.status = resp.status
self.headers = resp.headers
self.reason = resp.reason
self._data = resp.data # already fully read (preloaded)
self._pos = 0
def read(self, amt=None):
if self._pos >= len(self._data):
return b""
if amt is None:
chunk = self._data[self._pos:]
self._pos = len(self._data)
else:
chunk = self._data[self._pos:self._pos + amt]
self._pos += len(chunk)
return chunk
def close(self):
pass
def urlopen(req, *, timeout=None, context=None, retries=None, proxy=True):
"""HTTP urlopen with optional SOCKS5 proxy.
Uses connection pooling via urllib3 for proxied requests.
Uses connection pooling via urllib3 for proxied requests. Responses
are preloaded so the SOCKS connection returns to the pool immediately
(avoids opening 500+ fresh connections per session).
Falls back to legacy opener for custom SSL context.
When ``proxy=False``, uses stdlib ``urllib.request.urlopen`` directly.
Retries on transient SSL/connection errors with exponential backoff.
@@ -123,17 +159,14 @@ def urlopen(req, *, timeout=None, context=None, retries=None, proxy=True):
headers=headers,
body=body,
timeout=to,
preload_content=False,
preload_content=True,
)
if resp.status >= 400:
# Drain body so connection returns to pool, then raise
# urllib.error.HTTPError for backward compatibility.
resp.read()
raise urllib.error.HTTPError(
url, resp.status, resp.reason or "",
resp.headers, None,
)
return resp
return _PooledResponse(resp)
except urllib.error.HTTPError:
raise
except _RETRY_ERRORS as exc:
+295 -32
View File
@@ -45,9 +45,13 @@ def _strip_html(text: str) -> str:
return html.unescape(_TAG_RE.sub("", text))
_URL_RE = re.compile(r'(https?://[^\s<>&]+)')
def _escape_html(text: str) -> str:
"""Escape text for Mumble HTML messages."""
return html.escape(text, quote=False)
"""Escape text for Mumble HTML messages, auto-linking URLs."""
escaped = html.escape(text, quote=False)
return _URL_RE.sub(r'<a href="\1">\1</a>', escaped)
def _shell_quote(s: str) -> str:
@@ -160,6 +164,21 @@ class MumbleBot:
self._last_voice_ts: float = 0.0
self._connect_count: int = 0
self._sound_listeners: list = []
self._receive_sound: bool = mu_cfg.get("receive_sound", True)
self._self_mute: bool = mu_cfg.get("self_mute", False)
self._self_deaf: bool = mu_cfg.get("self_deaf", False)
self._mute_task: asyncio.Task | None = None
self._only_plugins: set[str] | None = (
set(mu_cfg["only_plugins"]) if "only_plugins" in mu_cfg else None
)
self._except_plugins: set[str] | None = (
set(mu_cfg["except_plugins"]) if "except_plugins" in mu_cfg else None
)
# Register in shared bot index so plugins can find peers
if not hasattr(registry, "_bots"):
registry._bots = {}
registry._bots[self._username] = self
rate_cfg = config.get("bot", {})
self._bucket = _TokenBucket(
@@ -198,7 +217,7 @@ class MumbleBot:
PYMUMBLE_CLBK_SOUNDRECEIVED,
self._on_sound_received,
)
self._mumble.set_receive_sound(True)
self._mumble.set_receive_sound(self._receive_sound)
self._mumble.start()
self._mumble.is_ready()
@@ -209,14 +228,36 @@ class MumbleBot:
session = getattr(self._mumble.users, "myself_session", "?")
log.info("mumble: %s as %s on %s:%d (session=%s)",
kind, self._username, self._host, self._port, session)
if self._self_mute:
try:
self._mumble.users.myself.mute()
except Exception:
log.exception("mumble: failed to self-mute on connect")
if self._self_deaf:
try:
self._mumble.users.myself.deafen()
except Exception:
log.exception("mumble: failed to self-deafen on connect")
if self._loop:
asyncio.run_coroutine_threadsafe(
self._notify_plugins_connected(), self._loop,
)
async def _notify_plugins_connected(self) -> None:
"""Call on_connected(bot) in each loaded plugin that defines it."""
"""Call on_connected(bot) in each loaded plugin that defines it.
Respects ``only_plugins`` / ``except_plugins`` so lifecycle hooks
only fire for plugins this bot is allowed to handle.
After plugin hooks, checks for a ``greet`` config on the connecting
bot. If present and this is the first connection, the greeting is
spoken through the voice-capable peer (the bot whose ``only_plugins``
includes ``voice``), so that a non-speaking bot like merlin can
still have an audible entrance announced by derp.
"""
for name, mod in self.registry._modules.items():
if not self._plugin_allowed(name, None):
continue
fn = getattr(mod, "on_connected", None)
if fn is None or not asyncio.iscoroutinefunction(fn):
continue
@@ -224,6 +265,22 @@ class MumbleBot:
await fn(self)
except Exception:
log.exception("mumble: on_connected hook failed in %s", name)
await self._play_greet()
async def _play_greet(self) -> None:
"""Speak the greeting via TTS on connect (voice only, no text)."""
greet = self.config.get("mumble", {}).get("greet")
if not greet:
return
voice_mod = self.registry._modules.get("voice")
tts_play = getattr(voice_mod, "_tts_play", None) if voice_mod else None
if tts_play is None:
return
for _ in range(20):
if self._is_audio_ready():
break
await asyncio.sleep(0.5)
self._spawn(tts_play(self, greet), name="voice-greet")
def _on_disconnected(self) -> None:
"""Callback from pymumble thread: connection lost."""
@@ -233,15 +290,18 @@ class MumbleBot:
def _on_sound_received(self, user, sound_chunk) -> None:
"""Callback from pymumble thread: voice audio received.
Updates the timestamp used by the music plugin's duck monitor.
When this callback is registered, pymumble passes decoded PCM
directly and does not queue it -- no memory buildup.
Ignores audio from our own bots entirely -- prevents self-ducking
and avoids STT transcribing bot TTS/music.
"""
name = user["name"] if isinstance(user, dict) else None
bots = getattr(self.registry, "_bots", {})
if name and name in bots:
return
prev = self._last_voice_ts
self._last_voice_ts = time.monotonic()
self.registry._voice_ts = self._last_voice_ts
if prev == 0.0:
name = user["name"] if isinstance(user, dict) else "?"
log.info("mumble: first voice packet from %s", name)
log.info("mumble: first voice packet from %s", name or "?")
for fn in self._sound_listeners:
try:
fn(user, sound_chunk)
@@ -263,6 +323,8 @@ class MumbleBot:
"""Process a text message from pymumble (runs on asyncio loop)."""
text = _strip_html(pb_msg.message)
actor = pb_msg.actor
log.debug("mumble: [%s] text from actor %s: %s",
self._username, actor, text[:100])
# Look up sender username
nick = None
@@ -291,6 +353,13 @@ class MumbleBot:
is_channel=is_channel,
params=[target or "", text],
)
# Check for direct addressing: "botname: command ..."
addressed = self._parse_addressed(text)
if addressed is not None:
await self._dispatch_addressed(msg, addressed)
return
await self._dispatch_command(msg)
# -- Lifecycle -----------------------------------------------------------
@@ -314,6 +383,60 @@ class MumbleBot:
self._mumble.stop()
self._mumble = None
# -- Direct addressing ---------------------------------------------------
def _parse_addressed(self, text: str) -> str | None:
"""Check if text is addressed to this bot: ``botname: rest``.
Returns the text after the address prefix, or None.
"""
name = self._username.lower()
lowered = text.lower()
for sep in (":", ",", " "):
prefix = name + sep
if lowered.startswith(prefix):
return text[len(prefix):].strip()
return None
def _find_voice_peer(self):
"""Find the voice-capable bot (the one with 'voice' in only_plugins)."""
bots = getattr(self.registry, "_bots", {})
for name, bot in bots.items():
if name == self._username:
continue
if bot._only_plugins and "voice" in bot._only_plugins:
return bot
return None
async def _dispatch_addressed(self, msg: MumbleMessage, text: str) -> None:
"""Handle a message directly addressed to this bot.
Supports a small set of built-in commands that don't use the
``!prefix`` convention. Currently: ``say <text>``.
TTS playback is routed through the voice-capable peer (e.g.
derp) so audio comes from the music bot's connection.
"""
parts = text.split(None, 1)
if not parts:
return
sub = parts[0].lower()
arg = parts[1] if len(parts) > 1 else ""
log.info("mumble: [%s] addressed command: %s (arg=%s)",
self._username, sub, arg[:80])
if sub == "say" and arg:
voice_mod = self.registry._modules.get("voice")
tts_play = getattr(voice_mod, "_tts_play", None) if voice_mod else None
if tts_play is None:
await self.reply(msg, "Voice not available")
return
# Route audio through the voice-capable peer
speaker = self._find_voice_peer() or self
speaker._spawn(tts_play(speaker, arg), name="addressed-say")
# Extend with elif for future addressed commands
# -- Command dispatch ----------------------------------------------------
async def _dispatch_command(self, msg: MumbleMessage) -> None:
@@ -357,12 +480,17 @@ class MumbleBot:
log.exception("mumble: error in command handler '%s'", cmd_name)
def _resolve_command(self, name: str):
"""Resolve command name with unambiguous prefix matching."""
"""Resolve command name with unambiguous prefix matching.
Only considers commands from plugins this bot is allowed to handle,
so filtered-out plugins never trigger ambiguity or dispatch.
"""
handler = self.registry.commands.get(name)
if handler is not None:
if handler is not None and self._plugin_allowed(handler.plugin, None):
return handler
matches = [v for k, v in self.registry.commands.items()
if k.startswith(name)]
if k.startswith(name)
and self._plugin_allowed(v.plugin, None)]
if len(matches) == 1:
return matches[0]
if len(matches) > 1:
@@ -370,7 +498,11 @@ class MumbleBot:
return None
def _plugin_allowed(self, plugin_name: str, channel: str | None) -> bool:
"""Channel filtering is IRC-only; all plugins are allowed on Mumble."""
"""Check if this bot handles commands from the given plugin."""
if self._only_plugins is not None:
return plugin_name in self._only_plugins
if self._except_plugins is not None:
return plugin_name not in self._except_plugins
return True
# -- Permission tiers ----------------------------------------------------
@@ -522,6 +654,8 @@ class MumbleBot:
seek: float = 0.0,
progress: list | None = None,
fade_step=None,
fade_in: float | bool = False,
seek_req: list | None = None,
) -> None:
"""Stream audio from URL through yt-dlp|ffmpeg to voice channel.
@@ -538,22 +672,40 @@ class MumbleBot:
current frame count each frame. ``fade_step`` is an optional
callable returning a float or None; when non-None it overrides
the default ramp step for fast fades (e.g. skip/stop).
``fade_in`` controls the initial volume ramp: ``False``/``0`` =
no fade-in, ``True`` = 5.0s ramp, or a float for a custom
duration in seconds. ``seek_req`` is a mutable ``[None]`` list;
when ``seek_req[0]`` is set to a float, the stream swaps its
ffmpeg pipeline in-place (fade-out, swap, fade-in) without
cancelling the task.
"""
if self._mumble is None:
return
# Unmute before streaming if self_mute is enabled
if self._self_mute:
if self._mute_task and not self._mute_task.done():
self._mute_task.cancel()
self._mute_task = None
try:
self._mumble.users.myself.unmute()
except Exception:
pass
_get_vol = volume if callable(volume) else lambda: volume
log.info("stream_audio: starting pipeline for %s (vol=%.0f%%, seek=%.1fs)",
url, _get_vol() * 100, seek)
seek_flag = f" -ss {seek:.3f}" if seek > 0 else ""
if os.path.isfile(url):
cmd = (f"ffmpeg{seek_flag} -i {_shell_quote(url)}"
f" -f s16le -ar 48000 -ac 1 -loglevel error pipe:1")
else:
cmd = (f"yt-dlp -o - -f bestaudio --no-warnings {_shell_quote(url)}"
f" | ffmpeg{seek_flag} -i pipe:0 -f s16le -ar 48000 -ac 1"
f" -loglevel error pipe:1")
def _build_cmd(seek_pos):
seek_flag = f" -ss {seek_pos:.3f}" if seek_pos > 0 else ""
if os.path.isfile(url):
return (f"ffmpeg{seek_flag} -i {_shell_quote(url)}"
f" -f s16le -ar 48000 -ac 1 -loglevel error pipe:1")
return (f"yt-dlp -o - -f bestaudio --no-warnings {_shell_quote(url)}"
f" | ffmpeg{seek_flag} -i pipe:0 -f s16le -ar 48000 -ac 1"
f" -loglevel error pipe:1")
cmd = _build_cmd(seek)
proc = await asyncio.create_subprocess_exec(
"sh", "-c", cmd,
stdout=asyncio.subprocess.PIPE,
@@ -561,13 +713,71 @@ class MumbleBot:
)
_max_step = 0.005 # max volume change per frame (~4s full ramp)
_cur_vol = _get_vol()
# Normalize fade_in to a duration in seconds
if fade_in is True:
_fade_dur = 5.0
elif fade_in:
_fade_dur = float(fade_in)
else:
_fade_dur = 0.0
_fade_in_target = _get_vol()
_cur_vol = 0.0 if _fade_dur > 0 else _fade_in_target
_fade_in_total = int(_fade_dur / 0.02) if _fade_dur > 0 else 0
_fade_in_frames = _fade_in_total
_fade_in_step = (_fade_in_target / _fade_in_total) if _fade_in_total else 0
_was_feeding = True # track connected/disconnected transitions
# Seek state (in-stream pipeline swap)
_seek_fading = False
_seek_target = 0.0
_seek_fade_out = 0
_SEEK_FADE_FRAMES = 10 # 0.2s ramp-down
frames = 0
try:
while True:
# Seek: swap pipeline when fade-out complete
if _seek_fading and _seek_fade_out <= 0:
try:
if self._is_audio_ready():
self._mumble.sound_output.clear_buffer()
except Exception:
pass
try:
proc.kill()
except ProcessLookupError:
pass
try:
await asyncio.wait_for(proc.stderr.read(), timeout=3)
await asyncio.wait_for(proc.wait(), timeout=3)
except (asyncio.TimeoutError, asyncio.CancelledError):
pass
cmd = _build_cmd(_seek_target)
proc = await asyncio.create_subprocess_exec(
"sh", "-c", cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
frames = 0
if progress is not None:
progress[0] = 0
seek = _seek_target
_fade_in_total = 25 # 0.5s fade-in
_fade_in_frames = _fade_in_total
_fade_in_target = _get_vol()
_fade_in_step = (
(_fade_in_target / _fade_in_total)
if _fade_in_total else 0
)
_cur_vol = 0.0
_seek_fading = False
log.info("stream_audio: seek to %.1fs", _seek_target)
continue
pcm = await proc.stdout.read(_FRAME_BYTES)
if not pcm and _seek_fading:
_seek_fade_out = 0
continue
if not pcm:
break
if len(pcm) < _FRAME_BYTES:
@@ -591,19 +801,54 @@ class MumbleBot:
"resuming feed at frame %d", frames)
_was_feeding = True
# Seek: fade-out in progress
if _seek_fading:
if (seek_req is not None
and seek_req[0] is not None
and seek_req[0] != _seek_target):
_seek_target = seek_req[0]
seek_req[0] = None
fade_ratio = _seek_fade_out / _SEEK_FADE_FRAMES
pcm = _scale_pcm(pcm, _cur_vol * fade_ratio)
try:
self._mumble.sound_output.add_sound(pcm)
except (TypeError, AttributeError, OSError):
pass
_seek_fade_out -= 1
try:
while (self._is_audio_ready()
and self._mumble.sound_output.get_buffer_size() > 1.0):
await asyncio.sleep(0.05)
except (TypeError, AttributeError):
pass
continue
# Seek: check for new request
if seek_req is not None and seek_req[0] is not None:
_seek_target = seek_req[0]
seek_req[0] = None
_seek_fading = True
_seek_fade_out = _SEEK_FADE_FRAMES
log.info("stream_audio: seek to %.1fs, fading out",
_seek_target)
target = _get_vol()
step = _max_step
if fade_step is not None:
if _fade_in_frames > 0:
step = _fade_in_step
_fade_in_frames -= 1
elif fade_step is not None:
fs = fade_step()
if fs:
step = fs
if _cur_vol == target:
# Fast path: flat scaling
diff = target - _cur_vol
if abs(diff) < 0.0001:
# Close enough -- flat scaling (no ramp artifacts)
if target != 1.0:
pcm = _scale_pcm(pcm, target)
_cur_vol = target
else:
# Ramp toward target, clamped to step per frame
diff = target - _cur_vol
if abs(diff) <= step:
next_vol = target
elif diff > 0:
@@ -640,12 +885,17 @@ class MumbleBot:
pass
log.info("stream_audio: finished, %d frames", frames)
except asyncio.CancelledError:
try:
if self._is_audio_ready():
self._mumble.sound_output.clear_buffer()
except Exception:
pass
log.info("stream_audio: cancelled at frame %d", frames)
# Only clear the buffer if volume is still audible -- if a
# fade-out has already driven _cur_vol to ~0 the remaining
# frames are silent and clearing mid-drain causes a click.
if _cur_vol > 0.01:
try:
if self._is_audio_ready():
self._mumble.sound_output.clear_buffer()
except Exception:
pass
log.info("stream_audio: cancelled at frame %d (vol=%.3f)",
frames, _cur_vol)
raise
except Exception:
log.exception("stream_audio: error at frame %d", frames)
@@ -665,6 +915,19 @@ class MumbleBot:
stderr_out.decode(errors="replace")[:500])
if on_done is not None:
on_done.set()
# Re-mute after audio finishes
if self._self_mute:
self._mute_task = self._spawn(
self._delayed_mute(3.0), name="self-mute",
)
async def _delayed_mute(self, delay: float) -> None:
"""Re-mute after a delay (lets the audio buffer drain fully)."""
await asyncio.sleep(delay)
try:
self._mumble.users.myself.mute()
except Exception:
pass
async def shorten_url(self, url: str) -> str:
"""Shorten a URL via FlaskPaste. Returns original on failure."""
+23 -5
View File
@@ -27,7 +27,13 @@ class Handler:
tier: str = "user"
def command(name: str, help: str = "", admin: bool = False, tier: str = "") -> Callable:
def command(
name: str,
help: str = "",
admin: bool = False,
tier: str = "",
aliases: list[str] | None = None,
) -> Callable:
"""Decorator to register an async function as a bot command.
Usage::
@@ -40,8 +46,8 @@ def command(name: str, help: str = "", admin: bool = False, tier: str = "") -> C
async def cmd_reload(bot, message):
...
@command("trusted_cmd", help="Trusted-only", tier="trusted")
async def cmd_trusted(bot, message):
@command("skip", help="Skip track", aliases=["next"])
async def cmd_skip(bot, message):
...
"""
@@ -50,6 +56,7 @@ def command(name: str, help: str = "", admin: bool = False, tier: str = "") -> C
func._derp_help = help # type: ignore[attr-defined]
func._derp_admin = admin # type: ignore[attr-defined]
func._derp_tier = tier if tier else ("admin" if admin else "user") # type: ignore[attr-defined]
func._derp_aliases = aliases or [] # type: ignore[attr-defined]
return func
return decorator
@@ -107,14 +114,25 @@ class PluginRegistry:
count = 0
for _name, obj in inspect.getmembers(module, inspect.isfunction):
if hasattr(obj, "_derp_command"):
cmd_tier = getattr(obj, "_derp_tier", "user")
cmd_admin = getattr(obj, "_derp_admin", False)
self.register_command(
obj._derp_command, obj,
help=getattr(obj, "_derp_help", ""),
plugin=plugin_name,
admin=getattr(obj, "_derp_admin", False),
tier=getattr(obj, "_derp_tier", "user"),
admin=cmd_admin,
tier=cmd_tier,
)
count += 1
for alias in getattr(obj, "_derp_aliases", []):
self.register_command(
alias, obj,
help=f"alias for !{obj._derp_command}",
plugin=plugin_name,
admin=cmd_admin,
tier=cmd_tier,
)
count += 1
if hasattr(obj, "_derp_event"):
self.register_event(obj._derp_event, obj, plugin=plugin_name)
count += 1
+212
View File
@@ -0,0 +1,212 @@
"""Tests for the alias plugin."""
import asyncio
import importlib.util
import sys
from derp.plugin import PluginRegistry
# -- Load plugin module directly ---------------------------------------------
_spec = importlib.util.spec_from_file_location("alias", "plugins/alias.py")
_mod = importlib.util.module_from_spec(_spec)
sys.modules["alias"] = _mod
_spec.loader.exec_module(_mod)
# -- Fakes -------------------------------------------------------------------
class _FakeState:
def __init__(self):
self._store: dict[str, dict[str, str]] = {}
def get(self, ns: str, key: str) -> str | None:
return self._store.get(ns, {}).get(key)
def set(self, ns: str, key: str, value: str) -> None:
self._store.setdefault(ns, {})[key] = value
def delete(self, ns: str, key: str) -> bool:
if ns in self._store and key in self._store[ns]:
del self._store[ns][key]
return True
return False
def keys(self, ns: str) -> list[str]:
return list(self._store.get(ns, {}).keys())
def clear(self, ns: str) -> int:
count = len(self._store.get(ns, {}))
self._store.pop(ns, None)
return count
class _FakeBot:
def __init__(self, *, admin: bool = False):
self.replied: list[str] = []
self.state = _FakeState()
self.registry = PluginRegistry()
self._admin = admin
async def reply(self, message, text: str) -> None:
self.replied.append(text)
def _is_admin(self, message) -> bool:
return self._admin
class _Msg:
def __init__(self, text="!alias"):
self.text = text
self.nick = "Alice"
self.target = "#test"
self.is_channel = True
self.prefix = "Alice!~alice@host"
# ---------------------------------------------------------------------------
# TestAliasAdd
# ---------------------------------------------------------------------------
class TestAliasAdd:
def test_add_creates_alias(self):
bot = _FakeBot()
# Register a target command
async def _noop(b, m): pass
bot.registry.register_command("skip", _noop, plugin="music")
msg = _Msg(text="!alias add s skip")
asyncio.run(_mod.cmd_alias(bot, msg))
assert bot.state.get("alias", "s") == "skip"
assert any("s -> skip" in r for r in bot.replied)
def test_add_rejects_existing_command(self):
bot = _FakeBot()
async def _noop(b, m): pass
bot.registry.register_command("skip", _noop, plugin="music")
msg = _Msg(text="!alias add skip stop")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("already a registered command" in r for r in bot.replied)
assert bot.state.get("alias", "skip") is None
def test_add_rejects_chaining(self):
bot = _FakeBot()
async def _noop(b, m): pass
bot.registry.register_command("skip", _noop, plugin="music")
bot.state.set("alias", "sk", "skip")
msg = _Msg(text="!alias add x sk")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("no chaining" in r for r in bot.replied)
def test_add_rejects_unknown_target(self):
bot = _FakeBot()
msg = _Msg(text="!alias add s nonexistent")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("unknown command" in r for r in bot.replied)
def test_add_lowercases_name(self):
bot = _FakeBot()
async def _noop(b, m): pass
bot.registry.register_command("skip", _noop, plugin="music")
msg = _Msg(text="!alias add S skip")
asyncio.run(_mod.cmd_alias(bot, msg))
assert bot.state.get("alias", "s") == "skip"
def test_add_missing_args(self):
bot = _FakeBot()
msg = _Msg(text="!alias add s")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("Usage" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestAliasDel
# ---------------------------------------------------------------------------
class TestAliasDel:
def test_del_removes_alias(self):
bot = _FakeBot()
bot.state.set("alias", "s", "skip")
msg = _Msg(text="!alias del s")
asyncio.run(_mod.cmd_alias(bot, msg))
assert bot.state.get("alias", "s") is None
assert any("removed" in r for r in bot.replied)
def test_del_nonexistent(self):
bot = _FakeBot()
msg = _Msg(text="!alias del x")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("no alias" in r for r in bot.replied)
def test_del_missing_name(self):
bot = _FakeBot()
msg = _Msg(text="!alias del")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("Usage" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestAliasList
# ---------------------------------------------------------------------------
class TestAliasList:
def test_list_empty(self):
bot = _FakeBot()
msg = _Msg(text="!alias list")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("No aliases" in r for r in bot.replied)
def test_list_shows_entries(self):
bot = _FakeBot()
bot.state.set("alias", "s", "skip")
bot.state.set("alias", "np", "nowplaying")
msg = _Msg(text="!alias list")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("s -> skip" in r for r in bot.replied)
assert any("np -> nowplaying" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestAliasClear
# ---------------------------------------------------------------------------
class TestAliasClear:
def test_clear_as_admin(self):
bot = _FakeBot(admin=True)
bot.state.set("alias", "s", "skip")
bot.state.set("alias", "np", "nowplaying")
msg = _Msg(text="!alias clear")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("Cleared 2" in r for r in bot.replied)
assert bot.state.keys("alias") == []
def test_clear_denied_non_admin(self):
bot = _FakeBot(admin=False)
bot.state.set("alias", "s", "skip")
msg = _Msg(text="!alias clear")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("Permission denied" in r for r in bot.replied)
assert bot.state.get("alias", "s") == "skip"
# ---------------------------------------------------------------------------
# TestAliasUsage
# ---------------------------------------------------------------------------
class TestAliasUsage:
def test_no_subcommand(self):
bot = _FakeBot()
msg = _Msg(text="!alias")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("Usage" in r for r in bot.replied)
def test_unknown_subcommand(self):
bot = _FakeBot()
msg = _Msg(text="!alias foo")
asyncio.run(_mod.cmd_alias(bot, msg))
assert any("Usage" in r for r in bot.replied)
+92
View File
@@ -0,0 +1,92 @@
"""Tests for the core plugin."""
import asyncio
import importlib.util
import sys
from unittest.mock import MagicMock
# -- Load plugin module directly ---------------------------------------------
_spec = importlib.util.spec_from_file_location("core", "plugins/core.py")
_mod = importlib.util.module_from_spec(_spec)
sys.modules["core"] = _mod
_spec.loader.exec_module(_mod)
# -- Fakes -------------------------------------------------------------------
class _FakeRegistry:
def __init__(self):
self._bots: dict = {}
class _FakeBot:
def __init__(self, *, mumble: bool = False):
self.replied: list[str] = []
self.registry = _FakeRegistry()
self.nick = "derp"
self._receive_sound = False
if mumble:
self._mumble = MagicMock()
async def reply(self, message, text: str) -> None:
self.replied.append(text)
def _make_listener():
"""Create a fake listener bot (merlin) with _receive_sound=True."""
listener = _FakeBot(mumble=True)
listener.nick = "merlin"
listener._receive_sound = True
return listener
class _Msg:
def __init__(self, text="!deaf"):
self.text = text
self.nick = "Alice"
self.target = "0"
self.is_channel = True
self.prefix = "Alice"
# -- Tests -------------------------------------------------------------------
class TestDeafCommand:
def test_deaf_targets_listener(self):
"""!deaf toggles the listener bot (merlin), not the calling bot."""
bot = _FakeBot(mumble=True)
listener = _make_listener()
bot.registry._bots = {"derp": bot, "merlin": listener}
listener._mumble.users.myself.get.return_value = False
msg = _Msg(text="!deaf")
asyncio.run(_mod.cmd_deaf(bot, msg))
listener._mumble.users.myself.deafen.assert_called_once()
assert any("merlin" in r and "deafened" in r for r in bot.replied)
def test_deaf_toggle_off(self):
bot = _FakeBot(mumble=True)
listener = _make_listener()
bot.registry._bots = {"derp": bot, "merlin": listener}
listener._mumble.users.myself.get.return_value = True
msg = _Msg(text="!deaf")
asyncio.run(_mod.cmd_deaf(bot, msg))
listener._mumble.users.myself.undeafen.assert_called_once()
listener._mumble.users.myself.unmute.assert_called_once()
assert any("merlin" in r and "undeafened" in r for r in bot.replied)
def test_deaf_non_mumble_silent(self):
bot = _FakeBot(mumble=False)
msg = _Msg(text="!deaf")
asyncio.run(_mod.cmd_deaf(bot, msg))
assert bot.replied == []
def test_deaf_fallback_no_listener(self):
"""Falls back to calling bot when no listener is registered."""
bot = _FakeBot(mumble=True)
bot._mumble.users.myself.get.return_value = False
msg = _Msg(text="!deaf")
asyncio.run(_mod.cmd_deaf(bot, msg))
bot._mumble.users.myself.deafen.assert_called_once()
+5 -1
View File
@@ -203,11 +203,15 @@ class TestUrlopen:
pool = MagicMock()
resp = MagicMock()
resp.status = 200
resp.data = b"ok"
resp.reason = "OK"
resp.headers = {}
pool.request.return_value = resp
mock_pool_fn.return_value = pool
result = urlopen("https://example.com/")
assert result is resp
assert result.status == 200
assert result.read() == b"ok"
@patch.object(derp.http, "_get_pool")
def test_context_falls_back_to_opener(self, mock_pool_fn):
+640 -65
View File
@@ -35,6 +35,13 @@ class _FakeState:
return list(self._store.get(ns, {}).keys())
class _FakeRegistry:
"""Minimal registry with shared voice timestamp."""
def __init__(self):
self._voice_ts: float = 0.0
class _FakeBot:
"""Minimal bot for music plugin testing."""
@@ -45,6 +52,7 @@ class _FakeBot:
self.config: dict = {}
self._pstate: dict = {}
self._tasks: set[asyncio.Task] = set()
self.registry = _FakeRegistry()
if mumble:
self.stream_audio = AsyncMock()
@@ -299,6 +307,19 @@ class TestNpCommand:
asyncio.run(_mod.cmd_np(bot, msg))
assert any("Cool Song" in r for r in bot.replied)
assert any("DJ" in r for r in bot.replied)
assert any("0:00" in r for r in bot.replied)
def test_np_shows_elapsed(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["current"] = _mod._Track(
url="x", title="Cool Song", requester="DJ",
)
ps["cur_seek"] = 60.0
ps["progress"] = [1500] # 1500 * 0.02 = 30s
msg = _Msg(text="!np")
asyncio.run(_mod.cmd_np(bot, msg))
assert any("1:30" in r for r in bot.replied)
# ---------------------------------------------------------------------------
@@ -527,6 +548,63 @@ class TestPlaylistExpansion:
assert tracks[0] == ("https://example.com/1", "First")
assert tracks[1] == ("https://example.com/2", "Second")
def test_resolve_tracks_preserves_playlist_url(self):
"""Video+playlist URL passes through to yt-dlp intact."""
result = MagicMock()
result.stdout = (
"https://youtube.com/watch?v=a\nFirst\n"
"https://youtube.com/watch?v=b\nSecond\n"
)
url = "https://www.youtube.com/watch?v=a&list=PLxyz&index=1"
with patch("subprocess.run", return_value=result) as mock_run:
tracks = _mod._resolve_tracks(url)
# URL must reach yt-dlp with &list= intact
called_url = mock_run.call_args[0][0][-1]
assert "list=PLxyz" in called_url
assert len(tracks) == 2
def test_random_fragment_shuffles(self):
"""#random fragment shuffles resolved playlist tracks."""
bot = _FakeBot()
msg = _Msg(text="!play https://example.com/playlist#random")
tracks = [(f"https://example.com/{i}", f"Track {i}") for i in range(20)]
with patch.object(_mod, "_resolve_tracks", return_value=list(tracks)) as mock_rt:
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_play(bot, msg))
# Fragment stripped before passing to resolver
called_url = mock_rt.call_args[0][0]
assert "#random" not in called_url
ps = _mod._ps(bot)
assert len(ps["queue"]) == 20
# Extremely unlikely (1/20!) that shuffle preserves exact order
titles = [t.title for t in ps["queue"]]
assert titles != [f"Track {i}" for i in range(20)] or len(titles) == 1
# Announces shuffle
assert any("shuffled" in r for r in bot.replied)
def test_random_fragment_single_track_no_error(self):
"""#random on a single-video URL works fine (nothing to shuffle)."""
bot = _FakeBot()
msg = _Msg(text="!play https://example.com/video#random")
tracks = [("https://example.com/video", "Solo Track")]
with patch.object(_mod, "_resolve_tracks", return_value=tracks):
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_play(bot, msg))
ps = _mod._ps(bot)
assert len(ps["queue"]) == 1
assert ps["queue"][0].title == "Solo Track"
def test_random_fragment_ignored_for_search(self):
"""#random is not treated specially for search queries."""
bot = _FakeBot()
msg = _Msg(text="!play jazz #random")
tracks = [("https://example.com/1", "Result")]
with patch.object(_mod, "_resolve_tracks", return_value=tracks) as mock_rt:
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_play(bot, msg))
# Search query passed as-is (not a URL, fragment not stripped)
assert mock_rt.call_args[0][0] == "ytsearch10:jazz #random"
def test_resolve_tracks_error_fallback(self):
"""On error, returns [(url, url)]."""
with patch("subprocess.run", side_effect=Exception("fail")):
@@ -541,6 +619,136 @@ class TestPlaylistExpansion:
tracks = _mod._resolve_tracks("https://example.com/empty")
assert tracks == [("https://example.com/empty", "https://example.com/empty")]
def test_resolve_tracks_start_param(self):
"""start= passes --playlist-start to yt-dlp."""
result = MagicMock()
result.stdout = "https://example.com/6\nTrack 6\n"
with patch("subprocess.run", return_value=result) as mock_run:
tracks = _mod._resolve_tracks("https://example.com/pl",
max_tracks=5, start=6)
cmd = mock_run.call_args[0][0]
assert "--playlist-start=6" in cmd
assert "--playlist-end=10" in cmd
assert tracks == [("https://example.com/6", "Track 6")]
def test_resolve_tracks_start_empty_returns_empty(self):
"""Paginated call with no results returns [] (not fallback)."""
result = MagicMock()
result.stdout = ""
with patch("subprocess.run", return_value=result):
tracks = _mod._resolve_tracks("https://example.com/pl",
start=100)
assert tracks == []
def test_resolve_tracks_start_error_returns_empty(self):
"""Paginated call on error returns [] (not fallback)."""
with patch("subprocess.run", side_effect=Exception("fail")):
tracks = _mod._resolve_tracks("https://example.com/pl",
start=10)
assert tracks == []
def test_playlist_url_triggers_batched_resolve(self):
"""Playlist URL resolves initial batch, spawns feeder for rest."""
bot = _FakeBot()
batch = _mod._PLAYLIST_BATCH
initial = [(f"https://example.com/{i}", f"T{i}")
for i in range(batch)]
spawned = []
orig_spawn = bot._spawn
def spy_spawn(coro, *, name=None):
spawned.append(name)
return orig_spawn(coro, name=name)
bot._spawn = spy_spawn
msg = _Msg(text="!play https://example.com/watch?v=a&list=PLxyz")
with patch.object(_mod, "_resolve_tracks", return_value=initial):
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_play(bot, msg))
ps = _mod._ps(bot)
assert len(ps["queue"]) == batch
assert "music-playlist-feeder" in spawned
assert any("resolving more" in r.lower() for r in bot.replied)
def test_non_playlist_url_no_feeder(self):
"""Single video URL does not spawn background feeder."""
bot = _FakeBot()
spawned = []
orig_spawn = bot._spawn
def spy_spawn(coro, *, name=None):
spawned.append(name)
return orig_spawn(coro, name=name)
bot._spawn = spy_spawn
tracks = [("https://example.com/v", "Video")]
msg = _Msg(text="!play https://example.com/v")
with patch.object(_mod, "_resolve_tracks", return_value=tracks):
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_play(bot, msg))
assert "music-playlist-feeder" not in spawned
def test_playlist_feeder_appends_to_queue(self):
"""Background feeder resolves remaining tracks into queue."""
bot = _FakeBot()
ps = _mod._ps(bot)
remaining = [("https://example.com/6", "Track 6"),
("https://example.com/7", "Track 7")]
async def _check():
with patch.object(_mod, "_resolve_tracks",
return_value=remaining):
await _mod._playlist_feeder(
bot, "https://example.com/pl", 6, 10,
False, "Alice", "https://example.com/pl",
)
assert len(ps["queue"]) == 2
assert ps["queue"][0].title == "Track 6"
assert ps["queue"][1].requester == "Alice"
asyncio.run(_check())
def test_playlist_feeder_shuffles(self):
"""Background feeder shuffles when shuffle=True."""
bot = _FakeBot()
ps = _mod._ps(bot)
remaining = [(f"https://example.com/{i}", f"T{i}")
for i in range(20)]
async def _check():
with patch.object(_mod, "_resolve_tracks",
return_value=list(remaining)):
await _mod._playlist_feeder(
bot, "https://example.com/pl", 6, 20,
True, "Alice", "",
)
titles = [t.title for t in ps["queue"]]
assert len(titles) == 20
# Extremely unlikely shuffle preserves order
assert titles != [f"T{i}" for i in range(20)]
asyncio.run(_check())
def test_playlist_feeder_respects_queue_cap(self):
"""Background feeder stops at _MAX_QUEUE."""
bot = _FakeBot()
ps = _mod._ps(bot)
# Pre-fill queue to near capacity
ps["queue"] = [_mod._Track(url="x", title="t", requester="a")
for _ in range(_mod._MAX_QUEUE - 2)]
remaining = [(f"https://example.com/{i}", f"T{i}")
for i in range(10)]
async def _check():
with patch.object(_mod, "_resolve_tracks",
return_value=remaining):
await _mod._playlist_feeder(
bot, "url", 6, 10, False, "a", "",
)
assert len(ps["queue"]) == _mod._MAX_QUEUE
asyncio.run(_check())
# ---------------------------------------------------------------------------
# TestResumeState
@@ -580,6 +788,56 @@ class TestResumeState:
bot.state.set("music", "resume", '{"title": "x"}')
assert _mod._load_resume(bot) is None
def test_save_strips_youtube_playlist_params(self):
"""_save_resume strips &list= and other playlist params from YouTube URLs."""
bot = _FakeBot()
track = _mod._Track(
url="https://www.youtube.com/watch?v=abc123&list=RDabc123&start_radio=1&pp=xyz",
title="Song", requester="Alice",
)
_mod._save_resume(bot, track, 60.0)
data = _mod._load_resume(bot)
assert data is not None
assert data["url"] == "https://www.youtube.com/watch?v=abc123"
def test_save_preserves_non_youtube_urls(self):
"""_save_resume leaves non-YouTube URLs unchanged."""
bot = _FakeBot()
track = _mod._Track(
url="https://soundcloud.com/artist/track?ref=playlist",
title="Song", requester="Alice",
)
_mod._save_resume(bot, track, 30.0)
data = _mod._load_resume(bot)
assert data["url"] == "https://soundcloud.com/artist/track?ref=playlist"
# ---------------------------------------------------------------------------
# TestStripPlaylistParams
# ---------------------------------------------------------------------------
class TestStripPlaylistParams:
def test_strips_list_param(self):
url = "https://www.youtube.com/watch?v=abc&list=PLxyz&index=3"
assert _mod._strip_playlist_params(url) == "https://www.youtube.com/watch?v=abc"
def test_strips_radio_params(self):
url = "https://www.youtube.com/watch?v=abc&list=RDabc&start_radio=1&pp=xyz"
assert _mod._strip_playlist_params(url) == "https://www.youtube.com/watch?v=abc"
def test_preserves_plain_url(self):
url = "https://www.youtube.com/watch?v=abc123"
assert _mod._strip_playlist_params(url) == "https://www.youtube.com/watch?v=abc123"
def test_non_youtube_unchanged(self):
url = "https://soundcloud.com/track?list=abc"
assert _mod._strip_playlist_params(url) == url
def test_youtu_be_without_v_param(self):
url = "https://youtu.be/abc123"
assert _mod._strip_playlist_params(url) == url
# ---------------------------------------------------------------------------
# TestResumeCommand
@@ -740,7 +998,7 @@ class TestDuckMonitor:
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_floor"] = 5
bot._last_voice_ts = time.monotonic()
bot.registry._voice_ts = time.monotonic()
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
@@ -760,7 +1018,7 @@ class TestDuckMonitor:
ps["duck_floor"] = 1
ps["duck_restore"] = 10 # 10s total restore
ps["volume"] = 50
bot._last_voice_ts = time.monotonic() - 100
bot.registry._voice_ts = time.monotonic() - 100
ps["duck_vol"] = 1.0 # already ducked
async def _check():
@@ -783,7 +1041,7 @@ class TestDuckMonitor:
ps["duck_floor"] = 1
ps["duck_restore"] = 1 # 1s restore -- completes quickly
ps["volume"] = 50
bot._last_voice_ts = time.monotonic() - 100
bot.registry._voice_ts = time.monotonic() - 100
ps["duck_vol"] = 1.0
async def _check():
@@ -805,14 +1063,14 @@ class TestDuckMonitor:
ps["duck_floor"] = 5
ps["duck_restore"] = 30
ps["volume"] = 50
bot._last_voice_ts = time.monotonic() - 100
bot.registry._voice_ts = time.monotonic() - 100
ps["duck_vol"] = 30.0 # mid-restore
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
await asyncio.sleep(0.5)
# Simulate voice arriving now
bot._last_voice_ts = time.monotonic()
bot.registry._voice_ts = time.monotonic()
await asyncio.sleep(1.5)
assert ps["duck_vol"] == 5.0 # re-ducked to floor
task.cancel()
@@ -826,7 +1084,7 @@ class TestDuckMonitor:
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = False
bot._last_voice_ts = time.monotonic()
bot.registry._voice_ts = time.monotonic()
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
@@ -839,6 +1097,56 @@ class TestDuckMonitor:
pass
asyncio.run(_check())
def test_tts_active_ducks(self):
"""TTS activity from voice peer triggers ducking."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_floor"] = 5
ps["duck_restore"] = 1 # fast restore for test
bot.registry._voice_ts = 0.0
bot.registry._tts_active = True
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
await asyncio.sleep(1.5)
assert ps["duck_vol"] == 5.0
# TTS ends -- restore should begin and complete quickly
bot.registry._tts_active = False
await asyncio.sleep(2.5)
assert ps["duck_vol"] is None
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_tts_active_overrides_all_muted(self):
"""TTS ducks even when all users are muted."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_floor"] = 5
bot.registry._voice_ts = time.monotonic()
bot.registry._tts_active = True
# Simulate all users muted
bot._mumble = MagicMock()
bot._mumble.users = {1: {"name": "human", "self_mute": True,
"mute": False, "self_deaf": False}}
bot.registry._bots = {}
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
await asyncio.sleep(1.5)
assert ps["duck_vol"] == 5.0
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
# ---------------------------------------------------------------------------
# TestAutoResume
@@ -850,7 +1158,7 @@ class TestAutoResume:
"""Auto-resume loads saved state when channel is silent."""
bot = _FakeBot()
bot._connect_count = 2
bot._last_voice_ts = 0.0
bot.registry._voice_ts = 0.0
track = _mod._Track(url="https://example.com/a", title="Song", requester="Alice")
_mod._save_resume(bot, track, 120.0)
@@ -878,7 +1186,7 @@ class TestAutoResume:
def test_no_resume_if_no_state(self):
"""Auto-resume returns early when nothing is saved."""
bot = _FakeBot()
bot._last_voice_ts = 0.0
bot.registry._voice_ts = 0.0
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod._auto_resume(bot))
mock_loop.assert_not_called()
@@ -887,7 +1195,7 @@ class TestAutoResume:
"""Auto-resume aborts if voice never goes silent within deadline."""
bot = _FakeBot()
now = time.monotonic()
bot._last_voice_ts = now
bot.registry._voice_ts = now
ps = _mod._ps(bot)
ps["duck_silence"] = 15
track = _mod._Track(url="https://example.com/a", title="Song", requester="Alice")
@@ -903,7 +1211,7 @@ class TestAutoResume:
async def _fast_sleep(s):
mono_val[0] += s
bot._last_voice_ts = mono_val[0]
bot.registry._voice_ts = mono_val[0]
await _real_sleep(0)
with patch.object(time, "monotonic", side_effect=_fast_mono):
@@ -918,6 +1226,9 @@ class TestAutoResume:
"""Watcher detects connect_count increment and calls _auto_resume."""
bot = _FakeBot()
bot._connect_count = 1
# Resume state must exist for watcher to call _auto_resume
track = _mod._Track(url="https://example.com/a", title="Song", requester="Alice")
_mod._save_resume(bot, track, 60.0)
async def _check():
with patch.object(_mod, "_auto_resume", new_callable=AsyncMock) as mock_ar:
@@ -1014,6 +1325,99 @@ class TestAutoResume:
assert spawned.count("music-reconnect-watcher") == 1
# ---------------------------------------------------------------------------
# TestAutoplayKept
# ---------------------------------------------------------------------------
class TestAutoplayKept:
def test_starts_loop_with_kept_tracks(self, tmp_path):
"""Autoplay starts play loop when kept tracks exist."""
bot = _FakeBot()
bot.registry._voice_ts = 0.0
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "a.opus").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/a", "title": "Track A",
"filename": "a.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod._autoplay_kept(bot))
mock_loop.assert_called_once_with(bot)
def test_skips_when_already_playing(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["current"] = _mod._Track(url="x", title="Playing", requester="a")
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod._autoplay_kept(bot))
mock_loop.assert_not_called()
def test_skips_when_no_kept_tracks(self):
bot = _FakeBot()
bot.registry._voice_ts = 0.0
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod._autoplay_kept(bot))
mock_loop.assert_not_called()
def test_load_kept_tracks_skips_missing_files(self, tmp_path):
"""Tracks with missing local files are excluded."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/a", "title": "Gone",
"filename": "missing.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
tracks = _mod._load_kept_tracks(bot)
assert tracks == []
def test_watcher_autoplay_on_boot_no_resume(self):
"""Watcher triggers autoplay on boot when no resume state exists."""
bot = _FakeBot()
bot._connect_count = 0
async def _check():
with patch.object(_mod, "_autoplay_kept",
new_callable=AsyncMock) as mock_ap:
task = asyncio.create_task(_mod._reconnect_watcher(bot))
await asyncio.sleep(0.5)
bot._connect_count = 1
await asyncio.sleep(3)
mock_ap.assert_called_once_with(bot)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_watcher_autoplay_on_reconnect_no_resume(self):
"""Watcher triggers autoplay on reconnect when no resume state."""
bot = _FakeBot()
bot._connect_count = 1
async def _check():
with patch.object(_mod, "_autoplay_kept",
new_callable=AsyncMock) as mock_ap:
task = asyncio.create_task(_mod._reconnect_watcher(bot))
await asyncio.sleep(0.5)
bot._connect_count = 2
await asyncio.sleep(3)
mock_ap.assert_called_once_with(bot)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
# ---------------------------------------------------------------------------
# TestDownloadTrack
# ---------------------------------------------------------------------------
@@ -1134,10 +1538,58 @@ class TestKeepCommand:
)
ps["current"] = track
msg = _Msg(text="!keep")
asyncio.run(_mod.cmd_keep(bot, msg))
music_dir = tmp_path / "kept"
music_dir.mkdir()
meta = {"title": "t", "artist": "", "duration": 0}
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_fetch_metadata", return_value=meta):
asyncio.run(_mod.cmd_keep(bot, msg))
assert track.keep is True
assert any("Keeping" in r for r in bot.replied)
def test_keep_duplicate_blocked(self, tmp_path):
bot = _FakeBot()
ps = _mod._ps(bot)
f = tmp_path / "abc123.opus"
f.write_bytes(b"audio")
track = _mod._Track(
url="https://example.com/v", title="t", requester="a",
local_path=f,
)
ps["current"] = track
# Pre-existing kept entry with same URL
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/v", "id": 1,
}))
bot.state.set("music", "keep_next_id", "2")
msg = _Msg(text="!keep")
asyncio.run(_mod.cmd_keep(bot, msg))
assert any("Already kept" in r for r in bot.replied)
assert any("#1" in r for r in bot.replied)
# ID counter should not have incremented
assert bot.state.get("music", "keep_next_id") == "2"
def test_keep_duplicate_with_playlist_params(self, tmp_path):
bot = _FakeBot()
ps = _mod._ps(bot)
f = tmp_path / "abc123.opus"
f.write_bytes(b"audio")
# Track URL has playlist cruft
track = _mod._Track(
url="https://www.youtube.com/watch?v=abc&list=RDabc&start_radio=1",
title="t", requester="a", local_path=f,
)
ps["current"] = track
# Existing entry stored with clean URL
bot.state.set("music", "keep:1", json.dumps({
"url": "https://www.youtube.com/watch?v=abc", "id": 1,
}))
bot.state.set("music", "keep_next_id", "2")
msg = _Msg(text="!keep")
asyncio.run(_mod.cmd_keep(bot, msg))
assert any("Already kept" in r for r in bot.replied)
assert bot.state.get("music", "keep_next_id") == "2"
def test_keep_non_mumble(self):
bot = _FakeBot(mumble=False)
msg = _Msg(text="!keep")
@@ -1151,23 +1603,27 @@ class TestKeepCommand:
class TestKeptCommand:
def test_kept_empty(self, tmp_path):
def test_kept_empty(self):
bot = _FakeBot()
with patch.object(_mod, "_MUSIC_DIR", tmp_path / "empty"):
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("No kept files" in r for r in bot.replied)
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("No kept tracks" in r for r in bot.replied)
def test_kept_lists_files(self, tmp_path):
def test_kept_lists_tracks(self, tmp_path):
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "abc123.opus").write_bytes(b"x" * 1024)
bot.state.set("music", "keep:1", json.dumps({
"title": "Test Track", "artist": "", "duration": 0,
"filename": "abc123.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Kept files" in r for r in bot.replied)
assert any("abc123.opus" in r for r in bot.replied)
assert any("Kept tracks" in r for r in bot.replied)
assert any("#1" in r for r in bot.replied)
assert any("Test Track" in r for r in bot.replied)
def test_kept_clear(self, tmp_path):
bot = _FakeBot()
@@ -1175,11 +1631,27 @@ class TestKeptCommand:
music_dir.mkdir()
(music_dir / "abc123.opus").write_bytes(b"audio")
(music_dir / "def456.webm").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({"id": 1}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept clear")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Deleted 2 file(s)" in r for r in bot.replied)
assert not list(music_dir.iterdir())
assert bot.state.get("music", "keep:1") is None
def test_kept_shows_missing_marker(self, tmp_path):
"""Tracks with missing files show [MISSING] in listing."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
bot.state.set("music", "keep:1", json.dumps({
"title": "Gone Track", "artist": "", "duration": 0,
"filename": "gone.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("MISSING" in r for r in bot.replied)
def test_kept_non_mumble(self):
bot = _FakeBot(mumble=False)
@@ -1256,50 +1728,43 @@ class TestSeekCommand:
def test_seek_absolute(self):
bot = _FakeBot()
ps = _mod._ps(bot)
track = _mod._Track(url="x", title="Song", requester="a")
ps["current"] = track
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
ps["current"] = _mod._Track(url="x", title="Song", requester="a")
ps["seek_req"] = [None]
ps["progress"] = [100]
msg = _Msg(text="!seek 1:30")
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod.cmd_seek(bot, msg))
mock_loop.assert_called_once_with(bot, seek=90.0)
assert ps["queue"][0] is track
asyncio.run(_mod.cmd_seek(bot, msg))
assert ps["seek_req"][0] == 90.0
assert ps["cur_seek"] == 90.0
assert ps["progress"][0] == 0
assert any("1:30" in r for r in bot.replied)
mock_task.cancel.assert_called_once()
def test_seek_relative_forward(self):
bot = _FakeBot()
ps = _mod._ps(bot)
track = _mod._Track(url="x", title="Song", requester="a")
ps["current"] = track
ps["current"] = _mod._Track(url="x", title="Song", requester="a")
ps["seek_req"] = [None]
ps["progress"] = [1500] # 1500 * 0.02 = 30s
ps["cur_seek"] = 60.0 # started at 60s
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!seek +30")
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod.cmd_seek(bot, msg))
# elapsed = 60 + 30 = 90, target = 90 + 30 = 120
mock_loop.assert_called_once_with(bot, seek=120.0)
asyncio.run(_mod.cmd_seek(bot, msg))
# elapsed = 60 + 30 = 90, target = 90 + 30 = 120
assert ps["seek_req"][0] == 120.0
assert ps["cur_seek"] == 120.0
assert ps["progress"][0] == 0
def test_seek_relative_backward_clamps(self):
bot = _FakeBot()
ps = _mod._ps(bot)
track = _mod._Track(url="x", title="Song", requester="a")
ps["current"] = track
ps["current"] = _mod._Track(url="x", title="Song", requester="a")
ps["seek_req"] = [None]
ps["progress"] = [500] # 500 * 0.02 = 10s
ps["cur_seek"] = 0.0
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!seek -30")
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod.cmd_seek(bot, msg))
# elapsed = 0 + 10 = 10, target = 10 - 30 = -20, clamped to 0
mock_loop.assert_called_once_with(bot, seek=0.0)
asyncio.run(_mod.cmd_seek(bot, msg))
# elapsed = 0 + 10 = 10, target = 10 - 30 = -20, clamped to 0
assert ps["seek_req"][0] == 0.0
assert ps["cur_seek"] == 0.0
assert ps["progress"][0] == 0
# ---------------------------------------------------------------------------
@@ -1533,7 +1998,7 @@ class TestFadeState:
class TestKeepMetadata:
def test_keep_stores_metadata(self, tmp_path):
"""!keep stores metadata JSON in bot.state."""
"""!keep stores metadata JSON in bot.state keyed by ID."""
bot = _FakeBot()
ps = _mod._ps(bot)
f = tmp_path / "abc123.opus"
@@ -1545,15 +2010,19 @@ class TestKeepMetadata:
ps["current"] = track
msg = _Msg(text="!keep")
meta = {"title": "My Song", "artist": "Artist", "duration": 195.0}
with patch.object(_mod, "_fetch_metadata", return_value=meta):
music_dir = tmp_path / "kept"
music_dir.mkdir()
with patch.object(_mod, "_fetch_metadata", return_value=meta), \
patch.object(_mod, "_MUSIC_DIR", music_dir):
asyncio.run(_mod.cmd_keep(bot, msg))
assert track.keep is True
raw = bot.state.get("music", "keep:abc123.opus")
raw = bot.state.get("music", "keep:1")
assert raw is not None
stored = json.loads(raw)
assert stored["title"] == "My Song"
assert stored["artist"] == "Artist"
assert stored["duration"] == 195.0
assert stored["id"] == 1
assert any("My Song" in r for r in bot.replied)
assert any("Artist" in r for r in bot.replied)
assert any("3:15" in r for r in bot.replied)
@@ -1571,7 +2040,10 @@ class TestKeepMetadata:
ps["current"] = track
msg = _Msg(text="!keep")
meta = {"title": "Song", "artist": "NA", "duration": 60.0}
with patch.object(_mod, "_fetch_metadata", return_value=meta):
music_dir = tmp_path / "kept"
music_dir.mkdir()
with patch.object(_mod, "_fetch_metadata", return_value=meta), \
patch.object(_mod, "_MUSIC_DIR", music_dir):
asyncio.run(_mod.cmd_keep(bot, msg))
# Should not contain "NA" as artist
assert not any("NA" in r and "--" in r for r in bot.replied)
@@ -1584,13 +2056,14 @@ class TestKeepMetadata:
class TestKeptMetadata:
def test_kept_shows_metadata(self, tmp_path):
"""!kept displays metadata from bot.state when available."""
"""!kept displays metadata from bot.state."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "abc123.opus").write_bytes(b"x" * 2048)
bot.state.set("music", "keep:abc123.opus", json.dumps({
bot.state.set("music", "keep:1", json.dumps({
"title": "Cool Song", "artist": "DJ Test", "duration": 225.0,
"filename": "abc123.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept")
@@ -1598,31 +2071,34 @@ class TestKeptMetadata:
assert any("Cool Song" in r for r in bot.replied)
assert any("DJ Test" in r for r in bot.replied)
assert any("3:45" in r for r in bot.replied)
assert any("#1" in r for r in bot.replied)
def test_kept_fallback_no_metadata(self, tmp_path):
"""!kept falls back to filename when no metadata stored."""
def test_kept_fallback_no_title(self):
"""!kept falls back to filename when no title in metadata."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "xyz789.webm").write_bytes(b"x" * 1024)
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
bot.state.set("music", "keep:1", json.dumps({
"title": "", "artist": "", "duration": 0,
"filename": "xyz789.webm", "id": 1,
}))
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("xyz789.webm" in r for r in bot.replied)
def test_kept_clear_removes_metadata(self, tmp_path):
"""!kept clear also removes stored metadata."""
"""!kept clear also removes stored metadata and resets ID."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "abc123.opus").write_bytes(b"audio")
bot.state.set("music", "keep:abc123.opus", json.dumps({
"title": "Song", "artist": "", "duration": 0,
bot.state.set("music", "keep:1", json.dumps({
"title": "Song", "artist": "", "duration": 0, "id": 1,
}))
bot.state.set("music", "keep_next_id", "2")
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept clear")
asyncio.run(_mod.cmd_kept(bot, msg))
assert bot.state.get("music", "keep:abc123.opus") is None
assert bot.state.get("music", "keep:1") is None
assert bot.state.get("music", "keep_next_id") is None
assert any("Deleted 1 file(s)" in r for r in bot.replied)
@@ -1656,3 +2132,102 @@ class TestFetchMetadata:
assert meta["title"] == ""
assert meta["artist"] == ""
assert meta["duration"] == 0
# ---------------------------------------------------------------------------
# TestKeptRepair
# ---------------------------------------------------------------------------
class TestKeptRepair:
def test_repair_nothing_missing(self, tmp_path):
"""Repair reports all present when files exist."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "song.opus").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/v", "title": "Song",
"filename": "song.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept repair")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("nothing to repair" in r.lower() for r in bot.replied)
def test_repair_downloads_missing(self, tmp_path):
"""Repair re-downloads missing files."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/v", "title": "Song",
"filename": "song.opus", "id": 1,
}))
dl_path = tmp_path / "cache" / "dl.opus"
dl_path.parent.mkdir()
dl_path.write_bytes(b"audio")
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_download_track", return_value=dl_path):
msg = _Msg(text="!kept repair")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("1 restored" in r for r in bot.replied)
assert (music_dir / "song.opus").is_file()
def test_repair_counts_failures(self, tmp_path):
"""Repair reports failed downloads."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/v", "title": "Song",
"filename": "song.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_download_track", return_value=None):
msg = _Msg(text="!kept repair")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("1 failed" in r for r in bot.replied)
def test_repair_no_url_skips(self, tmp_path):
"""Repair skips entries with no URL."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
bot.state.set("music", "keep:1", json.dumps({
"url": "", "title": "No URL",
"filename": "nourl.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept repair")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("1 failed" in r for r in bot.replied)
def test_repair_extension_mismatch(self, tmp_path):
"""Repair updates metadata when download extension differs."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/v", "title": "Song",
"filename": "song.opus", "id": 1,
}))
dl_path = tmp_path / "cache" / "dl.webm"
dl_path.parent.mkdir()
dl_path.write_bytes(b"audio")
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_download_track", return_value=dl_path):
msg = _Msg(text="!kept repair")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("1 restored" in r for r in bot.replied)
# Filename updated to new extension
raw = bot.state.get("music", "keep:1")
stored = json.loads(raw)
assert stored["filename"] == "song.webm"
assert (music_dir / "song.webm").is_file()
+119
View File
@@ -36,6 +36,20 @@ class TestDecorators:
assert handler._derp_event == "PRIVMSG"
def test_command_decorator_aliases(self):
@command("skip", help="skip track", aliases=["next", "s"])
async def handler(bot, msg):
pass
assert handler._derp_aliases == ["next", "s"]
def test_command_decorator_aliases_default(self):
@command("ping", help="ping")
async def handler(bot, msg):
pass
assert handler._derp_aliases == []
def test_command_decorator_admin(self):
@command("secret", help="admin only", admin=True)
async def handler(bot, msg):
@@ -208,6 +222,46 @@ class TestRegistry:
assert registry.commands["secret"].admin is True
assert registry.commands["public"].admin is False
def test_load_plugin_aliases(self, tmp_path: Path):
plugin_file = tmp_path / "aliased.py"
plugin_file.write_text(textwrap.dedent("""\
from derp.plugin import command
@command("skip", help="Skip track", aliases=["next", "s"])
async def cmd_skip(bot, msg):
pass
"""))
registry = PluginRegistry()
count = registry.load_plugin(plugin_file)
assert count == 3 # primary + 2 aliases
assert "skip" in registry.commands
assert "next" in registry.commands
assert "s" in registry.commands
# Aliases point to the same callback
assert registry.commands["next"].callback is registry.commands["skip"].callback
assert registry.commands["s"].callback is registry.commands["skip"].callback
# Alias help text references the primary command
assert registry.commands["next"].help == "alias for !skip"
def test_unload_removes_aliases(self, tmp_path: Path):
plugin_file = tmp_path / "aliased.py"
plugin_file.write_text(textwrap.dedent("""\
from derp.plugin import command
@command("skip", help="Skip track", aliases=["next"])
async def cmd_skip(bot, msg):
pass
"""))
registry = PluginRegistry()
registry.load_plugin(plugin_file)
assert "next" in registry.commands
registry.unload_plugin("aliased")
assert "skip" not in registry.commands
assert "next" not in registry.commands
def test_load_plugin_stores_path(self, tmp_path: Path):
plugin_file = tmp_path / "pathed.py"
plugin_file.write_text(textwrap.dedent("""\
@@ -677,6 +731,71 @@ class TestChannelFilter:
assert bot._plugin_allowed("encode", "&local") is False
class TestAliasDispatch:
"""Test alias fallback in _dispatch_command."""
@staticmethod
def _make_bot_with_alias(alias_name: str, target_cmd: str) -> tuple[Bot, list]:
"""Create a Bot with a command and an alias pointing to it."""
config = {
"server": {"host": "localhost", "port": 6667, "tls": False,
"nick": "test", "user": "test", "realname": "test"},
"bot": {"prefix": "!", "channels": [], "plugins_dir": "plugins"},
}
registry = PluginRegistry()
called = []
async def _handler(bot, msg):
called.append(msg.text)
registry.register_command(target_cmd, _handler, plugin="test")
bot = Bot("test", config, registry)
bot.conn = _FakeConnection()
bot.state.set("alias", alias_name, target_cmd)
return bot, called
def test_alias_resolves_command(self):
"""An alias triggers the target command handler."""
bot, called = self._make_bot_with_alias("s", "skip")
msg = Message(raw="", prefix="nick!u@h", nick="nick",
command="PRIVMSG", params=["#ch", "!s"], tags={})
async def _run():
bot._dispatch_command(msg)
await asyncio.sleep(0.05) # let spawned task run
asyncio.run(_run())
assert len(called) == 1
def test_alias_ignored_when_command_exists(self):
"""Direct command match takes priority over alias."""
bot, called = self._make_bot_with_alias("skip", "stop")
# "skip" is both a real command and an alias to "stop"; real wins
msg = Message(raw="", prefix="nick!u@h", nick="nick",
command="PRIVMSG", params=["#ch", "!skip"], tags={})
async def _run():
bot._dispatch_command(msg)
await asyncio.sleep(0.05)
asyncio.run(_run())
assert len(called) == 1
# Handler was the "skip" handler, not "stop"
def test_no_alias_no_crash(self):
"""Unknown command with no alias silently returns."""
config = {
"server": {"host": "localhost", "port": 6667, "tls": False,
"nick": "test", "user": "test", "realname": "test"},
"bot": {"prefix": "!", "channels": [], "plugins_dir": "plugins"},
}
bot = Bot("test", config, PluginRegistry())
bot.conn = _FakeConnection()
msg = Message(raw="", prefix="nick!u@h", nick="nick",
command="PRIVMSG", params=["#ch", "!nonexistent"], tags={})
bot._dispatch_command(msg) # should not raise
class TestSplitUtf8:
"""Test UTF-8 safe message splitting."""
+38
View File
@@ -0,0 +1,38 @@
#!/usr/bin/env bash
# Shared helpers for derp container tools.
# Sourced, not executed.
# shellcheck disable=SC2034
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[1]}")" && pwd)"
PROJECT_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
# Compose command detection
if podman compose version &>/dev/null; then
COMPOSE="podman compose"
elif command -v podman-compose &>/dev/null; then
COMPOSE="podman-compose"
else
echo "error: podman compose or podman-compose required" >&2
exit 1
fi
CONTAINER_NAME="derp"
# podman-compose names images <project>_<service>
IMAGE_NAME="derp_derp"
# Colors (suppressed if NO_COLOR is set or stdout isn't a tty)
if [[ -z "${NO_COLOR:-}" ]] && [[ -t 1 ]]; then
GRN='\e[38;5;108m'
RED='\e[38;5;131m'
BLU='\e[38;5;110m'
DIM='\e[2m'
RST='\e[0m'
else
GRN='' RED='' BLU='' DIM='' RST=''
fi
info() { printf "${GRN}%s${RST} %s\n" "✓" "$*"; }
err() { printf "${RED}%s${RST} %s\n" "✗" "$*" >&2; }
dim() { printf "${DIM} %s${RST}\n" "$*"; }
Executable
+21
View File
@@ -0,0 +1,21 @@
#!/usr/bin/env bash
# Build or rebuild the derp container image.
# Usage: tools/build [--no-cache]
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
cd "$PROJECT_DIR" || exit 1
args=()
[[ "${1:-}" == "--no-cache" ]] && args+=(--no-cache)
dim "Building image..."
$COMPOSE build "${args[@]}"
size=$(podman image inspect "$IMAGE_NAME" --format '{{.Size}}' 2>/dev/null || true)
if [[ -n "$size" ]]; then
human=$(numfmt --to=iec-i --suffix=B "$size" 2>/dev/null || echo "${size} bytes")
info "Image built ($human)"
else
info "Image built"
fi
Executable
+9
View File
@@ -0,0 +1,9 @@
#!/usr/bin/env bash
# Tail container logs.
# Usage: tools/logs [N]
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
tail_n="${1:-30}"
podman logs -f --tail "$tail_n" "$CONTAINER_NAME"
Executable
+26
View File
@@ -0,0 +1,26 @@
#!/usr/bin/env bash
# Full teardown: stop container and remove image.
# Usage: tools/nuke
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
cd "$PROJECT_DIR" || exit 1
dim "Stopping container..."
$COMPOSE down 2>/dev/null || true
before=$(podman system df --format '{{.Size}}' 2>/dev/null | head -1 || true)
dim "Removing image..."
podman rmi "$IMAGE_NAME" 2>/dev/null || true
# Also remove any dangling derp images
podman images --filter "reference=*derp*" --format '{{.ID}}' 2>/dev/null | \
xargs -r podman rmi 2>/dev/null || true
after=$(podman system df --format '{{.Size}}' 2>/dev/null | head -1 || true)
if [[ -n "$before" && -n "$after" ]]; then
info "Teardown complete (images: $before -> $after)"
else
info "Teardown complete"
fi
Executable
+97
View File
@@ -0,0 +1,97 @@
#!/usr/bin/env bash
# Analyze cProfile data from the bot process.
# Usage: tools/profile [OPTIONS] [FILE]
#
# Options:
# -n NUM Show top NUM entries (default: 30)
# -s SORT Sort by: cumtime, tottime, calls, name (default: cumtime)
# -f PATTERN Filter to entries matching PATTERN
# -c Callers view (who calls the hot functions)
# -h Show this help
#
# Examples:
# tools/profile # top 30 by cumulative time
# tools/profile -s tottime -n 20 # top 20 by total time
# tools/profile -f mumble # only mumble-related functions
# tools/profile -c -f stream_audio # who calls stream_audio
# tools/profile data/old.prof # analyze a specific file
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
DEFAULT_PROF="$PROJECT_DIR/data/derp.prof"
TOP=30
SORT="cumtime"
PATTERN=""
CALLERS=false
usage() {
sed -n '2,/^$/s/^# \?//p' "$0"
exit 0
}
while getopts ":n:s:f:ch" opt; do
case $opt in
n) TOP="$OPTARG" ;;
s) SORT="$OPTARG" ;;
f) PATTERN="$OPTARG" ;;
c) CALLERS=true ;;
h) usage ;;
:) err "option -$OPTARG requires an argument"; exit 2 ;;
*) err "unknown option -$OPTARG"; exit 2 ;;
esac
done
shift $((OPTIND - 1))
PROF="${1:-$DEFAULT_PROF}"
if [[ ! -f "$PROF" ]]; then
err "profile not found: $PROF"
dim "run the bot with --cprofile and stop it gracefully"
exit 1
fi
# Validate sort key
case "$SORT" in
cumtime|tottime|calls|name) ;;
*) err "invalid sort key: $SORT (use cumtime, tottime, calls, name)"; exit 2 ;;
esac
# Profile metadata
size=$(stat -c %s "$PROF" 2>/dev/null || stat -f %z "$PROF" 2>/dev/null)
human=$(numfmt --to=iec-i --suffix=B "$size" 2>/dev/null || echo "${size}B")
modified=$(stat -c %y "$PROF" 2>/dev/null | cut -d. -f1)
printf '%b%s%b\n' "$BLU" "Profile" "$RST"
dim "$PROF ($human, $modified)"
echo
# Build pstats script
read -r -d '' PYSCRIPT << 'PYEOF' || true
import pstats
import sys
import io
prof_path = sys.argv[1]
sort_key = sys.argv[2]
top_n = int(sys.argv[3])
pattern = sys.argv[4]
callers = sys.argv[5] == "1"
p = pstats.Stats(prof_path, stream=sys.stdout)
p.strip_dirs()
p.sort_stats(sort_key)
if pattern:
if callers:
p.print_callers(pattern, top_n)
else:
p.print_stats(pattern, top_n)
else:
if callers:
p.print_callers(top_n)
else:
p.print_stats(top_n)
PYEOF
exec python3 -c "$PYSCRIPT" "$PROF" "$SORT" "$TOP" "$PATTERN" "$( $CALLERS && echo 1 || echo 0 )"
Executable
+15
View File
@@ -0,0 +1,15 @@
#!/usr/bin/env bash
# Stop, rebuild, and start the derp container.
# Usage: tools/restart [--no-cache]
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
args=()
[[ "${1:-}" == "--no-cache" ]] && args+=("--no-cache")
"$SCRIPT_DIR/stop"
echo
"$SCRIPT_DIR/build" "${args[@]}"
echo
"$SCRIPT_DIR/start"
Executable
+23
View File
@@ -0,0 +1,23 @@
#!/usr/bin/env bash
# Start the derp container.
# Usage: tools/start
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
cd "$PROJECT_DIR" || exit 1
# Build first if no image exists
if ! podman image exists "$IMAGE_NAME" 2>/dev/null; then
dim "No image found, building..."
"$SCRIPT_DIR/build"
echo
fi
dim "Starting container..."
$COMPOSE up -d
sleep 3
dim "Recent logs:"
podman logs --tail 15 "$CONTAINER_NAME" 2>&1 || true
echo
info "Container started"
Executable
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env bash
# Show container and image state.
# Usage: tools/status
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
# -- Container ----------------------------------------------------------------
printf '%b%s%b\n' "$BLU" "Container" "$RST"
state=$(podman inspect "$CONTAINER_NAME" --format '{{.State.Status}}' 2>/dev/null || true)
if [[ -z "$state" ]]; then
dim "absent"
elif [[ "$state" == "running" ]]; then
uptime=$(podman inspect "$CONTAINER_NAME" --format '{{.State.StartedAt}}' 2>/dev/null || true)
info "running (since ${uptime%.*})"
else
info "$state"
fi
echo
# -- Image --------------------------------------------------------------------
printf '%b%s%b\n' "$BLU" "Image" "$RST"
if podman image exists "$IMAGE_NAME" 2>/dev/null; then
img_info=$(podman image inspect "$IMAGE_NAME" --format '{{.Created}} {{.Size}}' 2>/dev/null || true)
created="${img_info%% *}"
size="${img_info##* }"
human=$(numfmt --to=iec-i --suffix=B "$size" 2>/dev/null || echo "${size}B")
info "$IMAGE_NAME ($human, ${created%T*})"
else
dim "no image"
fi
echo
# -- Volumes ------------------------------------------------------------------
printf '%b%s%b\n' "$BLU" "Mounts" "$RST"
mounts=(src plugins config/derp.toml data secrets)
for m in "${mounts[@]}"; do
path="$PROJECT_DIR/$m"
if [[ -e "$path" ]]; then
info "$m"
else
err "$m (missing)"
fi
done
Executable
+11
View File
@@ -0,0 +1,11 @@
#!/usr/bin/env bash
# Stop and remove the derp container.
# Usage: tools/stop
# shellcheck source=tools/_common.sh
source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/_common.sh"
cd "$PROJECT_DIR" || exit 1
dim "Stopping container..."
$COMPOSE down
info "Container stopped"