Twitch chat bot: !skip → MPD next-track via IRC
Self-hosted async Python bot, runs as systemd --user unit (obs-twitch-bot.service, sibling to obs-mpd-bridge). Connects to Twitch IRC over TLS as vault118, listens in #ophi118, advances MPD on !skip and posts the new track back to chat. 5s cooldown swallows skip-spam silently. Strips Unicode Tags-block codepoints (U+E0000–U+E007F) before matching the command so Twitch's anti-duplicate suffix doesn't break every-other !skip when the broadcaster fires the same command twice in a row. Bot account is separate (vault118) to keep the broadcaster's display name out of bot replies; .env holds access token + refresh token + client id (refresh + client id stored for future token rotation, unused today).
This commit is contained in:
Jakub Zych
251
twitch-bot/bot.py
Normal file
251
twitch-bot/bot.py
Normal file
@@ -0,0 +1,251 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Twitch chat → MPD control bot.
|
||||
|
||||
Connects to Twitch IRC over TLS, joins the configured channel, and on `!skip`
|
||||
advances MPD to the next track. The new track shows up on stream because
|
||||
bridges/mpd-state.py is already broadcasting MPD state to the loading overlay.
|
||||
|
||||
Reads .env from the same directory:
|
||||
TWITCH_ACCESS_TOKEN chat OAuth token (with or without `oauth:` prefix)
|
||||
TWITCH_NICK bot login name (lowercase)
|
||||
TWITCH_CHANNEL channel to join, no leading `#` — defaults to TWITCH_NICK
|
||||
|
||||
TWITCH_REFRESH_TOKEN and TWITCH_CLIENT_ID are also stored in .env (paired with
|
||||
the access token at generation time) but the bot doesn't use them yet — token
|
||||
refresh would land here later if access tokens start expiring mid-stream.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import ssl
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from mpd.asyncio import MPDClient
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
ENV = HERE / ".env"
|
||||
|
||||
MPD_HOST = os.environ.get("MPD_HOST", "localhost")
|
||||
MPD_PORT = int(os.environ.get("MPD_PORT", "6600"))
|
||||
|
||||
IRC_HOST = "irc.chat.twitch.tv"
|
||||
IRC_PORT = 6697
|
||||
PING_TIMEOUT = 360.0 # twitch PINGs ~every 5min; reconnect if silent longer
|
||||
SKIP_COOLDOWN = 5.0 # seconds between successful !skip invocations
|
||||
|
||||
# Set TWITCH_BOT_DEBUG=1 (env or .env) to dump every received IRC line. Useful
|
||||
# when a brand-new bot account looks "connected" but doesn't see chat — Twitch
|
||||
# silently degrades unverified accounts in ways that don't surface as errors.
|
||||
DEBUG_RAW = os.environ.get("TWITCH_BOT_DEBUG", "").strip() not in ("", "0", "false")
|
||||
|
||||
|
||||
# ─── tiny .env loader ───────────────────────────────────────────────────────
|
||||
|
||||
def load_env(path: Path) -> dict:
|
||||
"""KEY=VALUE lines, # comments, optional surrounding quotes. Avoids a
|
||||
python-dotenv dependency for three vars."""
|
||||
out = {}
|
||||
if not path.exists():
|
||||
return out
|
||||
for raw in path.read_text().splitlines():
|
||||
line = raw.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, _, v = line.partition("=")
|
||||
out[k.strip()] = v.strip().strip('"').strip("'")
|
||||
return out
|
||||
|
||||
|
||||
# ─── MPD ────────────────────────────────────────────────────────────────────
|
||||
|
||||
class Mpd:
|
||||
"""One shared MPDClient with reconnect-on-failure.
|
||||
|
||||
python-mpd2's asyncio client doesn't auto-reconnect, so each command is
|
||||
wrapped to retry once after dropping the connection — covers MPD
|
||||
restarts and idle-socket timeouts without surfacing transient errors.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._cli = None
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def _ensure(self):
|
||||
if self._cli is None:
|
||||
cli = MPDClient()
|
||||
await cli.connect(MPD_HOST, MPD_PORT)
|
||||
self._cli = cli
|
||||
|
||||
async def skip(self) -> str:
|
||||
async with self._lock:
|
||||
last = None
|
||||
for attempt in (1, 2):
|
||||
try:
|
||||
await self._ensure()
|
||||
await self._cli.next()
|
||||
return _fmt_song(await self._cli.currentsong())
|
||||
except Exception as e:
|
||||
last = e
|
||||
self._cli = None # force reconnect on retry
|
||||
raise last # type: ignore[misc]
|
||||
|
||||
|
||||
def _fmt_song(song: dict) -> str:
|
||||
artist = song.get("artist") or song.get("albumartist") or ""
|
||||
title = song.get("title") or song.get("file", "").rsplit("/", 1)[-1]
|
||||
if isinstance(artist, list): artist = ", ".join(a for a in artist if a)
|
||||
if isinstance(title, list): title = title[0] if title else ""
|
||||
out = f"{artist} — {title}".strip(" —")
|
||||
return out or "(unknown track)"
|
||||
|
||||
|
||||
# ─── Twitch IRC ─────────────────────────────────────────────────────────────
|
||||
|
||||
class TwitchBot:
|
||||
def __init__(self, nick: str, token: str, channel: str, mpd: Mpd):
|
||||
self.nick = nick.lower()
|
||||
self.token = token if token.startswith("oauth:") else f"oauth:{token}"
|
||||
self.channel = "#" + channel.lower().lstrip("#")
|
||||
self.mpd = mpd
|
||||
self._last_skip = 0.0
|
||||
|
||||
async def _send(self, writer: asyncio.StreamWriter, raw: str):
|
||||
writer.write((raw + "\r\n").encode("utf-8"))
|
||||
await writer.drain()
|
||||
|
||||
async def _say(self, writer: asyncio.StreamWriter, msg: str):
|
||||
# twitch chat lines max ~500 chars; truncate defensively, strip newlines
|
||||
msg = msg.replace("\r", " ").replace("\n", " ")[:480]
|
||||
await self._send(writer, f"PRIVMSG {self.channel} :{msg}")
|
||||
|
||||
async def run(self):
|
||||
"""Connect/auth/listen loop with quiet exponential backoff.
|
||||
|
||||
Mirrors the connect-quietly pattern in bridges/mpd-state.py:
|
||||
first failure prints once, then silent retry until the connection
|
||||
comes back. Auth failure is fatal — no point retrying with a bad token.
|
||||
"""
|
||||
backoff = 2
|
||||
BACKOFF_MAX = 30
|
||||
prev_status = None # None | "up" | "down"
|
||||
ctx = ssl.create_default_context()
|
||||
|
||||
while True:
|
||||
try:
|
||||
reader, writer = await asyncio.open_connection(
|
||||
IRC_HOST, IRC_PORT, ssl=ctx,
|
||||
)
|
||||
try:
|
||||
await self._send(writer, f"PASS {self.token}")
|
||||
await self._send(writer, f"NICK {self.nick}")
|
||||
await self._send(writer, f"JOIN {self.channel}")
|
||||
|
||||
if prev_status != "up":
|
||||
print(f"[twitch] connected → {self.channel} as {self.nick}",
|
||||
flush=True)
|
||||
prev_status = "up"
|
||||
backoff = 2
|
||||
|
||||
while True:
|
||||
line_bytes = await asyncio.wait_for(
|
||||
reader.readline(), timeout=PING_TIMEOUT,
|
||||
)
|
||||
if not line_bytes:
|
||||
raise ConnectionError("server closed connection")
|
||||
line = line_bytes.decode("utf-8", errors="replace").rstrip("\r\n")
|
||||
await self._handle(writer, line)
|
||||
finally:
|
||||
writer.close()
|
||||
try:
|
||||
await writer.wait_closed()
|
||||
except Exception:
|
||||
pass
|
||||
except RuntimeError as e:
|
||||
# auth-failed and similar fatals — let systemd flag it
|
||||
print(f"[twitch] fatal: {e}", file=sys.stderr, flush=True)
|
||||
raise
|
||||
except (asyncio.TimeoutError, ConnectionError, OSError) as e:
|
||||
if prev_status == "up":
|
||||
print(f"[twitch] disconnected ({type(e).__name__}: {e}) — retrying",
|
||||
flush=True)
|
||||
elif prev_status is None:
|
||||
print(f"[twitch] unreachable ({type(e).__name__}) — will retry silently",
|
||||
flush=True)
|
||||
prev_status = "down"
|
||||
await asyncio.sleep(backoff)
|
||||
backoff = min(backoff * 2, BACKOFF_MAX)
|
||||
|
||||
async def _handle(self, writer: asyncio.StreamWriter, line: str):
|
||||
if DEBUG_RAW:
|
||||
print(f"[rx] {line}", flush=True)
|
||||
|
||||
# PING/PONG keepalive — twitch sends `PING :tmi.twitch.tv` periodically
|
||||
if line.startswith("PING "):
|
||||
await self._send(writer, "PONG " + line[5:])
|
||||
return
|
||||
|
||||
# NOTICE on bad creds: ":tmi.twitch.tv NOTICE * :Login authentication failed"
|
||||
if " NOTICE " in line and "authentication failed" in line.lower():
|
||||
raise RuntimeError("twitch auth failed — check TWITCH_KEY / TWITCH_NICK")
|
||||
|
||||
# PRIVMSG format: ":user!user@user.tmi.twitch.tv PRIVMSG #chan :body"
|
||||
if " PRIVMSG " not in line:
|
||||
return
|
||||
try:
|
||||
prefix, rest = line.split(" PRIVMSG ", 1)
|
||||
user = prefix[1:].split("!", 1)[0]
|
||||
_, _, body = rest.partition(":")
|
||||
except ValueError:
|
||||
return
|
||||
|
||||
# Twitch's anti-duplicate logic appends an invisible Tags-block
|
||||
# codepoint (U+E0000–U+E007F) to back-to-back identical messages
|
||||
# from the same sender within ~30s. Without stripping it, every
|
||||
# other `!skip` in a series silently fails to match.
|
||||
body = "".join(c for c in body if not (0xE0000 <= ord(c) <= 0xE007F))
|
||||
head = body.strip().split(" ", 1)[0].lower()
|
||||
if head == "!skip":
|
||||
await self._cmd_skip(writer, user)
|
||||
|
||||
async def _cmd_skip(self, writer: asyncio.StreamWriter, user: str):
|
||||
# Silent rate-limit so a flurry of !skip from chat doesn't multi-skip
|
||||
# or spam the channel with replies. First click within the window wins.
|
||||
now = time.monotonic()
|
||||
if now - self._last_skip < SKIP_COOLDOWN:
|
||||
return
|
||||
try:
|
||||
now_playing = await self.mpd.skip()
|
||||
except Exception as e:
|
||||
print(f"[skip] failed for @{user}: {type(e).__name__}: {e}", flush=True)
|
||||
await self._say(writer, f"@{user} couldn't skip — MPD didn't respond")
|
||||
return
|
||||
self._last_skip = now
|
||||
print(f"[skip] @{user} → {now_playing}", flush=True)
|
||||
await self._say(writer, f"⏭ skipped by @{user} → now: {now_playing}")
|
||||
|
||||
|
||||
# ─── entrypoint ─────────────────────────────────────────────────────────────
|
||||
|
||||
async def main():
|
||||
env = {**load_env(ENV), **os.environ} # process env wins over .env
|
||||
token = env.get("TWITCH_ACCESS_TOKEN", "").strip()
|
||||
nick = env.get("TWITCH_NICK", "").strip()
|
||||
channel = env.get("TWITCH_CHANNEL", nick).strip()
|
||||
|
||||
missing = [k for k, v in (("TWITCH_ACCESS_TOKEN", token), ("TWITCH_NICK", nick)) if not v]
|
||||
if missing:
|
||||
print(f"[err] missing in {ENV}: {', '.join(missing)}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
bot = TwitchBot(nick=nick, token=token, channel=channel, mpd=Mpd())
|
||||
await bot.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
sys.exit(0)
|
||||
Reference in New Issue
Block a user