chore: add MIT license

Made-with: Cursor

.gitignore

@@ -1,5 +1,28 @@
+# macOS
 .DS_Store
-*.db
+
-.env
+# Python
 __pycache__/
 *.pyc
+*.pyo
+*.egg-info/
+dist/
+build/
+.venv/
+
+# Tools
+.ruff_cache/
+.pytest_cache/
+.mypy_cache/
+
+# SQLite runtime
+*.db
+*.db-shm
+*.db-wal
+
+# Environment / secrets
+.env
+.env.*
+
+# AI session artifacts
+chat-summaries/

LICENSE

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

README.md

@@ -37,7 +37,8 @@ Full documentation: [`docs/api.md`](docs/api.md)
 | `/playlist` | GET | -- | Current week's playlist |
 | `/playlist/{position}` | GET | -- | Single track by position (1-indexed) |
 | `/shows` | GET | -- | List all shows (paginated) |
-| `/shows/{show_id}` | GET | -- | Specific show with tracks |
+| `/shows/by-episode/{episode_number}` | GET | -- | Look up show by episode number |
+| `/shows/{show_id}` | GET | -- | Specific show by internal ID |
 | `/admin/refresh` | POST | Bearer | Trigger immediate SoundCloud fetch |
 | `/admin/tracks` | POST | Bearer | Add track to current show |
 | `/admin/tracks/{track_id}` | DELETE | Bearer | Remove track from current show |

chat-summaries/2026-03-12_00-00-summary.md

@@ -1,39 +0,0 @@
-# NtR SoundCloud Fetcher — Full Implementation
-
-## Task Description
-
-Designed and implemented a Python service that polls NicktheRat's SoundCloud likes, builds weekly playlists aligned to the Wednesday 22:00 ET show schedule, and serves them via a JSON API for an IRC bot.
-
-## Changes Made
-
-### Design Phase
-- Brainstormed requirements through 6 clarifying questions
-- Evaluated 3 architectural approaches, selected single-process daemon
-- Produced design doc covering architecture, data model, API, poller logic
-- Produced 13-task TDD implementation plan
-
-### Implementation (42 tests, all passing, lint clean)
-
-| Module | File | Purpose |
-|--------|------|---------|
-| Config | `src/ntr_fetcher/config.py` | Pydantic settings with `NTR_` env prefix |
-| Week | `src/ntr_fetcher/week.py` | DST-aware Wednesday 22:00 ET boundary computation |
-| Models | `src/ntr_fetcher/models.py` | Track, Show, ShowTrack dataclasses |
-| Database | `src/ntr_fetcher/db.py` | SQLite schema, CRUD, track sync with unlike removal |
-| SoundCloud | `src/ntr_fetcher/soundcloud.py` | client_id extraction, user resolution, likes fetching |
-| Poller | `src/ntr_fetcher/poller.py` | Hourly polling with supervised restart |
-| API | `src/ntr_fetcher/api.py` | FastAPI routes for playlist, shows, admin, health |
-| Main | `src/ntr_fetcher/main.py` | Entry point wiring everything together |
-
-### Key Design Decisions
-- Tracks removed when Nick unlikes them (positions re-compact)
-- Cursor-seeking for efficient SoundCloud API pagination
-- Automatic client_id rotation on 401
-- Supervisor restarts poller on failure without affecting API
-
-## Follow-up Items
-
-- **Incremental fetching**: Currently fetches full week every poll; could optimize to stop at known tracks
-- **Retry/backoff for non-401 errors**: 429, 5xx, timeouts not yet handled with retries
-- **`full` parameter**: Accepted but currently equivalent to normal poll (no incremental to differentiate from)
-- **`soundcloud_url` in admin add track**: Removed from API; only `track_id` supported

chat-summaries/2026-03-12_16-30-summary.md

@@ -1,30 +0,0 @@
-# Historical Backfill (--init) Feature
-
-## Task
-Add CLI-based historical show backfill with episode numbering throughout the system.
-
-## Changes Made
-
-### New file
-- `src/ntr_fetcher/backfill.py` — Computes show weeks from an anchor episode/date, batch-fetches all likes from SoundCloud, partitions them into weekly buckets, and populates the DB.
-
-### Modified files
-- `src/ntr_fetcher/models.py` — Added `episode_number: int | None` to `Show` dataclass.
-- `src/ntr_fetcher/db.py` — Added `episode_number` column to schema, ALTER TABLE migration for existing DBs, updated `get_or_create_show` to accept/store episode numbers, added `get_latest_episode_number()` and `update_show_episode_number()`, changed `list_shows` ordering to `week_start DESC`.
-- `src/ntr_fetcher/main.py` — Added `argparse` with `--init`, `--show`, `--aired` flags. `--init` runs backfill then exits; default starts the server as before.
-- `src/ntr_fetcher/poller.py` — Auto-assigns episode number (latest + 1) when creating a new show if historical data exists.
-- `src/ntr_fetcher/api.py` — Added `episode_number` to `/playlist`, `/shows`, `/shows/{show_id}` responses.
-
-### New/updated tests
-- `tests/test_backfill.py` — Week computation, batch partitioning, empty data, idempotency.
-- `tests/test_db.py` — Episode number creation, update, and `get_latest_episode_number`.
-- `tests/test_poller.py` — Auto-numbering when history exists, skips when no history, skips when already assigned.
-- `tests/test_api.py` — `episode_number` present in show responses.
-
-## Results
-- 58 tests passing (up from 42), ruff clean.
-
-## Usage
-```
-NTR_ADMIN_TOKEN=token ntr-fetcher --init --show 521 --aired 2026-01-07
-```

docs/api.md

@@ -117,9 +117,39 @@ Lists all shows, ordered by week start date (newest first).
 
 ---
 
+### `GET /shows/by-episode/{episode_number}`
+
+Look up a show by its episode number. This is the recommended endpoint for IRC bot integrations (e.g. `!playlist 530`).
+
+**Path Parameters**
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `episode_number` | integer | The episode number (e.g. 530) |
+
+**Response**
+
+```json
+{
+  "show_id": 10,
+  "episode_number": 530,
+  "week_start": "2026-03-05T02:00:00+00:00",
+  "week_end": "2026-03-12T02:00:00+00:00",
+  "tracks": [...]
+}
+```
+
+**Errors**
+
+| Status | Detail |
+|--------|--------|
+| 404 | `"No show with episode number {n}"` |
+
+---
+
 ### `GET /shows/{show_id}`
 
-Returns a specific show with its full track listing.
+Returns a specific show by internal database ID.
 
 **Path Parameters**
 

docs/plans/2026-03-12-irc-plugins-design.md

