From 5a8d568931d9b23ce0df1265d05259a7012081c9 Mon Sep 17 00:00:00 2001 From: Ahmed Date: Sun, 14 Jun 2026 01:46:29 +0300 Subject: init: mostly vibed --- daft/__init__.py | 0 daft/daft.cfg | 16 +++ daft/dj.py | 332 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 348 insertions(+) create mode 100644 daft/__init__.py create mode 100644 daft/daft.cfg create mode 100644 daft/dj.py (limited to 'daft') diff --git a/daft/__init__.py b/daft/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/daft/daft.cfg b/daft/daft.cfg new file mode 100644 index 0000000..1a22761 --- /dev/null +++ b/daft/daft.cfg @@ -0,0 +1,16 @@ +[core] +nick = daft +user = daft +name = daft +host = 127.0.0.1 +port = 6667 +owner = yournick +password = REPLACEME +prefix = ! +channels = #yourchannel +extra = /path/to/irc-bots/daft +enable = + dj +logdir = /var/log/daft +logging_level = INFO +timeout = 120 diff --git a/daft/dj.py b/daft/dj.py new file mode 100644 index 0000000..5a0ec27 --- /dev/null +++ b/daft/dj.py @@ -0,0 +1,332 @@ +""" +Sopel plugin: dj +Playlist manager with admin playback controls. + + !dj - queue a YouTube track (anyone) + !np - show current track (anyone) + !queue - show upcoming tracks (anyone) + !play - start/resume playback (op) + !stop - pause playback (op) + !skip - skip to next track (op) + !remove - remove current song from queue (op) + !loop - loop the queue circularly (op) + !clear - clear queue and stop playback (op) + !join #chan - join a channel (owner) + !part [#chan] - leave a channel (owner) + !chans - list joined channels (owner) + !help - show usage +""" +import re +import threading +from urllib.parse import urlparse, parse_qs + +from sopel import plugin + +QUEUE_CAP = 50 +TRACK_DURATION = 120 # seconds + +_YOUTUBE_HOSTS = frozenset({ + "youtube.com", "www.youtube.com", "m.youtube.com", + "music.youtube.com", "youtu.be", +}) +_REJECT_PATHS = ("/shorts/", "/embed/", "/live/") +_VIDEO_ID_RE = re.compile(r'^[A-Za-z0-9_-]{11}$') + + +def _extract_video_id(url): + """Return the 11-char video ID from a YouTube URL, or None if invalid/rejected.""" + if "://" not in url: + url = "https://" + url + try: + p = urlparse(url) + except Exception: + return None + if p.scheme not in ("http", "https"): + return None + if p.netloc.lower() not in _YOUTUBE_HOSTS: + return None + for bad in _REJECT_PATHS: + if p.path.startswith(bad): + return None + if p.netloc.lower() == "youtu.be": + vid = p.path.strip("/") + return vid if _VIDEO_ID_RE.match(vid) else None + # youtube.com/watch?v=ID or /playlist?v=ID&list=... (take v= only) + qs = parse_qs(p.query) + vid = qs.get("v", [None])[0] + return vid if vid and _VIDEO_ID_RE.match(vid) else None + + +def _normalize(video_id): + return f"https://youtu.be/{video_id}" + + +def setup(bot): + bot.memory.setdefault("dj_queue", []) + bot.memory.setdefault("dj_pos", -1) + bot.memory.setdefault("dj_playing", False) + bot.memory.setdefault("dj_timer", None) + bot.memory.setdefault("dj_loop", False) + bot.memory.setdefault("dj_channel", None) + bot.memory.setdefault("dj_lock", threading.RLock()) + + +def _is_owner(bot, trigger): + return trigger.nick == bot.settings.core.owner + + +def _cancel_timer(bot): + t = bot.memory.get("dj_timer") + if t: + t.cancel() + bot.memory["dj_timer"] = None + + +def _play_at(bot, sender, pos): + queue = bot.memory["dj_queue"] + bot.memory["dj_pos"] = pos + bot.memory["dj_channel"] = str(sender) + bot.say(f"now playing: {queue[pos]}", sender) + t = threading.Timer(TRACK_DURATION, _on_track_end, args=(bot, sender)) + t.daemon = True + t.start() + bot.memory["dj_timer"] = t + + +def _on_track_end(bot, sender): + with bot.memory["dj_lock"]: + if not bot.memory["dj_playing"]: + return + queue = bot.memory["dj_queue"] + pos = bot.memory["dj_pos"] + next_pos = pos + 1 + if next_pos >= len(queue): + if bot.memory["dj_loop"] and len(queue) > 0: + next_pos = 0 + else: + bot.say("queue finished, playback stopped", sender) + bot.memory["dj_playing"] = False + bot.memory["dj_pos"] = -1 + bot.memory["dj_timer"] = None + return + _play_at(bot, sender, next_pos) + + +@plugin.command("dj") +def cmd_dj(bot, trigger): + url = (trigger.group(2) or "").strip() + if not url: + bot.reply("usage: !dj ") + return + vid = _extract_video_id(url) + if not vid: + bot.reply("invalid YouTube URL (shorts, embeds, and live links are not accepted)") + return + normalized = _normalize(vid) + with bot.memory["dj_lock"]: + queue = bot.memory["dj_queue"] + if normalized in queue: + return # duplicate, skip silently + if len(queue) >= QUEUE_CAP: + bot.reply(f"queue is full ({QUEUE_CAP} tracks max)") + return + queue.append(normalized) + pos = len(queue) + bot.say(f"queued at position {pos}") + + +@plugin.command("np") +def cmd_np(bot, trigger): + with bot.memory["dj_lock"]: + queue = bot.memory["dj_queue"] + pos = bot.memory["dj_pos"] + if bot.memory["dj_playing"] and 0 <= pos < len(queue): + bot.say(f"now playing: {queue[pos]}") + else: + bot.say("nothing playing") + + +@plugin.command("queue") +def cmd_queue(bot, trigger): + with bot.memory["dj_lock"]: + queue = list(bot.memory["dj_queue"]) + pos = bot.memory["dj_pos"] + loop = bot.memory["dj_loop"] + + if not queue: + bot.say("queue is empty") + return + + start = pos + 1 if pos >= 0 else 0 + + if loop: + n = len(queue) + upcoming = [] + for i in range(min(4, n - (1 if pos >= 0 else 0))): + idx = (start + i) % n + if idx == pos: + break + upcoming.append(queue[idx]) + label = f"{n} in queue (looping)" + else: + upcoming = queue[start:start + 4] + remaining = max(0, len(queue) - start) + label = f"{remaining} remaining" + + if not upcoming: + bot.say("no upcoming songs in queue") + return + + bot.say(f"up next ({label}):") + for i, song in enumerate(upcoming, 1): + bot.say(f" {i}. {song}") + if not loop and remaining > 4: + bot.say(f" ... and {remaining - 4} more") + + +@plugin.command("play") +@plugin.require_privilege(plugin.OP, "ops only") +def cmd_play(bot, trigger): + with bot.memory["dj_lock"]: + if bot.memory["dj_playing"]: + bot.say("already playing") + return + queue = bot.memory["dj_queue"] + if not queue: + bot.say("queue is empty") + return + pos = bot.memory["dj_pos"] + start = pos if 0 <= pos < len(queue) else 0 + bot.memory["dj_playing"] = True + _play_at(bot, trigger.sender, start) + + +@plugin.command("stop") +@plugin.require_privilege(plugin.OP, "ops only") +def cmd_stop(bot, trigger): + with bot.memory["dj_lock"]: + if not bot.memory["dj_playing"]: + bot.say("already stopped") + return + bot.memory["dj_playing"] = False + _cancel_timer(bot) + bot.say("playback stopped") + + +@plugin.command("skip") +@plugin.require_privilege(plugin.OP, "ops only") +def cmd_skip(bot, trigger): + with bot.memory["dj_lock"]: + if not bot.memory["dj_playing"]: + bot.say("nothing playing") + return + _cancel_timer(bot) + _on_track_end(bot, trigger.sender) + + +@plugin.command("remove") +@plugin.require_privilege(plugin.OP, "ops only") +def cmd_remove(bot, trigger): + with bot.memory["dj_lock"]: + queue = bot.memory["dj_queue"] + pos = bot.memory["dj_pos"] + if not queue or pos < 0 or pos >= len(queue): + bot.say("nothing to remove") + return + removed = queue.pop(pos) + bot.say(f"removed: {removed}") + if bot.memory["dj_playing"]: + # pos now points to the next song; step back so _on_track_end advances to it + bot.memory["dj_pos"] = pos - 1 + _cancel_timer(bot) + _on_track_end(bot, bot.memory.get("dj_channel") or str(trigger.sender)) + else: + bot.memory["dj_pos"] = max(-1, pos - 1) + + +@plugin.command("loop") +@plugin.require_privilege(plugin.OP, "ops only") +def cmd_loop(bot, trigger): + arg = (trigger.group(2) or "").strip().lower() + if arg == "on": + bot.memory["dj_loop"] = True + bot.say("loop on, queue will repeat") + elif arg == "off": + bot.memory["dj_loop"] = False + bot.say("loop off") + else: + state = "on" if bot.memory["dj_loop"] else "off" + bot.say(f"loop is {state}, usage: !loop ") + + +@plugin.command("clear") +@plugin.require_privilege(plugin.OP, "ops only") +def cmd_clear(bot, trigger): + with bot.memory["dj_lock"]: + bot.memory["dj_playing"] = False + _cancel_timer(bot) + bot.memory["dj_queue"].clear() + bot.memory["dj_pos"] = -1 + bot.memory["dj_loop"] = False + bot.memory["dj_channel"] = None + bot.say("queue cleared, playback stopped") + + +@plugin.command("join") +def cmd_join(bot, trigger): + if not _is_owner(bot, trigger): + return + channel = (trigger.group(2) or "").strip() + if not channel.startswith("#"): + bot.say("usage: !join #channel") + return + bot.join(channel) + + +@plugin.command("part") +def cmd_part(bot, trigger): + if not _is_owner(bot, trigger): + return + channel = (trigger.group(2) or "").strip() or trigger.sender + bot.part(channel) + + +@plugin.command("chans") +def cmd_chans(bot, trigger): + if not _is_owner(bot, trigger): + return + chans = sorted(str(c) for c in bot.channels) + bot.say("in: " + " ".join(chans) if chans else "not in any channels") + + +@plugin.command("help") +def cmd_help(bot, trigger): + channel = bot.channels.get(trigger.sender) + is_op = False + if channel: + priv = channel.privileges.get(trigger.nick, 0) + is_op = priv >= plugin.OP + + lines = [ + "!dj - queue a YouTube track", + "!np - show current track", + "!queue - show upcoming tracks", + ] + if is_op or _is_owner(bot, trigger): + lines += [ + "!play - start/resume playback (op)", + "!stop - pause playback (op)", + "!skip - skip current track (op)", + "!remove - remove current song from queue (op)", + "!loop - loop the queue (op)", + "!clear - clear queue and stop (op)", + ] + if _is_owner(bot, trigger): + lines += [ + "!join #channel - join a channel (owner)", + "!part [#channel] - leave a channel (owner)", + "!chans - list joined channels (owner)", + ] + lines.append("!help - this message") + for line in lines: + bot.say(line, trigger.nick) -- cgit v1.2.3