feat(sopel): add NtR playlist IRC plugin
Made-with: Cursor
This commit is contained in:
cottongin
220
plugins/sopel/ntr_playlist.py
Normal file
220
plugins/sopel/ntr_playlist.py
Normal file
@@ -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 <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:
|
||||
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")
|
||||
Reference in New Issue
Block a user