diff --git a/plugins/limnoria/NtrPlaylist/__init__.py b/plugins/limnoria/NtrPlaylist/__init__.py new file mode 100644 index 0000000..7e918b3 --- /dev/null +++ b/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" diff --git a/plugins/limnoria/NtrPlaylist/config.py b/plugins/limnoria/NtrPlaylist/config.py new file mode 100644 index 0000000..29c6c7d --- /dev/null +++ b/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).""", + ), +) diff --git a/plugins/limnoria/NtrPlaylist/plugin.py b/plugins/limnoria/NtrPlaylist/plugin.py new file mode 100644 index 0000000..8882f97 --- /dev/null +++ b/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): + """ + + Returns a track from a specific episode's playlist. + """ + if not text: + irc.reply("Usage: !song ") + return + parts = text.strip().split() + if len(parts) < 2: + irc.reply("Usage: !song ") + return + try: + episode, position = int(parts[0]), int(parts[1]) + except ValueError: + irc.reply("Usage: !song ") + 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): + """[] + + 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 diff --git a/plugins/limnoria/NtrPlaylist/test.py b/plugins/limnoria/NtrPlaylist/test.py new file mode 100644 index 0000000..72ff2d7 --- /dev/null +++ b/plugins/limnoria/NtrPlaylist/test.py @@ -0,0 +1,5 @@ +from supybot.test import PluginTestCase + + +class NtrPlaylistTestCase(PluginTestCase): + plugins = ("NtrPlaylist",)