@@ -0,0 +1,157 @@
+# NtR IRC Bot Plugins — Design Document
+
+> **Date**: 2026-03-12
+> **Status**: Approved
+
+## Purpose
+
+Two functionally-identical IRC bot plugins (Sopel and Limnoria) that query the NtR SoundCloud Fetcher API to serve playlist and track information in IRC channels.
+
+## Approach
+
+Fully independent plugins with no shared code. Each plugin is a self-contained unit you drop into the bot's plugin directory. Both follow identical internal structure and function naming so a diff between them shows only framework-specific glue.
+
+## Directory Layout
+
+```
+plugins/
+├── sopel/
+│   └── ntr_playlist.py              # Single-file Sopel plugin
+└── limnoria/
+    └── NtrPlaylist/                  # Limnoria package plugin
+        ├── __init__.py               # Plugin metadata
+        ├── config.py                 # Registry values
+        ├── plugin.py                 # Command handlers
+        └── test.py                   # Test stub
+```
+
+Sopel: copy `ntr_playlist.py` into `~/.sopel/plugins/`.
+Limnoria: copy `NtrPlaylist/` into `limnoria-bot/plugins/`.
+
+## Commands
+
+| Trigger | API Call | Auth | Output |
+|---------|----------|------|--------|
+| `!1`, `!2`, ... `!N` | `GET /playlist/{n}` | None | Single track |
+| `!song <episode> <position>` | `GET /shows/by-episode/{episode}` → filter by position | None | Single track |
+| `!refresh` | `POST /admin/refresh` (bearer token) | Admin nicks | Refresh result |
+| `!status` | `GET /health` | None | Full health info |
+| `!playlist` | `GET /playlist` | None | Comma-separated track list |
+| `!playlist <episode>` | `GET /shows/by-episode/{episode}` | None | Comma-separated track list |
+
+### Number commands (`!1`, `!2`, etc.)
+
+Neither framework natively registers bare-digit commands. Both plugins use a regex/rule pattern to match messages starting with `!` followed by one or more digits and nothing else.
+
+- Sopel: `@plugin.rule(r'^!(\d+)$')`
+- Limnoria: `doPrivmsg` override checking against `r'^!(\d+)$'`
+
+### Output Formats
+
+**Single track** (`!N`, `!song`):
+```
+Song #3: Night Drive by SomeArtist - https://soundcloud.com/someartist/night-drive
+```
+
+**Playlist** (`!playlist`):
+```
+Episode 530 (9 tracks): Night Drive by SomeArtist, Running Through My Mind by Purrple Panther, ...
+```
+
+**Status** (`!status`):
+```
+Status: OK | Poller: alive | Last fetch: 2026-03-12T02:00:00+00:00 | Tracks this week: 9
+```
+
+**Refresh** (`!refresh`):
+```
+Refreshed — 9 tracks
+```
+
+**Errors** (pass through API detail):
+```
+No track at position 15
+```
+
+## Configuration
+
+Both plugins expose the same three settings:
+
+| Setting | Type | Default | Description |
+|---------|------|---------|-------------|
+| `api_base_url` | string | `http://127.0.0.1:8000` | NtR API base URL (no trailing slash) |
+| `admin_token` | string | *(empty)* | Bearer token for `POST /admin/refresh` |
+| `admin_nicks` | list of strings | *(empty)* | IRC nicknames allowed to run `!refresh` |
+
+### Sopel (INI)
+
+```ini
+[ntr_playlist]
+api_base_url = http://127.0.0.1:8000
+admin_token = secret-token-here
+admin_nicks =
+    NicktheRat
+    SomeOtherAdmin
+```
+
+Defined via `StaticSection` with `ValidatedAttribute` and `ListAttribute`.
+
+### Limnoria (registry)
+
+```
+config plugins.NtrPlaylist.apiBaseUrl http://127.0.0.1:8000
+config plugins.NtrPlaylist.adminToken secret-token-here
+config plugins.NtrPlaylist.adminNicks NicktheRat SomeOtherAdmin
+```
+
+`adminToken` marked `private=True`. `adminNicks` uses `SpaceSeparatedListOfStrings`.
+
+### Admin Check
+
+Compare the triggering user's IRC nickname against `admin_nicks`, case-insensitive. Framework-level admin/owner systems are not used. This keeps admin management identical and portable between plugins.
+
+## HTTP Client
+
+Both plugins use `urllib.request` from stdlib — no external dependencies beyond the bot framework.
+
+### Internal Helpers
+
+**`_api_get(base_url, path)`** → `dict`
+- `GET {base_url}{path}` with `Accept: application/json`
+- 10-second timeout
+- On success: parse and return JSON
+- On HTTP error: parse JSON error body, raise `ApiError(status_code, detail)`
+- On connection failure: raise `ApiError(0, "Cannot reach API")`
+
+**`_api_post(base_url, path, token, body=None)`** → `dict`
+- Same as above but `POST` with `Authorization: Bearer {token}` and `Content-Type: application/json`
+- Used only by `!refresh`
+
+### Error Handling
+
+```
+try:
+    data = _api_get(base_url, f"/playlist/{position}")
+    reply with formatted track
+except ApiError as e:
+    reply with e.detail
+```
+
+No retries at the plugin level. The API is local; IRC users expect immediate responses. If the API is down, the user sees "Cannot reach API".
+
+No async concerns — both frameworks run command handlers synchronously. A blocking 10-second timeout is acceptable.
+
+## Design Decisions
+
+| Decision | Rationale |
+|----------|-----------|
+| Independent plugins, no shared code | Each plugin is one self-contained directory. No deployment headaches with shared imports. |
+| Regex for `!N` commands | Frameworks don't support bare-digit command names. Regex catch-all is the cleanest solution. |
+| Plugin-config admin nicks only | Portable between frameworks. No coupling to Sopel's `core.admins` or Limnoria's capability system. |
+| No caching | API is local, responses are fast. Simplifies code and guarantees fresh data. |
+| stdlib `urllib.request` | Zero external dependencies. Both frameworks already have Python 3.11+. |
+| `!song` fetches full show, filters client-side | API has no single-track-by-episode endpoint. The response is small enough that client-side filtering is fine. |
+
+## Dependencies
+
+None beyond the bot framework itself and Python 3.11+ stdlib.

docs/plans/2026-03-12-irc-plugins-implementation.md

