feat(limnoria): add NtrPlaylist IRC plugin

Made-with: Cursor

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) + 4 > _MAX_IRC_LINE:  # +4 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",)