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

# 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"
```