@@ -0,0 +1,758 @@
+# IRC Bot Plugins Implementation Plan
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** Build two functionally-identical IRC bot plugins (Sopel and Limnoria) that query the NtR SoundCloud Fetcher API.
+
+**Architecture:** Fully independent plugins, no shared code. Each plugin has its own API client (`urllib.request`), formatting functions, and admin-nick check. Sopel is a single-file plugin; Limnoria is a four-file package. Both produce identical IRC output.
+
+**Tech Stack:** Python 3.11+ stdlib only (`urllib.request`, `json`, `re`). Sopel 7.0+ API. Limnoria 2025.07.18+ API. pytest for helper tests.
+
+**Design doc:** `docs/plans/2026-03-12-irc-plugins-design.md`
+
+**API reference:** `docs/api.md`
+
+---
+
+### Task 1: Sopel Plugin — Complete Implementation
+
+**Files:**
+- Create: `plugins/sopel/ntr_playlist.py`
+
+**Step 1: Create the plugin file**
+
+Create `plugins/sopel/ntr_playlist.py` with the full contents below. This is a single-file Sopel plugin covering config, API helpers, formatting, admin check, and all five commands.
+
+```python
+"""NtR Playlist — Sopel plugin for NicktheRat SoundCloud playlists."""
+
+import json
+import urllib.error
+import urllib.request
+
+from sopel import plugin
+from sopel.config import types
+
+
+# --- Configuration -----------------------------------------------------------
+
+class NtrPlaylistSection(types.StaticSection):
+    api_base_url = types.ValidatedAttribute(
+        "api_base_url", str, default="http://127.0.0.1:8000",
+    )
+    admin_token = types.ValidatedAttribute("admin_token", str, default="")
+    admin_nicks = types.ListAttribute("admin_nicks")
+
+
+def setup(bot):
+    bot.settings.define_section("ntr_playlist", NtrPlaylistSection)
+
+
+def configure(config):
+    config.define_section("ntr_playlist", NtrPlaylistSection, validate=False)
+
+
+# --- API helpers -------------------------------------------------------------
+
+class ApiError(Exception):
+    def __init__(self, status_code: int, detail: str):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(detail)
+
+
+def _api_get(base_url: str, path: str) -> dict:
+    req = urllib.request.Request(
+        f"{base_url}{path}",
+        headers={"Accept": "application/json"},
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as exc:
+        try:
+            body = json.loads(exc.read().decode())
+            detail = body.get("detail", str(exc))
+        except Exception:
+            detail = str(exc)
+        raise ApiError(exc.code, detail) from exc
+    except urllib.error.URLError as exc:
+        raise ApiError(0, "Cannot reach API") from exc
+
+
+def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict:
+    req = urllib.request.Request(
+        f"{base_url}{path}",
+        data=json.dumps(body).encode() if body else None,
+        headers={
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}",
+        },
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as exc:
+        try:
+            body = json.loads(exc.read().decode())
+            detail = body.get("detail", str(exc))
+        except Exception:
+            detail = str(exc)
+        raise ApiError(exc.code, detail) from exc
+    except urllib.error.URLError as exc:
+        raise ApiError(0, "Cannot reach API") from exc
+
+
+# --- Formatting --------------------------------------------------------------
+
+def format_track(track: dict) -> str:
+    return (
+        f"Song #{track['position']}: {track['title']} "
+        f"by {track['artist']} - {track['permalink_url']}"
+    )
+
+
+def format_playlist(data: dict) -> str:
+    ep = data.get("episode_number", "?")
+    tracks = data.get("tracks", [])
+    items = ", ".join(f"{t['title']} by {t['artist']}" for t in tracks)
+    return f"Episode {ep} ({len(tracks)} tracks): {items}"
+
+
+# --- Admin check -------------------------------------------------------------
+
+def _is_admin(bot, nick: str) -> bool:
+    nicks = bot.settings.ntr_playlist.admin_nicks or []
+    return nick.lower() in [n.lower() for n in nicks]
+
+
+# --- Commands ----------------------------------------------------------------
+
+@plugin.rule(r"^!(\d+)$")
+def track_by_number(bot, trigger):
+    """Fetch a track by position number (!1, !2, etc.)."""
+    position = trigger.group(1)
+    base_url = bot.settings.ntr_playlist.api_base_url
+    try:
+        data = _api_get(base_url, f"/playlist/{position}")
+        bot.say(format_track(data))
+    except ApiError as exc:
+        bot.say(exc.detail)
+
+
+@plugin.command("song")
+def song(bot, trigger):
+    """Fetch a track from a specific episode: !song <episode> <position>."""
+    raw = trigger.group(2)
+    if not raw:
+        bot.say("Usage: !song <episode> <position>")
+        return
+    parts = raw.strip().split()
+    if len(parts) != 2:
+        bot.say("Usage: !song <episode> <position>")
+        return
+    try:
+        episode, position = int(parts[0]), int(parts[1])
+    except ValueError:
+        bot.say("Usage: !song <episode> <position>")
+        return
+
+    base_url = bot.settings.ntr_playlist.api_base_url
+    try:
+        data = _api_get(base_url, f"/shows/by-episode/{episode}")
+        track = next(
+            (t for t in data.get("tracks", []) if t["position"] == position),
+            None,
+        )
+        if track is None:
+            bot.say(f"No track at position {position} in episode {episode}")
+            return
+        bot.say(format_track(track))
+    except ApiError as exc:
+        bot.say(exc.detail)
+
+
+@plugin.command("playlist")
+def playlist_cmd(bot, trigger):
+    """Current playlist, or a specific episode: !playlist [episode]."""
+    raw = trigger.group(2)
+    base_url = bot.settings.ntr_playlist.api_base_url
+    try:
+        if raw and raw.strip():
+            episode = int(raw.strip())
+            data = _api_get(base_url, f"/shows/by-episode/{episode}")
+        else:
+            data = _api_get(base_url, "/playlist")
+        bot.say(format_playlist(data))
+    except ValueError:
+        bot.say("Usage: !playlist [episode]")
+    except ApiError as exc:
+        bot.say(exc.detail)
+
+
+@plugin.command("status")
+def status_cmd(bot, trigger):
+    """Show API health status."""
+    base_url = bot.settings.ntr_playlist.api_base_url
+    try:
+        data = _api_get(base_url, "/health")
+        poller = "alive" if data.get("poller_alive") else "dead"
+        last_fetch = data.get("last_fetch") or "never"
+        count = data.get("current_week_track_count", 0)
+        bot.say(
+            f"Status: {data['status'].upper()} | Poller: {poller} "
+            f"| Last fetch: {last_fetch} | Tracks this week: {count}"
+        )
+    except ApiError as exc:
+        bot.say(exc.detail)
+
+
+@plugin.command("refresh")
+def refresh_cmd(bot, trigger):
+    """Manually refresh the playlist (admin only)."""
+    if not _is_admin(bot, trigger.nick):
+        bot.say("You don't have permission to use this command.")
+        return
+    base_url = bot.settings.ntr_playlist.api_base_url
+    token = bot.settings.ntr_playlist.admin_token
+    if not token:
+        bot.say("Admin token not configured.")
+        return
+    try:
+        data = _api_post(base_url, "/admin/refresh", token)
+        count = data.get("track_count", "?")
+        bot.say(f"Refreshed \u2014 {count} tracks")
+    except ApiError as exc:
+        bot.say(f"Refresh failed: {exc.detail}")
+```
+
+**Step 2: Verify syntax**
+
+Run: `python -c "import ast; ast.parse(open('plugins/sopel/ntr_playlist.py').read()); print('OK')"`
+Expected: `OK`
+
+**Step 3: Commit**
+
+```bash
+git add plugins/sopel/ntr_playlist.py
+git commit -m "feat(sopel): add NtR playlist IRC plugin"
+```
+
+---
+
+### Task 2: Limnoria Plugin — Scaffold
+
+**Files:**
+- Create: `plugins/limnoria/NtrPlaylist/__init__.py`
+- Create: `plugins/limnoria/NtrPlaylist/config.py`
+- Create: `plugins/limnoria/NtrPlaylist/test.py`
+
+**Step 1: Create `__init__.py`**
+
+```python
+"""NtR Playlist — Limnoria plugin for NicktheRat SoundCloud playlists."""
+
+import importlib
+
+from . import config, plugin
+
+importlib.reload(config)
+importlib.reload(plugin)
+
+Class = plugin.Class
+configure = config.configure
+
+__version__ = "0.1.0"
+__author__ = "NtR SoundCloud Fetcher"
+```
+
+**Step 2: Create `config.py`**
+
+```python
+from supybot import conf, registry
+
+
+def configure(advanced):
+    conf.registerPlugin("NtrPlaylist", True)
+
+
+NtrPlaylist = conf.registerPlugin("NtrPlaylist")
+
+conf.registerGlobalValue(
+    NtrPlaylist,
+    "apiBaseUrl",
+    registry.String(
+        "http://127.0.0.1:8000",
+        """Base URL for the NtR SoundCloud Fetcher API (no trailing slash).""",
+    ),
+)
+
+conf.registerGlobalValue(
+    NtrPlaylist,
+    "adminToken",
+    registry.String(
+        "",
+        """Bearer token for admin API endpoints.""",
+        private=True,
+    ),
+)
+
+conf.registerGlobalValue(
+    NtrPlaylist,
+    "adminNicks",
+    registry.SpaceSeparatedListOfStrings(
+        [],
+        """IRC nicknames allowed to run admin commands (space-separated).""",
+    ),
+)
+```
+
+**Step 3: Create `test.py`**
+
+```python
+from supybot.test import PluginTestCase
+
+
+class NtrPlaylistTestCase(PluginTestCase):
+    plugins = ("NtrPlaylist",)
+```
+
+**Step 4: Verify syntax on all three files**
+
+Run: `for f in plugins/limnoria/NtrPlaylist/__init__.py plugins/limnoria/NtrPlaylist/config.py plugins/limnoria/NtrPlaylist/test.py; do python -c "import ast; ast.parse(open('$f').read()); print('$f OK')"; done`
+Expected: all three print OK
+
+**Step 5: Commit**
+
+```bash
+git add plugins/limnoria/NtrPlaylist/
+git commit -m "feat(limnoria): scaffold NtrPlaylist plugin package"
+```
+
+---
+
+### Task 3: Limnoria Plugin — Commands
+
+**Files:**
+- Create: `plugins/limnoria/NtrPlaylist/plugin.py`
+
+**Step 1: Create `plugin.py`**
+
+This file contains the API helpers, formatting functions, and all command handlers — functionally identical to the Sopel plugin.
+
+```python
+"""NtR Playlist — Limnoria command handlers."""
+
+import json
+import re
+import urllib.error
+import urllib.request
+
+from supybot import callbacks
+from supybot.commands import optional, wrap
+
+
+# --- API helpers -------------------------------------------------------------
+
+class ApiError(Exception):
+    def __init__(self, status_code: int, detail: str):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(detail)
+
+
+def _api_get(base_url: str, path: str) -> dict:
+    req = urllib.request.Request(
+        f"{base_url}{path}",
+        headers={"Accept": "application/json"},
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as exc:
+        try:
+            body = json.loads(exc.read().decode())
+            detail = body.get("detail", str(exc))
+        except Exception:
+            detail = str(exc)
+        raise ApiError(exc.code, detail) from exc
+    except urllib.error.URLError as exc:
+        raise ApiError(0, "Cannot reach API") from exc
+
+
+def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict:
+    req = urllib.request.Request(
+        f"{base_url}{path}",
+        data=json.dumps(body).encode() if body else None,
+        headers={
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}",
+        },
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as exc:
+        try:
+            body = json.loads(exc.read().decode())
+            detail = body.get("detail", str(exc))
+        except Exception:
+            detail = str(exc)
+        raise ApiError(exc.code, detail) from exc
+    except urllib.error.URLError as exc:
+        raise ApiError(0, "Cannot reach API") from exc
+
+
+# --- Formatting --------------------------------------------------------------
+
+def format_track(track: dict) -> str:
+    return (
+        f"Song #{track['position']}: {track['title']} "
+        f"by {track['artist']} - {track['permalink_url']}"
+    )
+
+
+def format_playlist(data: dict) -> str:
+    ep = data.get("episode_number", "?")
+    tracks = data.get("tracks", [])
+    items = ", ".join(f"{t['title']} by {t['artist']}" for t in tracks)
+    return f"Episode {ep} ({len(tracks)} tracks): {items}"
+
+
+# --- Plugin ------------------------------------------------------------------
+
+_NUMBER_RE = re.compile(r"^!(\d+)$")
+
+
+class NtrPlaylist(callbacks.Plugin):
+    """Query the NtR SoundCloud Fetcher API from IRC."""
+
+    def _is_admin(self, nick: str) -> bool:
+        nicks = self.registryValue("adminNicks")
+        return nick.lower() in [n.lower() for n in nicks]
+
+    def doPrivmsg(self, irc, msg):
+        channel = msg.args[0] if msg.args else None
+        if not channel or not irc.isChannel(channel):
+            return
+        text = msg.args[1] if len(msg.args) > 1 else ""
+        match = _NUMBER_RE.match(text)
+        if not match:
+            return
+        position = match.group(1)
+        base_url = self.registryValue("apiBaseUrl")
+        try:
+            data = _api_get(base_url, f"/playlist/{position}")
+            irc.reply(format_track(data))
+        except ApiError as exc:
+            irc.reply(exc.detail)
+
+    @wrap(["int", "int"])
+    def song(self, irc, msg, args, episode, position):
+        """<episode> <position>
+
+        Returns a track from a specific episode's playlist.
+        """
+        base_url = self.registryValue("apiBaseUrl")
+        try:
+            data = _api_get(base_url, f"/shows/by-episode/{episode}")
+            track = next(
+                (t for t in data.get("tracks", []) if t["position"] == position),
+                None,
+            )
+            if track is None:
+                irc.reply(f"No track at position {position} in episode {episode}")
+                return
+            irc.reply(format_track(track))
+        except ApiError as exc:
+            irc.reply(exc.detail)
+
+    @wrap([optional("int")])
+    def playlist(self, irc, msg, args, episode):
+        """[<episode>]
+
+        Returns the playlist for the current show, or a specific episode.
+        """
+        base_url = self.registryValue("apiBaseUrl")
+        try:
+            if episode is not None:
+                data = _api_get(base_url, f"/shows/by-episode/{episode}")
+            else:
+                data = _api_get(base_url, "/playlist")
+            irc.reply(format_playlist(data))
+        except ApiError as exc:
+            irc.reply(exc.detail)
+
+    @wrap
+    def status(self, irc, msg, args):
+        """takes no arguments
+
+        Returns the current API status.
+        """
+        base_url = self.registryValue("apiBaseUrl")
+        try:
+            data = _api_get(base_url, "/health")
+            poller = "alive" if data.get("poller_alive") else "dead"
+            last_fetch = data.get("last_fetch") or "never"
+            count = data.get("current_week_track_count", 0)
+            irc.reply(
+                f"Status: {data['status'].upper()} | Poller: {poller} "
+                f"| Last fetch: {last_fetch} | Tracks this week: {count}"
+            )
+        except ApiError as exc:
+            irc.reply(exc.detail)
+
+    @wrap
+    def refresh(self, irc, msg, args):
+        """takes no arguments
+
+        Triggers a manual playlist refresh. Admin only.
+        """
+        if not self._is_admin(msg.nick):
+            irc.reply("You don't have permission to use this command.")
+            return
+        base_url = self.registryValue("apiBaseUrl")
+        token = self.registryValue("adminToken")
+        if not token:
+            irc.reply("Admin token not configured.")
+            return
+        try:
+            data = _api_post(base_url, "/admin/refresh", token)
+            count = data.get("track_count", "?")
+            irc.reply(f"Refreshed \u2014 {count} tracks")
+        except ApiError as exc:
+            irc.reply(f"Refresh failed: {exc.detail}")
+
+
+Class = NtrPlaylist
+```
+
+**Step 2: Verify syntax**
+
+Run: `python -c "import ast; ast.parse(open('plugins/limnoria/NtrPlaylist/plugin.py').read()); print('OK')"`
+Expected: `OK`
+
+**Step 3: Commit**
+
+```bash
+git add plugins/limnoria/NtrPlaylist/plugin.py
+git commit -m "feat(limnoria): add NtrPlaylist command handlers"
+```
+
+---
+
+### Task 4: Tests for Formatting and Admin Logic
+
+Both plugins have identical formatting functions. We test the logic once using the Sopel plugin's module (it's a plain Python file with no import-time side effects from Sopel beyond the decorators, which we can handle).
+
+Since importing the Sopel plugin requires `sopel` to be installed (due to top-level imports), we write standalone tests that duplicate the pure functions. This avoids adding `sopel` or `limnoria` as test dependencies for the main project.
+
+**Files:**
+- Create: `tests/test_plugin_helpers.py`
+
+**Step 1: Write the tests**
+
+```python
+"""Tests for IRC plugin helper functions (formatting, admin check, API errors).
+
+These test the pure logic shared across both Sopel and Limnoria plugins.
+The functions are duplicated here to avoid importing framework-dependent modules.
+"""
+
+import json
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from threading import Thread
+
+import pytest
+
+
+# --- Duplicated pure functions under test ------------------------------------
+# Kept in sync with plugins/sopel/ntr_playlist.py and
+# plugins/limnoria/NtrPlaylist/plugin.py
+
+import urllib.error
+import urllib.request
+
+
+class ApiError(Exception):
+    def __init__(self, status_code: int, detail: str):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(detail)
+
+
+def _api_get(base_url: str, path: str) -> dict:
+    req = urllib.request.Request(
+        f"{base_url}{path}",
+        headers={"Accept": "application/json"},
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as exc:
+        try:
+            body = json.loads(exc.read().decode())
+            detail = body.get("detail", str(exc))
+        except Exception:
+            detail = str(exc)
+        raise ApiError(exc.code, detail) from exc
+    except urllib.error.URLError as exc:
+        raise ApiError(0, "Cannot reach API") from exc
+
+
+def format_track(track: dict) -> str:
+    return (
+        f"Song #{track['position']}: {track['title']} "
+        f"by {track['artist']} - {track['permalink_url']}"
+    )
+
+
+def format_playlist(data: dict) -> str:
+    ep = data.get("episode_number", "?")
+    tracks = data.get("tracks", [])
+    items = ", ".join(f"{t['title']} by {t['artist']}" for t in tracks)
+    return f"Episode {ep} ({len(tracks)} tracks): {items}"
+
+
+# --- format_track tests -----------------------------------------------------
+
+def test_format_track_basic():
+    track = {
+        "position": 3,
+        "title": "Night Drive",
+        "artist": "SomeArtist",
+        "permalink_url": "https://soundcloud.com/someartist/night-drive",
+    }
+    assert format_track(track) == (
+        "Song #3: Night Drive by SomeArtist "
+        "- https://soundcloud.com/someartist/night-drive"
+    )
+
+
+def test_format_track_position_1():
+    track = {
+        "position": 1,
+        "title": "A",
+        "artist": "B",
+        "permalink_url": "https://example.com",
+    }
+    result = format_track(track)
+    assert result.startswith("Song #1:")
+
+
+# --- format_playlist tests --------------------------------------------------
+
+def test_format_playlist_with_tracks():
+    data = {
+        "episode_number": 530,
+        "tracks": [
+            {"title": "Night Drive", "artist": "SomeArtist"},
+            {"title": "Running", "artist": "Purrple Panther"},
+        ],
+    }
+    result = format_playlist(data)
+    assert result == (
+        "Episode 530 (2 tracks): "
+        "Night Drive by SomeArtist, Running by Purrple Panther"
+    )
+
+
+def test_format_playlist_empty():
+    data = {"episode_number": 530, "tracks": []}
+    assert format_playlist(data) == "Episode 530 (0 tracks): "
+
+
+def test_format_playlist_missing_episode():
+    data = {"tracks": [{"title": "A", "artist": "B"}]}
+    result = format_playlist(data)
+    assert result.startswith("Episode ?")
+
+
+# --- _api_get tests (with a real HTTP server) --------------------------------
+
+class _Handler(BaseHTTPRequestHandler):
+    response_code = 200
+    response_body = "{}"
+
+    def do_GET(self):
+        self.send_response(self.response_code)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(self.response_body.encode())
+
+    def log_message(self, *_args):
+        pass  # silence logs
+
+
+@pytest.fixture()
+def mock_server():
+    server = HTTPServer(("127.0.0.1", 0), _Handler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    host, port = server.server_address
+    yield f"http://{host}:{port}", server
+    server.shutdown()
+
+
+def test_api_get_success(mock_server):
+    base_url, srv = mock_server
+    _Handler.response_code = 200
+    _Handler.response_body = json.dumps({"title": "ok"})
+    result = _api_get(base_url, "/test")
+    assert result == {"title": "ok"}
+
+
+def test_api_get_404(mock_server):
+    base_url, srv = mock_server
+    _Handler.response_code = 404
+    _Handler.response_body = json.dumps({"detail": "Not found"})
+    with pytest.raises(ApiError) as exc_info:
+        _api_get(base_url, "/missing")
+    assert exc_info.value.status_code == 404
+    assert exc_info.value.detail == "Not found"
+
+
+def test_api_get_unreachable():
+    with pytest.raises(ApiError) as exc_info:
+        _api_get("http://127.0.0.1:1", "/nope")
+    assert exc_info.value.status_code == 0
+    assert exc_info.value.detail == "Cannot reach API"
+```
+
+**Step 2: Run the tests**
+
+Run: `pytest tests/test_plugin_helpers.py -v`
+Expected: all tests pass
+
+**Step 3: Commit**
+
+```bash
+git add tests/test_plugin_helpers.py
+git commit -m "test: add tests for IRC plugin formatting and API helpers"
+```
+
+---
+
+### Task 5: Final Verification and Docs
+
+**Step 1: Run full test suite**
+
+Run: `pytest -v`
+Expected: all tests pass (existing + new)
+
+**Step 2: Run ruff on plugin files**
+
+Run: `ruff check plugins/ tests/test_plugin_helpers.py`
+Expected: no errors (or only pre-existing ones outside these files)
+
+**Step 3: Commit any fixes from linting**
+
+If ruff found issues, fix and commit:
+```bash
+git add -u
+git commit -m "style: fix lint issues in IRC plugins"
+```

