diff --git a/plugins/sopel/ntr_playlist.py b/plugins/sopel/ntr_playlist.py new file mode 100644 index 0000000..b3998ed --- /dev/null +++ b/plugins/sopel/ntr_playlist.py @@ -0,0 +1,220 @@ +""" +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) + 4 > _MAX_IRC_LINE: # +4 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 ") + return + parts = raw.strip().split() + if len(parts) < 2: + bot.say("Usage: !song ") + return + try: + episode = int(parts[0]) + position = int(parts[1]) + except ValueError: + bot.say("Usage: !song ") + 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: + bot.say(e.detail) + return + else: + try: + data = _api_get(base_url, "/playlist") + except ApiError as 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: + 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: + bot.say(f"Refresh failed: {e.detail}") + return + count = data.get("track_count", 0) + bot.say(f"Refreshed — {count} tracks")