Fix radio downloads: ship EJS solver scripts and allow runtime refresh

yt-dlp now solves YouTube JS challenges via external EJS scripts (yt-dlp-ejs package). A bare yt-dlp from PyPI does not include them, and remote fetching is disabled by default, so images built after the EJS migration download nothing and fail with HTTP 403 while metadata and playlists keep working. Pin yt-dlp[default] so yt-dlp-ejs ships in the image, and pass --remote-components ejs:github so solver scripts refresh at runtime when YouTube rotates challenges.

Dockerfile

@@ -1,12 +1,22 @@
-FROM python:3
+FROM python:3.14
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
 
 ENV PYTHONUNBUFFERED=1
 
 WORKDIR /usr/src/app
 
-COPY main.py parker.gif requirements.txt ./
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends curl ffmpeg quickjs && \
+    rm -rf /var/lib/apt/lists/*
 
-RUN pip install --no-cache-dir -r requirements.txt
+COPY requirements.txt ./
+RUN uv pip install --system --no-cache -r requirements.txt
+
+COPY main.py parker.gif ./
+
+RUN mkdir -p /usr/src/app/.cache && \
+    chown -R 1000:1000 /usr/src/app/.cache
 
 USER 1000:1000
 

README.md

@@ -1,16 +1,17 @@
 # ParkerBot
 
 ParkerBot is a Matrix bot that monitors a channel for YouTube links and
-generates weekly playlists from them.
+generates weekly playlists from them. It also sends YouTube link titles to the
+channel it's in.
 
 ## Running locally
 
 1. Clone the repo
 2. Install the dependencies, preferably in a venv:
    ```shell
-   python3 -m venv venv
+   uv venv --python 3.14 --seed
-   source ./venv/bin/activate
+   source .venv/bin/activate
-   pip3 install -r requirements.txt
+   uv pip install -r requirements.txt
    ```
 3. Copy [example.env](example.env) to `.env`, customize it.
 4. Source `.env`:

example.env

@@ -12,3 +12,9 @@ MATRIX_PASSWORD = ""
 YOUTUBE_PLAYLIST_TITLE = ""
 # YouTube API client secret json path.
 YOUTUBE_CLIENT_SECRETS_FILE = ""
+# AzuraCast API key.
+AZURACAST_API_KEY = ""
+# AzuraCast API URL.
+AZURACAST_URL = ""
+# AzuraCast station ID.
+AZURACAST_STATION_ID = ""

main.py

@@ -4,17 +4,20 @@
 import argparse
 import asyncio
 import datetime
+import glob
 import html
+import json
 import os
 import pickle
 import re
 import sqlite3
 import time
 
+import aiohttp
 from google.auth.transport.requests import Request
 from google_auth_oauthlib.flow import InstalledAppFlow
-from googleapiclient.discovery import build
 from googleapiclient import errors
+from googleapiclient.discovery import build
 from nio import AsyncClient, RoomMessageText, SyncResponse, UploadResponse
 
 DATA_DIR = os.getenv("DATA_DIR", "./")
@@ -30,6 +33,10 @@ MATRIX_PASSWORD = os.getenv("MATRIX_PASSWORD")
 YOUTUBE_CLIENT_SECRETS_FILE = os.getenv("YOUTUBE_CLIENT_SECRETS_FILE")
 YOUTUBE_PLAYLIST_TITLE = os.getenv("YOUTUBE_PLAYLIST_TITLE")
 
+AZURACAST_API_KEY = os.getenv("AZURACAST_API_KEY")
+AZURACAST_URL = os.getenv("AZURACAST_URL")
+AZURACAST_STATION_ID = os.getenv("AZURACAST_STATION_ID")
+
 
 def connect_db():
     """Connect to DB and return connection and cursor."""
@@ -38,6 +45,89 @@ def connect_db():
     return conn, cursor
 
 
+def azura_headers():
+    """Return auth headers for the AzuraCast API."""
+    return {"X-API-Key": AZURACAST_API_KEY}
+
+
+async def azura_request(method, path, params=None, json_body=None, timeout=60):
+    """Make an AzuraCast API request, return (status, parsed JSON or None)."""
+    url = f"{AZURACAST_URL.rstrip('/')}{path}"
+    try:
+        async with aiohttp.ClientSession(
+            timeout=aiohttp.ClientTimeout(total=timeout), headers=azura_headers()
+        ) as session, session.request(
+            method, url, params=params, json=json_body
+        ) as resp:
+            text = await resp.text()
+            try:
+                data = json.loads(text) if text else None
+            except json.JSONDecodeError:
+                data = None
+            return resp.status, data
+    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+        print(f"❌ AzuraCast request failed ({method} {path}): {e}")
+        return 0, None
+
+
+async def azura_upload_file(filepath, filename):
+    """Upload a local file to the station media root. Returns True on success."""
+    url = f"{AZURACAST_URL.rstrip('/')}/api/station/{AZURACAST_STATION_ID}/files/upload"
+    try:
+        # File handle is kept open for the async upload and closed in `finally`
+        # (a `with` block would close it before the request body is consumed).
+        file_handle = open(filepath, "rb")  # noqa: SIM115
+        try:
+            data = aiohttp.FormData()
+            data.add_field(
+                "file", file_handle, filename=filename, content_type="audio/mpeg"
+            )
+            async with aiohttp.ClientSession(
+                timeout=aiohttp.ClientTimeout(total=600), headers=azura_headers()
+            ) as session, session.post(url, data=data) as resp:
+                if resp.status == 200:
+                    return True
+                body = await resp.text()
+                print(
+                    f"❌ AzuraCast upload failed (HTTP {resp.status}): {body[:200]}"
+                )
+                return False
+        finally:
+            file_handle.close()
+    except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as e:
+        print(f"❌ AzuraCast upload failed for {filename}: {e}")
+        return False
+
+
+async def azura_get_paginated_files():
+    """Fetch the full station media list, following pagination if present."""
+    path = f"/api/station/{AZURACAST_STATION_ID}/files"
+    page = 1
+    files = []
+    while True:
+        status, data = await azura_request("GET", path, params={"page": page})
+        if status != 200 or data is None:
+            print(f"❌ Failed to list AzuraCast files (HTTP {status})")
+            return None
+        if isinstance(data, list):
+            return data
+        rows = data.get("rows", [])
+        files.extend(rows)
+        if page >= int(data.get("total_pages") or 1):
+            return files
+        page += 1
+
+
+async def azura_batch(payload):
+    """Run a bulk operation (do: playlist/queue/...) on the files batch API."""
+    path = f"/api/station/{AZURACAST_STATION_ID}/files/batch"
+    status, data = await azura_request("PUT", path, json_body=payload)
+    if status != 200 or not isinstance(data, dict) or not data.get("success", True):
+        print(f"⚠️ AzuraCast batch op failed (HTTP {status}): {data}")
+        return False
+    return True
+
+
 def parse_arguments():
     """Parse command line arguments."""
     parser = argparse.ArgumentParser(
@@ -76,7 +166,7 @@ def define_tables(conn, cursor):
                 playlist_id TEXT UNIQUE,
                 creation_date DATE)"""
         )
-        cursor.execute(  # TODO: Write migration script to add video_id.
+        cursor.execute(
             """CREATE TABLE IF NOT EXISTS playlist_tracks (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 playlist_id INTEGER,
@@ -91,12 +181,10 @@ def define_tables(conn, cursor):
 def get_authenticated_service():
     """Get an authentivated YouTube service."""
     credentials = None
-    # Stores the user's access and refresh tokens.
     if os.path.exists(PICKLE_PATH):
         with open(PICKLE_PATH, "rb") as token:
             credentials = pickle.load(token)
 
-    # If there are no valid credentials available, let the user log in.
     if not credentials or not credentials.valid:
         if credentials and credentials.expired and credentials.refresh_token:
             credentials.refresh(Request())
@@ -106,7 +194,6 @@ def get_authenticated_service():
                 scopes=["https://www.googleapis.com/auth/youtube.force-ssl"],
             )
             credentials = flow.run_local_server(port=8080)
-            # Save the credentials for the next run
             with open(PICKLE_PATH, "wb") as token:
                 pickle.dump(credentials, token)
 
@@ -150,7 +237,7 @@ def get_or_make_playlist(conn, cursor, youtube, playlist_date):
 
     playlist_id = make_playlist(youtube, title)
     with conn:
-        cursor.execute(  # TODO: https://docs.python.org/3/library/sqlite3.html#default-adapters-and-converters-deprecated
+        cursor.execute(
             "INSERT INTO playlists (title, playlist_id, creation_date) VALUES (?, ?, ?)",
             (title, playlist_id, playlist_date),
         )
@@ -179,30 +266,46 @@ def add_video_to_playlist(youtube, playlist_id, video_id, retry_count=6):
             raise error
 
 
+def parse_iso8601_duration(duration_str):
+    """Parse ISO 8601 duration string into seconds."""
+    pattern = re.compile(r'^P(?:(\d+)W)?(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)?$')
+    match = pattern.match(duration_str)
+    if not match:
+        return 0
+    weeks = int(match.group(1)) if match.group(1) else 0
+    days = int(match.group(2)) if match.group(2) else 0
+    hours = int(match.group(3)) if match.group(3) else 0
+    minutes = int(match.group(4)) if match.group(4) else 0
+    seconds = int(match.group(5)) if match.group(5) else 0
+    return weeks * 604800 + days * 86400 + hours * 3600 + minutes * 60 + seconds
+
+
 def get_video_info(youtube, video_id):
-    """Check whether a YouTube video is music and return its title and channel."""
+    """Check whether a YouTube video is music and return its title, channel, and duration."""
     try:
-        video_details = youtube.videos().list(id=video_id, part="snippet").execute()
+        video_details = youtube.videos().list(id=video_id, part="snippet,contentDetails").execute()
 
-        # Check if the video actually exists/is accessible
         if not video_details.get("items"):
-            return False, "[Video unavailable or private]", ""
+            return False, "[Video unavailable or private]", "", 0
 
-        snippet = video_details["items"][0]["snippet"]
+        item = video_details["items"][0]
+        snippet = item["snippet"]
+        content_details = item.get("contentDetails", {})
+        duration_str = content_details.get("duration", "PT0S")
+        duration = parse_iso8601_duration(duration_str)
 
-        # Check if the video category is Music (10) or Entertainment (24)
         is_music = snippet.get("categoryId") in ("10", "24")
         title = snippet.get("title", "[Unknown Title]")
         channel = snippet.get("channelTitle", "[Unknown Channel]")
 
-        return is_music, title, channel
+        return is_music, title, channel, duration
 
     except errors.HttpError as error:
         print(f"YouTube API error fetching info for {video_id}: {error}")
-        return False, "[Error fetching title]", ""
+        return False, "[Error fetching title]", "", 0
     except Exception as e:
         print(f"Unexpected error fetching info for {video_id}: {e}")
-        return False, "[Error fetching title]", ""
+        return False, "[Error fetching title]", "", 0
 
 
 async def send_intro_message(client, sender, room_id):
@@ -218,11 +321,9 @@ async def send_intro_message(client, sender, room_id):
         content={"msgtype": "m.text", "body": intro_message},
     )
 
-    # TODO: Figure out how to properly send GIF, this is broken as shit.
     with open("./parker.gif", "rb") as gif_file:
         response = await client.upload(gif_file, content_type="image/gif")
     if isinstance(response, UploadResponse):
-        print("Image was uploaded successfully to server. ")
         gif_uri = response.content_uri
         await client.room_send(
             room_id=room_id,
@@ -234,8 +335,6 @@ async def send_intro_message(client, sender, room_id):
                 "info": {"mimetype": "image/gif"},
             },
         )
-    else:
-        print(f"Failed to upload image. Failure response: {response}")
 
 
 async def send_playlist_of_week(client, sender, room_id, playlist_id):
@@ -274,11 +373,11 @@ async def message_callback(conn, cursor, youtube, client, room, event):
         )
 
         timestamp_sec = datetime.datetime.fromtimestamp(
-            event.server_timestamp / 1000, datetime.UTC  # millisec to sec
+            event.server_timestamp / 1000,
+            datetime.UTC,
         )
         current_time = datetime.datetime.now(datetime.UTC)
 
-        # Account for up to 5 minutes of clock drift
         recent = abs(current_time - timestamp_sec) < datetime.timedelta(minutes=5)
 
         if body == "!parkerbot" and recent:
@@ -302,19 +401,17 @@ async def message_callback(conn, cursor, youtube, client, room, event):
         for link in youtube_links:
             video_id = link.split("v=")[-1].split("&")[0].split("/")[-1]
 
-            # Safely fetch the category check, title, and channel
+            is_music_vid, title, channel, duration = get_video_info(youtube, video_id)
-            is_music_vid, title, channel = get_video_info(youtube, video_id)
 
-            # Send the title to the channel so people know what the link is
-            # Only do this for recent messages to prevent spam during backwards-sync
             if recent:
                 plain_text = f"{title}"
                 if channel:
                     plain_text += f" - {channel}"
 
-                # Escape HTML characters to prevent broken rendering in Matrix
                 escaped_text = html.escape(plain_text)
-                html_text = f"<em><span data-mx-color='#808080'>{escaped_text}</span></em>"
+                html_text = (
+                    f"<em><span data-mx-color='#808080'>{escaped_text}</span></em>"
+                )
 
                 await client.room_send(
                     room_id=room.room_id,
@@ -323,17 +420,31 @@ async def message_callback(conn, cursor, youtube, client, room, event):
                         "msgtype": "m.text",
                         "body": plain_text,
                         "format": "org.matrix.custom.html",
-                        "formatted_body": html_text
+                        "formatted_body": html_text,
                     },
                 )
 
-            # Only add to the playlist if it's categorized as music/entertainment
+                if is_music_vid and duration > 1200:
+                    warning_text = f"⚠️ Track is too long for the radio ({(duration // 60)}m {duration % 60}s). Skipped radio upload."
+                    await client.room_send(
+                        room_id=room.room_id,
+                        message_type="m.room.message",
+                        content={
+                            "msgtype": "m.text",
+                            "body": warning_text,
+                            "format": "org.matrix.custom.html",
+                            "formatted_body": f"<em><span data-mx-color='#808080'>{html.escape(warning_text)}</span></em>",
+                        },
+                    )
+
+                elif not is_music_vid:
+                    print(f"📻 Skipping radio upload for {title} (Not categorized as Music/Entertainment)")
+
             if is_music_vid:
                 message_id = record_message(conn, cursor, sender, link, timestamp)
                 if in_playlist(cursor, video_id, playlist_id):
                     print(f"Track is already in this week's playlist: {link}")
                 else:
-                    # Add video to playlists and record it in the database
                     add_video_to_playlist(youtube, playlist_id, video_id)
                     add_video_to_playlist(youtube, all_playlist_id, video_id)
                     with conn:
@@ -345,6 +456,14 @@ async def message_callback(conn, cursor, youtube, client, room, event):
                             (playlist_id, message_id, video_id),
                         )
                     print(f"Added track to this week's playlist: {link}")
+                    if recent:
+                        if 0 < duration <= 1200:
+                            asyncio.create_task(process_radio_track(link, video_id, title))
+                        else:
+                            if duration == 0:
+                                print(f"📻 Skipping radio upload for {title} (duration unknown or live stream)")
+                            else:
+                                print(f"📻 Skipping radio upload for {title} (duration: {duration}s > 1200s)")
 
 
 def in_playlist(cursor, video_id, playlist_id):
@@ -415,22 +534,199 @@ async def backwards_sync(conn, cursor, youtube, client, room, start_token):
     from_token = start_token
     room_id = room.room_id
     while True:
-        # Fetch room messages
         response = await client.room_messages(room_id, from_token, direction="b")
 
-        # Process each message
         for event in response.chunk:
             if isinstance(event, RoomMessageText):
                 await message_callback(conn, cursor, youtube, client, room, event)
 
-        # Break if there are no more messages to fetch
         if not response.end or response.end == from_token:
             break
 
-        # Update the from_token for the next iteration
         from_token = response.end
 
 
+async def process_radio_track(video_link, video_id, title):
+    """Downloads native audio and pushes it to AzuraCast API in the background."""
+    if not AZURACAST_API_KEY:
+        return
+
+    safe_title = "".join(c for c in title if c.isalnum() or c in " -_").strip()
+    base_filename = f"{safe_title}_{video_id}"
+    base_filepath = os.path.join(DATA_DIR, base_filename)
+
+    print(f"📻 Downloading audio for radio: {title}")
+
+    dl_cmd = [
+        "yt-dlp",
+        "--cache-dir",
+        "/usr/src/app/.cache",
+        "--extract-audio",
+        "--embed-metadata",
+        "--embed-thumbnail",
+        "--convert-thumbnails",
+        "jpg",
+        "--js-runtimes",
+        "quickjs",
+        # Refresh EJS challenge-solver scripts from GitHub at runtime so a
+        # stale image doesn't break downloads when YouTube rotates challenges.
+        "--remote-components",
+        "ejs:github",
+        "--output",
+        f"{base_filepath}.%(ext)s",
+        video_link,
+    ]
+    process = await asyncio.create_subprocess_exec(
+        *dl_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+    )
+    stdout, stderr = await process.communicate()
+    if process.returncode != 0:
+        print(f"❌ yt-dlp crashed! Error:\n{stderr.decode().strip()}")
+        return
+
+    downloaded_files = glob.glob(f"{base_filepath}.*")
+
+    if not downloaded_files:
+        print(f"❌ Failed to download audio for {title}")
+        return
+
+    filepath = downloaded_files[0]
+    filename = os.path.basename(filepath)
+
+    print(f"📻 Uploading to AzuraCast: {title}")
+    if not await azura_upload_file(filepath, filename):
+        return
+    print(f"✅ Uploaded to AzuraCast: {filename}")
+
+    # 1. Add the track to the "Currents" playlist FIRST
+    # AzuraCast rejects requests for files that aren't assigned to any active playlist.
+    playlists_status, playlists = await azura_request(
+        "GET", f"/api/station/{AZURACAST_STATION_ID}/playlists"
+    )
+    currents_id = None
+    if playlists_status == 200 and isinstance(playlists, list):
+        currents_id = next(
+            (p.get("id") for p in playlists if p.get("name", "").lower() == "currents"),
+            None,
+        )
+
+    if currents_id:
+        assigned = await azura_batch(
+            {
+                "do": "playlist",
+                "playlists": [currents_id],
+                "files": [filename],
+            }
+        )
+        if assigned:
+            print(f"✅ Auto-assigned to 'Currents' (Path: {filename})")
+    else:
+        print("⚠️ 'Currents' playlist not found on AzuraCast.")
+
+    # 2. Issue the Play Next Request SECOND
+    # Now that the file is legally in a playlist, AzuraCast will accept the request.
+    queued = await azura_batch(
+        {
+            "do": "queue",
+            "files": [filename],
+        }
+    )
+    if queued:
+        print(f"✅ Queued on radio to play next: {title}")
+
+    if os.path.exists(filepath):
+        os.remove(filepath)
+
+
+async def automated_playlist_rotation():
+    """Runs continuously, sorting AzuraCast files by age once an hour."""
+    while True:
+        try:
+            if not AZURACAST_API_KEY:
+                await asyncio.sleep(3600)
+                continue
+
+            pl_status, playlists = await azura_request(
+                "GET", f"/api/station/{AZURACAST_STATION_ID}/playlists"
+            )
+            if pl_status != 200 or not isinstance(playlists, list):
+                await asyncio.sleep(3600)
+                continue
+
+            pl_map = {p.get("name", "").lower(): p.get("id") for p in playlists}
+
+            req_playlists = ["currents", "recurrents", "gold"]
+            if not all(p in pl_map for p in req_playlists):
+                print(
+                    "⚠️ Missing required playlists. Please create 'Currents', 'Recurrents', and 'Gold'."
+                )
+                await asyncio.sleep(3600)
+                continue
+
+            currents_id = pl_map["currents"]
+            recurrents_id = pl_map["recurrents"]
+            gold_id = pl_map["gold"]
+
+            file_list = await azura_get_paginated_files()
+            if file_list is None:
+                await asyncio.sleep(3600)
+                continue
+
+            currents = []
+            recurrents = []
+            gold = []
+            current_time = time.time()
+
+            for item in file_list:
+                if item.get("type") != "media":
+                    continue
+
+                path = item.get("path")
+                # The files API exposes the file's mtime as "timestamp" (unix ts).
+                mtime = item.get("timestamp")
+                if not path or mtime is None:
+                    continue
+
+                age_days = (current_time - mtime) / 86400
+
+                if age_days <= 7:
+                    currents.append(path)
+                elif age_days <= 14:
+                    recurrents.append(path)
+                else:
+                    gold.append(path)
+
+            async def batch_assign(files, pl_id):
+                if not files:
+                    return True
+                ok = True
+                for i in range(0, len(files), 50):
+                    chunk = files[i : i + 50]
+                    if not await azura_batch(
+                        {"do": "playlist", "playlists": [pl_id], "files": chunk}
+                    ):
+                        ok = False
+                return ok
+
+            results = await asyncio.gather(
+                batch_assign(currents, currents_id),
+                batch_assign(recurrents, recurrents_id),
+                batch_assign(gold, gold_id),
+            )
+
+            if all(results):
+                print(
+                    f"✅ Playlist rotation: {len(currents)} Currents, {len(recurrents)} Recurrents, {len(gold)} Gold."
+                )
+            else:
+                print("⚠️ Playlist rotation completed with errors.")
+
+        except Exception as e:
+            print(f"💥 Error in automated_playlist_rotation: {e}")
+
+        await asyncio.sleep(3600)
+
+
 async def main():
     """Get DB and Matrix client ready, and start syncing."""
     args = parse_arguments()
@@ -440,12 +736,13 @@ async def main():
     client = await get_client(conn, cursor, youtube)
     sync_token = load_sync_token()
 
-    # This is incredibly dumb and most probably will exceed your YouTube API quota.
     if args.backwards_sync:
         init_sync = await client.sync(30000)
         room = await client.room_resolve_alias(MATRIX_ROOM)
         await backwards_sync(conn, cursor, youtube, client, room, init_sync.next_batch)
 
+    asyncio.create_task(automated_playlist_rotation())
+
     await client.sync_forever(30000, full_state=True, since=sync_token)
 
 

requirements.txt

@@ -1,3 +1,7 @@
 matrix-nio == 0.24.0
-google-auth-oauthlib
+google-auth-oauthlib == 1.3.1
-google-api-python-client
+google-api-python-client == 2.194.0
+# The "default" extra pulls in yt-dlp-ejs, the JS challenge solver scripts
+# YouTube requires for downloads. Keep it in lockstep with yt-dlp.
+yt-dlp[default]
+mutagen