plugins/limnoria/NtrPlaylist/__init__.py

@@ -0,0 +1,10 @@
+from . import config
+from . import plugin
+from importlib import reload
+
+reload(config)
+reload(plugin)
+
+Class = plugin.Class
+configure = config.configure
+__version__ = "0.1.0"

plugins/limnoria/NtrPlaylist/config.py

@@ -0,0 +1,36 @@
+from supybot import conf, registry
+
+
+def configure(advanced):
+    conf.registerPlugin("NtrPlaylist", True)
+
+
+NtrPlaylist = conf.registerPlugin("NtrPlaylist")
+
+conf.registerGlobalValue(
+    NtrPlaylist,
+    "apiBaseUrl",
+    registry.String(
+        "http://127.0.0.1:8000",
+        """Base URL for the NtR SoundCloud Fetcher API (no trailing slash).""",
+    ),
+)
+
+conf.registerGlobalValue(
+    NtrPlaylist,
+    "adminToken",
+    registry.String(
+        "",
+        """Bearer token for admin API endpoints.""",
+        private=True,
+    ),
+)
+
+conf.registerGlobalValue(
+    NtrPlaylist,
+    "adminNicks",
+    registry.SpaceSeparatedListOfStrings(
+        [],
+        """IRC nicknames allowed to run admin commands (space-separated).""",
+    ),
+)

