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

@@ -7,7 +7,7 @@ ENV PYTHONUNBUFFERED=1
 WORKDIR /usr/src/app
 
 RUN apt-get update && \
-    apt-get install -y --no-install-recommends curl ffmpeg yt-dlp && \
+    apt-get install -y --no-install-recommends curl ffmpeg quickjs && \
     rm -rf /var/lib/apt/lists/*
 
 COPY requirements.txt ./
@@ -15,6 +15,9 @@ 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
 
 CMD ["python", "./main.py"]

main.py

@@ -13,10 +13,11 @@ 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", "./")
@@ -44,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(
@@ -82,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,
@@ -97,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())
@@ -112,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)
 
@@ -156,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),
         )
@@ -185,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):
@@ -224,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,
@@ -240,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):
@@ -281,11 +374,10 @@ async def message_callback(conn, cursor, youtube, client, room, event):
 
         timestamp_sec = datetime.datetime.fromtimestamp(
             event.server_timestamp / 1000,
-            datetime.UTC,  # millisec to sec
+            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:
@@ -309,17 +401,13 @@ 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 = get_video_info(youtube, video_id)
+            is_music_vid, title, channel, duration = 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>"
@@ -336,13 +424,27 @@ async def message_callback(conn, cursor, youtube, client, room, event):
                     },
                 )
 
-            # 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:
@@ -355,7 +457,13 @@ async def message_callback(conn, cursor, youtube, client, room, event):
                         )
                     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):
@@ -426,19 +534,15 @@ 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
 
 
@@ -447,28 +551,39 @@ async def process_radio_track(video_link, video_id, title):
     if not AZURACAST_API_KEY:
         return
 
-    # Sanitize title for the local filesystem
     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}")
 
-    # 1. Download with yt-dlp (Keeping native AAC/Opus format)
     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.DEVNULL, stderr=asyncio.subprocess.DEVNULL
+        *dl_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
     )
-    await process.communicate()
+    stdout, stderr = await process.communicate()
+    if process.returncode != 0:
+        print(f"❌ yt-dlp crashed! Error:\n{stderr.decode().strip()}")
+        return
 
-    # Check for the file using glob since the extension (.m4a, .webm, .opus) is dynamic
     downloaded_files = glob.glob(f"{base_filepath}.*")
 
     if not downloaded_files:
@@ -476,59 +591,142 @@ async def process_radio_track(video_link, video_id, 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}")
 
-    # 2. Upload to AzuraCast using curl
-    upload_cmd = [
-        "curl",
-        "-s",
-        "-X",
-        "POST",
-        f"{AZURACAST_URL}/api/station/{AZURACAST_STATION_ID}/files",
-        "-H",
-        f"X-API-Key: {AZURACAST_API_KEY}",
-        "-F",
-        f"file=@{filepath}",
-    ]
-
-    curl_proc = await asyncio.create_subprocess_exec(
-        *upload_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+    # 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"
     )
-    stdout, _ = await curl_proc.communicate()
-
-    # 3. Extract unique_id and push to the live queue
-    try:
-        response = json.loads(stdout.decode())
-        unique_id = response.get("unique_id")
-
-        if unique_id:
-            req_cmd = [
-                "curl",
-                "-s",
-                "-X",
-                "POST",
-                f"{AZURACAST_URL}/api/station/{AZURACAST_STATION_ID}/request/{unique_id}",
-                "-H",
-                f"X-API-Key: {AZURACAST_API_KEY}",
-            ]
-            req_proc = await asyncio.create_subprocess_exec(
-                *req_cmd,
-                stdout=asyncio.subprocess.DEVNULL,
-                stderr=asyncio.subprocess.DEVNULL,
+    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,
         )
-            await req_proc.communicate()
-            print(f"✅ Queued on radio: {title}")
+
+    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(f"⚠️ Uploaded, but no unique_id returned. Response: {stdout.decode()}")
-    except json.JSONDecodeError:
-        print(f"❌ Failed to parse AzuraCast response: {stdout.decode()}")
+        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}")
 
-    # Cleanup the local file so your container doesn't bloat
     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()
@@ -538,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 == 1.3.1
 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