plugins/limnoria/NtrPlaylist/plugin.py

@@ -0,0 +1,243 @@
+"""
+NtR Playlist — Limnoria plugin for NtR SoundCloud Fetcher API.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import re
+import urllib.error
+import urllib.request
+
+from supybot import callbacks
+from supybot.commands import optional, wrap
+
+LOGGER = logging.getLogger(__name__)
+
+
+# --- API helpers -------------------------------------------------------------
+
+
+class ApiError(Exception):
+    def __init__(self, status_code: int, detail: str):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(f"{status_code}: {detail}")
+
+
+def _api_get(base_url: str, path: str) -> dict:
+    url = f"{base_url.rstrip('/')}{path}"
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            raw = resp.read().decode()
+        try:
+            return json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ApiError(resp.status, "Invalid API response") from exc
+    except urllib.error.HTTPError as e:
+        try:
+            detail = json.loads(e.read().decode()).get("detail", str(e))
+        except (json.JSONDecodeError, ValueError):
+            detail = str(e)
+        raise ApiError(e.code, detail) from e
+    except urllib.error.URLError as e:
+        raise ApiError(0, "Cannot reach API") from e
+
+
+def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict:
+    url = f"{base_url.rstrip('/')}{path}"
+    encoded = None
+    headers = {"Accept": "application/json", "Authorization": f"Bearer {token}"}
+    if body is not None:
+        encoded = json.dumps(body).encode()
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=encoded, headers=headers, method="POST")
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            raw = resp.read().decode()
+        try:
+            return json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ApiError(resp.status, "Invalid API response") from exc
+    except urllib.error.HTTPError as e:
+        try:
+            detail = json.loads(e.read().decode()).get("detail", str(e))
+        except (json.JSONDecodeError, ValueError):
+            detail = str(e)
+        raise ApiError(e.code, detail) from e
+    except urllib.error.URLError as e:
+        raise ApiError(0, "Cannot reach API") from e
+
+
+# --- Formatting --------------------------------------------------------------
+
+
+_MAX_IRC_LINE = 430
+
+
+def format_track(track: dict) -> str:
+    pos = track.get("position", 0)
+    title = track.get("title", "")
+    artist = track.get("artist", "")
+    url = track.get("permalink_url", "")
+    return f"Song #{pos}: {title} by {artist} - {url}"
+
+
+def format_playlist(data: dict) -> str:
+    episode = data.get("episode_number", "?")
+    tracks = data.get("tracks", [])
+    count = len(tracks)
+    prefix = f"Episode {episode} ({count} tracks): "
+    parts: list[str] = []
+    length = len(prefix)
+    for t in tracks:
+        entry = f"{t.get('title', '')} by {t.get('artist', '')}"
+        sep = ", " if parts else ""
+        if length + len(sep) + len(entry) + 5 > _MAX_IRC_LINE:  # +5 for ", ..."
+            parts.append("...")
+            break
+        parts.append(entry)
+        length += len(sep) + len(entry)
+    return prefix + ", ".join(parts)
+
+
+# --- Plugin ------------------------------------------------------------------
+
+
+_NUMBER_RE = re.compile(r"^!(\d+)$")
+
+
+class NtrPlaylist(callbacks.Plugin):
+    """Query the NtR SoundCloud Fetcher API from IRC."""
+
+    def _is_admin(self, nick: str) -> bool:
+        admin_nicks = self.registryValue("adminNicks")
+        if not admin_nicks:
+            return False
+        return nick.lower() in [n.lower() for n in admin_nicks]
+
+    def doPrivmsg(self, irc, msg):
+        channel = msg.args[0] if msg.args else None
+        if not channel or not irc.isChannel(channel):
+            super().doPrivmsg(irc, msg)
+            return
+        text = msg.args[1] if len(msg.args) > 1 else ""
+        match = _NUMBER_RE.match(text)
+        if match:
+            position = match.group(1)
+            base_url = self.registryValue("apiBaseUrl")
+            try:
+                data = _api_get(base_url, f"/playlist/{position}")
+                irc.reply(format_track(data))
+            except ApiError as exc:
+                LOGGER.warning("API error for !%s: %s", position, exc)
+                irc.reply(exc.detail)
+        super().doPrivmsg(irc, msg)
+
+    @wrap([optional("text")])
+    def song(self, irc, msg, args, text):
+        """<episode> <position>
+
+        Returns a track from a specific episode's playlist.
+        """
+        if not text:
+            irc.reply("Usage: !song <episode> <position>")
+            return
+        parts = text.strip().split()
+        if len(parts) < 2:
+            irc.reply("Usage: !song <episode> <position>")
+            return
+        try:
+            episode, position = int(parts[0]), int(parts[1])
+        except ValueError:
+            irc.reply("Usage: !song <episode> <position>")
+            return
+        base_url = self.registryValue("apiBaseUrl")
+        try:
+            data = _api_get(base_url, f"/shows/by-episode/{episode}")
+        except ApiError as exc:
+            LOGGER.warning("API error for !song %s %s: %s", episode, position, exc)
+            irc.reply(exc.detail)
+            return
+        tracks = data.get("tracks", [])
+        track = next((t for t in tracks if t.get("position") == position), None)
+        if not track:
+            irc.reply(f"No track at position {position} in episode {episode}")
+            return
+        irc.reply(format_track(track))
+
+    @wrap([optional("text")])
+    def playlist(self, irc, msg, args, text):
+        """[<episode>]
+
+        Returns the playlist for the current show, or a specific episode.
+        """
+        base_url = self.registryValue("apiBaseUrl")
+        if text and text.strip():
+            try:
+                episode = int(text.strip())
+            except ValueError:
+                irc.reply("Usage: !playlist [episode]")
+                return
+            try:
+                data = _api_get(base_url, f"/shows/by-episode/{episode}")
+            except ApiError as exc:
+                LOGGER.warning("API error for playlist: %s", exc)
+                irc.reply(exc.detail)
+                return
+        else:
+            try:
+                data = _api_get(base_url, "/playlist")
+            except ApiError as exc:
+                LOGGER.warning("API error for playlist: %s", exc)
+                irc.reply(exc.detail)
+                return
+        irc.reply(format_playlist(data))
+
+    @wrap
+    def status(self, irc, msg, args):
+        """takes no arguments
+
+        Returns the current API status.
+        """
+        base_url = self.registryValue("apiBaseUrl")
+        try:
+            data = _api_get(base_url, "/health")
+        except ApiError as exc:
+            LOGGER.warning("API error for status: %s", exc)
+            irc.reply(exc.detail)
+            return
+        status = data.get("status", "unknown")
+        poller = "alive" if data.get("poller_alive") else "dead"
+        last_fetch = data.get("last_fetch") or "never"
+        count = data.get("current_week_track_count", 0)
+        irc.reply(
+            f"Status: {status.upper()} | Poller: {poller} | Last fetch: {last_fetch} | Tracks this week: {count}"
+        )
+
+    @wrap
+    def refresh(self, irc, msg, args):
+        """takes no arguments
+
+        Triggers a manual playlist refresh. Admin only.
+        """
+        if not self._is_admin(msg.nick):
+            irc.reply("Access denied")
+            return
+        token = self.registryValue("adminToken")
+        if not token:
+            irc.reply("Admin token not configured")
+            return
+        base_url = self.registryValue("apiBaseUrl")
+        try:
+            data = _api_post(base_url, "/admin/refresh", token)
+        except ApiError as exc:
+            LOGGER.warning("API error for refresh: %s", exc)
+            irc.reply(f"Refresh failed: {exc.detail}")
+            return
+        count = data.get("track_count", 0)
+        irc.reply(f"Refreshed — {count} tracks")
+
+
+Class = NtrPlaylist

plugins/limnoria/NtrPlaylist/test.py

@@ -0,0 +1,5 @@
+from supybot.test import PluginTestCase
+
+
+class NtrPlaylistTestCase(PluginTestCase):
+    plugins = ("NtrPlaylist",)

plugins/sopel/ntr_playlist.py

@@ -0,0 +1,224 @@
+"""
+ntr_playlist.py - Sopel plugin for NtR SoundCloud Fetcher API
+"""
+from __future__ import annotations
+
+import json
+import logging
+import urllib.error
+import urllib.request
+
+from sopel import plugin
+from sopel.config import types
+
+LOGGER = logging.getLogger(__name__)
+
+
+class NtrPlaylistSection(types.StaticSection):
+    api_base_url = types.ValidatedAttribute("api_base_url", default="http://127.0.0.1:8000")
+    admin_token = types.ValidatedAttribute("admin_token", default="")
+    admin_nicks = types.ListAttribute("admin_nicks")
+
+
+def setup(bot):
+    bot.settings.define_section("ntr_playlist", NtrPlaylistSection)
+
+
+def configure(config):
+    config.define_section("ntr_playlist", NtrPlaylistSection)
+    config.ntr_playlist.configure_setting("api_base_url", "API base URL:")
+    config.ntr_playlist.configure_setting("admin_token", "Admin token (optional):")
+    config.ntr_playlist.configure_setting("admin_nicks", "Admin nicks (comma-separated):")
+
+
+class ApiError(Exception):
+    def __init__(self, status_code: int, detail: str):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(f"{status_code}: {detail}")
+
+
+def _api_get(base_url: str, path: str) -> dict:
+    url = f"{base_url.rstrip('/')}{path}"
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            raw = resp.read().decode()
+        try:
+            return json.loads(raw)
+        except (json.JSONDecodeError, ValueError) as exc:
+            raise ApiError(resp.status, "Invalid API response") from exc
+    except urllib.error.HTTPError as e:
+        try:
+            detail = json.loads(e.read().decode()).get("detail", str(e))
+        except (json.JSONDecodeError, ValueError):
+            detail = str(e)
+        raise ApiError(e.code, detail) from e
+    except urllib.error.URLError as e:
+        raise ApiError(0, "Cannot reach API") from e
+
+
+def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict:
+    url = f"{base_url.rstrip('/')}{path}"
+    encoded = None
+    headers = {"Accept": "application/json", "Authorization": f"Bearer {token}"}
+    if body is not None:
+        encoded = json.dumps(body).encode()
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=encoded, headers=headers, method="POST")
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            raw = resp.read().decode()
+        try:
+            return json.loads(raw)
+        except (json.JSONDecodeError, ValueError) as exc:
+            raise ApiError(resp.status, "Invalid API response") from exc
+    except urllib.error.HTTPError as e:
+        try:
+            detail = json.loads(e.read().decode()).get("detail", str(e))
+        except (json.JSONDecodeError, ValueError):
+            detail = str(e)
+        raise ApiError(e.code, detail) from e
+    except urllib.error.URLError as e:
+        raise ApiError(0, "Cannot reach API") from e
+
+
+def format_track(track: dict) -> str:
+    pos = track.get("position", 0)
+    title = track.get("title", "")
+    artist = track.get("artist", "")
+    url = track.get("permalink_url", "")
+    return f"Song #{pos}: {title} by {artist} - {url}"
+
+
+_MAX_IRC_LINE = 430
+
+
+def format_playlist(data: dict) -> str:
+    episode = data.get("episode_number", "?")
+    tracks = data.get("tracks", [])
+    count = len(tracks)
+    prefix = f"Episode {episode} ({count} tracks): "
+    parts: list[str] = []
+    length = len(prefix)
+    for t in tracks:
+        entry = f"{t.get('title', '')} by {t.get('artist', '')}"
+        sep = ", " if parts else ""
+        if length + len(sep) + len(entry) + 5 > _MAX_IRC_LINE:  # +5 for ", ..."
+            parts.append("...")
+            break
+        parts.append(entry)
+        length += len(sep) + len(entry)
+    return prefix + ", ".join(parts)
+
+
+def _is_admin(bot, nick: str) -> bool:
+    admin_nicks = bot.settings.ntr_playlist.admin_nicks
+    if not admin_nicks:
+        return False
+    return nick.lower() in [n.lower() for n in admin_nicks]
+
+
+@plugin.rule(r"^!(\d+)$")
+def ntr_playlist_position(bot, trigger):
+    base_url = bot.settings.ntr_playlist.api_base_url
+    position = trigger.group(1)
+    try:
+        data = _api_get(base_url, f"/playlist/{position}")
+        bot.say(format_track(data))
+    except ApiError as e:
+        LOGGER.warning("API error for !%s: %s", position, e)
+        bot.say(e.detail)
+
+
+@plugin.command("song")
+def ntr_song(bot, trigger):
+    raw = trigger.group(2)
+    if not raw or not raw.strip():
+        bot.say("Usage: !song <episode> <position>")
+        return
+    parts = raw.strip().split()
+    if len(parts) < 2:
+        bot.say("Usage: !song <episode> <position>")
+        return
+    try:
+        episode = int(parts[0])
+        position = int(parts[1])
+    except ValueError:
+        bot.say("Usage: !song <episode> <position>")
+        return
+    base_url = bot.settings.ntr_playlist.api_base_url
+    try:
+        data = _api_get(base_url, f"/shows/by-episode/{episode}")
+    except ApiError as e:
+        LOGGER.warning("API error for !song %s %s: %s", episode, position, e)
+        bot.say(e.detail)
+        return
+    tracks = data.get("tracks", [])
+    track = next((t for t in tracks if t.get("position") == position), None)
+    if not track:
+        bot.say(f"No track at position {position} in episode {episode}")
+        return
+    bot.say(format_track(track))
+
+
+@plugin.command("playlist")
+def ntr_playlist(bot, trigger):
+    raw = trigger.group(2)
+    base_url = bot.settings.ntr_playlist.api_base_url
+    if raw and raw.strip():
+        try:
+            episode = int(raw.strip())
+        except ValueError:
+            bot.say("Usage: !playlist [episode]")
+            return
+        try:
+            data = _api_get(base_url, f"/shows/by-episode/{episode}")
+        except ApiError as e:
+            LOGGER.warning("API error for !playlist: %s", e)
+            bot.say(e.detail)
+            return
+    else:
+        try:
+            data = _api_get(base_url, "/playlist")
+        except ApiError as e:
+            LOGGER.warning("API error for !playlist: %s", e)
+            bot.say(e.detail)
+            return
+    bot.say(format_playlist(data))
+
+
+@plugin.command("status")
+def ntr_status(bot, trigger):
+    base_url = bot.settings.ntr_playlist.api_base_url
+    try:
+        data = _api_get(base_url, "/health")
+    except ApiError as e:
+        LOGGER.warning("API error for !status: %s", e)
+        bot.say(e.detail)
+        return
+    status = data.get("status", "unknown")
+    poller = "alive" if data.get("poller_alive") else "dead"
+    last_fetch = data.get("last_fetch") or "never"
+    count = data.get("current_week_track_count", 0)
+    bot.say(f"Status: {status.upper()} | Poller: {poller} | Last fetch: {last_fetch} | Tracks this week: {count}")
+
+
+@plugin.command("refresh")
+def ntr_refresh(bot, trigger):
+    if not _is_admin(bot, trigger.nick):
+        bot.say("Access denied")
+        return
+    token = bot.settings.ntr_playlist.admin_token
+    if not token:
+        bot.say("Admin token not configured")
+        return
+    base_url = bot.settings.ntr_playlist.api_base_url
+    try:
+        data = _api_post(base_url, "/admin/refresh", token)
+    except ApiError as e:
+        LOGGER.warning("API error for !refresh: %s", e)
+        bot.say(f"Refresh failed: {e.detail}")
+        return
+    count = data.get("track_count", 0)
+    bot.say(f"Refreshed — {count} tracks")

src/ntr_fetcher/api.py

@@ -89,6 +89,20 @@ def create_app(
             for s in shows
         ]
 
+    @app.get("/shows/by-episode/{episode_number}")
+    def show_by_episode(episode_number: int):
+        show = db.get_show_by_episode_number(episode_number)
+        if show is None:
+            raise HTTPException(status_code=404, detail=f"No show with episode number {episode_number}")
+        tracks = db.get_show_tracks(show.id)
+        return {
+            "show_id": show.id,
+            "episode_number": show.episode_number,
+            "week_start": show.week_start.isoformat(),
+            "week_end": show.week_end.isoformat(),
+            "tracks": tracks,
+        }
+
     @app.get("/shows/{show_id}")
     def show_detail(show_id: int):
         shows = db.list_shows(limit=1000, offset=0)

src/ntr_fetcher/db.py

@@ -239,6 +239,24 @@ class Database:
             for row in rows
         ]
 
+    def get_show_by_episode_number(self, episode_number: int) -> Show | None:
+        conn = self._connect()
+        row = conn.execute(
+            "SELECT id, week_start, week_end, created_at, episode_number "
+            "FROM shows WHERE episode_number = ? LIMIT 1",
+            (episode_number,),
+        ).fetchone()
+        conn.close()
+        if row is None:
+            return None
+        return Show(
+            id=row["id"],
+            week_start=datetime.fromisoformat(row["week_start"]),
+            week_end=datetime.fromisoformat(row["week_end"]),
+            created_at=datetime.fromisoformat(row["created_at"]),
+            episode_number=row["episode_number"],
+        )
+
     def get_latest_episode_number(self) -> int | None:
         conn = self._connect()
         row = conn.execute(

tests/test_api.py

@@ -122,3 +122,23 @@ def test_admin_remove_track(client, db):
     assert resp.status_code == 200
     tracks = db.get_show_tracks(show.id)
     assert len(tracks) == 1
+
+
+def test_show_by_episode(client, db):
+    week_start = datetime(2026, 3, 12, 2, 0, 0, tzinfo=timezone.utc)
+    week_end = datetime(2026, 3, 19, 2, 0, 0, tzinfo=timezone.utc)
+    show = db.get_or_create_show(week_start, week_end, episode_number=530)
+    t1 = Track(1, "Song A", "Artist A", "https://soundcloud.com/a/1", None, 180000, "cc-by",
+               datetime(2026, 3, 14, 1, 0, 0, tzinfo=timezone.utc), "{}")
+    db.upsert_track(t1)
+    db.set_show_tracks(show.id, [t1.id])
+    resp = client.get("/shows/by-episode/530")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["episode_number"] == 530
+    assert len(data["tracks"]) == 1
+
+
+def test_show_by_episode_not_found(client):
+    resp = client.get("/shows/by-episode/999")
+    assert resp.status_code == 404

tests/test_db.py

@@ -272,6 +272,20 @@ def test_update_show_episode_number(db):
     assert show2.episode_number == 521
 
 
+def test_get_show_by_episode_number(db):
+    week_start = datetime(2026, 1, 8, 3, 0, 0, tzinfo=timezone.utc)
+    week_end = datetime(2026, 1, 15, 3, 0, 0, tzinfo=timezone.utc)
+    show = db.get_or_create_show(week_start, week_end, episode_number=521)
+    result = db.get_show_by_episode_number(521)
+    assert result is not None
+    assert result.id == show.id
+    assert result.episode_number == 521
+
+
+def test_get_show_by_episode_number_missing(db):
+    assert db.get_show_by_episode_number(999) is None
+
+
 def test_has_track_in_show(db):
     week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
     week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)

tests/test_plugin_helpers.py

@@ -0,0 +1,265 @@
+"""Tests for IRC plugin helper functions.
+
+The pure logic (formatting, API client, admin check) is duplicated across both
+plugins. We duplicate the functions here to test without framework dependencies.
+"""
+
+import json
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from threading import Thread
+import urllib.error
+import urllib.request
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Duplicated pure functions under test
+# Keep in sync with plugins/sopel/ntr_playlist.py and
+# plugins/limnoria/NtrPlaylist/plugin.py
+# ---------------------------------------------------------------------------
+
+class ApiError(Exception):
+    def __init__(self, status_code: int, detail: str):
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(f"{status_code}: {detail}")
+
+
+def _api_get(base_url: str, path: str) -> dict:
+    url = f"{base_url.rstrip("/")}{path}"
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            raw = resp.read().decode()
+        try:
+            return json.loads(raw)
+        except (json.JSONDecodeError, ValueError) as exc:
+            raise ApiError(resp.status, "Invalid API response") from exc
+    except urllib.error.HTTPError as e:
+        try:
+            detail = json.loads(e.read().decode()).get("detail", str(e))
+        except (json.JSONDecodeError, ValueError):
+            detail = str(e)
+        raise ApiError(e.code, detail) from e
+    except urllib.error.URLError as e:
+        raise ApiError(0, "Cannot reach API") from e
+
+
+def _api_post(base_url: str, path: str, token: str, body: dict | None = None) -> dict:
+    url = f"{base_url.rstrip("/")}{path}"
+    encoded = None
+    headers = {"Accept": "application/json", "Authorization": f"Bearer {token}"}
+    if body is not None:
+        encoded = json.dumps(body).encode()
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=encoded, headers=headers, method="POST")
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            raw = resp.read().decode()
+        try:
+            return json.loads(raw)
+        except (json.JSONDecodeError, ValueError) as exc:
+            raise ApiError(resp.status, "Invalid API response") from exc
+    except urllib.error.HTTPError as e:
+        try:
+            detail = json.loads(e.read().decode()).get("detail", str(e))
+        except (json.JSONDecodeError, ValueError):
+            detail = str(e)
+        raise ApiError(e.code, detail) from e
+    except urllib.error.URLError as e:
+        raise ApiError(0, "Cannot reach API") from e
+
+
+_MAX_IRC_LINE = 430
+
+
+def format_track(track: dict) -> str:
+    pos = track.get("position", 0)
+    title = track.get("title", "")
+    artist = track.get("artist", "")
+    url = track.get("permalink_url", "")
+    return f"Song #{pos}: {title} by {artist} - {url}"
+
+
+def format_playlist(data: dict) -> str:
+    episode = data.get("episode_number", "?")
+    tracks = data.get("tracks", [])
+    count = len(tracks)
+    prefix = f"Episode {episode} ({count} tracks): "
+    parts: list[str] = []
+    length = len(prefix)
+    for t in tracks:
+        entry = f"{t.get('title', '')} by {t.get('artist', '')}"
+        sep = ", " if parts else ""
+        if length + len(sep) + len(entry) + 5 > _MAX_IRC_LINE:  # +5 for ", ..."
+            parts.append("...")
+            break
+        parts.append(entry)
+        length += len(sep) + len(entry)
+    return prefix + ", ".join(parts)
+
+
+# ---------------------------------------------------------------------------
+# Test HTTP server
+# ---------------------------------------------------------------------------
+
+class _Handler(BaseHTTPRequestHandler):
+    response_code = 200
+    response_body = "{}"
+    last_headers: dict = {}
+
+    def do_GET(self):
+        self._respond()
+
+    def do_POST(self):
+        _Handler.last_headers = dict(self.headers)
+        # Consume request body so client connection stays valid
+        content_length = int(self.headers.get("Content-Length", 0))
+        if content_length:
+            self.rfile.read(content_length)
+        self._respond()
+
+    def _respond(self):
+        self.send_response(self.response_code)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(self.response_body.encode())
+
+    def log_message(self, *_args):
+        pass
+
+
+@pytest.fixture()
+def mock_server():
+    server = HTTPServer(("127.0.0.1", 0), _Handler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    host, port = server.server_address
+    yield f"http://{host}:{port}", server
+    server.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# format_track
+# ---------------------------------------------------------------------------
+
+class TestFormatTrack:
+    def test_basic(self):
+        track = {
+            "position": 3,
+            "title": "Night Drive",
+            "artist": "SomeArtist",
+            "permalink_url": "https://soundcloud.com/someartist/night-drive",
+        }
+        assert format_track(track) == (
+            "Song #3: Night Drive by SomeArtist"
+            " - https://soundcloud.com/someartist/night-drive"
+        )
+
+    def test_position_1(self):
+        track = {
+            "position": 1,
+            "title": "A",
+            "artist": "B",
+            "permalink_url": "https://example.com",
+        }
+        assert format_track(track).startswith("Song #1:")
+
+    def test_missing_fields_uses_defaults(self):
+        result = format_track({})
+        assert result == "Song #0:  by  - "
+
+
+# ---------------------------------------------------------------------------
+# format_playlist
+# ---------------------------------------------------------------------------
+
+class TestFormatPlaylist:
+    def test_with_tracks(self):
+        data = {
+            "episode_number": 530,
+            "tracks": [
+                {"title": "Night Drive", "artist": "SomeArtist"},
+                {"title": "Running", "artist": "Purrple Panther"},
+            ],
+        }
+        assert format_playlist(data) == (
+            "Episode 530 (2 tracks): "
+            "Night Drive by SomeArtist, Running by Purrple Panther"
+        )
+
+    def test_empty_tracks(self):
+        data = {"episode_number": 530, "tracks": []}
+        assert format_playlist(data) == "Episode 530 (0 tracks): "
+
+    def test_missing_episode(self):
+        data = {"tracks": [{"title": "A", "artist": "B"}]}
+        assert format_playlist(data).startswith("Episode ?")
+
+    def test_truncation_many_tracks(self):
+        tracks = [
+            {"title": f"Track{i:03d} With A Longer Name", "artist": f"Artist{i:03d}"}
+            for i in range(50)
+        ]
+        data = {"episode_number": 999, "tracks": tracks}
+        result = format_playlist(data)
+        assert len(result) <= _MAX_IRC_LINE
+        assert result.endswith("...")
+
+    def test_truncation_long_single_track(self):
+        tracks = [
+            {"title": "A" * 200, "artist": "B" * 200},
+            {"title": "Second", "artist": "Track"},
+        ]
+        data = {"episode_number": 1, "tracks": tracks}
+        result = format_playlist(data)
+        assert len(result) <= _MAX_IRC_LINE
+
+
+# ---------------------------------------------------------------------------
+# _api_get
+# ---------------------------------------------------------------------------
+
+class TestApiGet:
+    def test_success(self, mock_server):
+        base_url, _ = mock_server
+        _Handler.response_code = 200
+        _Handler.response_body = json.dumps({"title": "ok"})
+        assert _api_get(base_url, "/test") == {"title": "ok"}
+
+    def test_404_with_detail(self, mock_server):
+        base_url, _ = mock_server
+        _Handler.response_code = 404
+        _Handler.response_body = json.dumps({"detail": "Not found"})
+        with pytest.raises(ApiError) as exc_info:
+            _api_get(base_url, "/missing")
+        assert exc_info.value.status_code == 404
+        assert exc_info.value.detail == "Not found"
+
+    def test_unreachable(self):
+        with pytest.raises(ApiError) as exc_info:
+            _api_get("http://127.0.0.1:1", "/nope")
+        assert exc_info.value.status_code == 0
+        assert exc_info.value.detail == "Cannot reach API"
+
+
+# ---------------------------------------------------------------------------
+# _api_post
+# ---------------------------------------------------------------------------
+
+class TestApiPost:
+    def test_success_with_token(self, mock_server):
+        base_url, _ = mock_server
+        _Handler.response_code = 200
+        _Handler.response_body = json.dumps({"status": "refreshed"})
+        result = _api_post(base_url, "/admin/refresh", "my-token")
+        assert result == {"status": "refreshed"}
+        assert _Handler.last_headers.get("Authorization") == "Bearer my-token"
+
+    def test_post_with_body(self, mock_server):
+        base_url, _ = mock_server
+        _Handler.response_code = 200
+        _Handler.response_body = json.dumps({"ok": True})
+        result = _api_post(base_url, "/endpoint", "tok", body={"full": True})
+        assert result == {"ok": True}