Fix AzuraCast integration against verified API behavior

Fix AzuraCast integration against verified API behavior

Verified against AzuraCast source (Files\BatchAction, FlowUploadAction,
Files\ListAction, Paginator, StationPlaylistMediaRepository):

- Rotation read a nonexistent 'mtime' key; the files API returns
  'timestamp' (unix epoch). Age buckets now work, so tracks actually
  migrate Currents -> Recurrents -> Gold as intended.
- Rotation now filters on type == 'media' (dirs are 'directory', not
  'dir') and skips rows without a path/timestamp.
- Follow ?page= pagination on /files: the default (no params) response
  is an unpaginated JSON array, but any paginated response is now
  traversed instead of only page 1 being visible.
- POST /files/upload returns only {success, message}; it never returns
  unique_id/path, and the file lands at the uploaded filename. Removed
  the dead unique_id search; playlist+queue now use the known path.
- Added success/error checks for batch ops (playlist/queue) instead of
  discarding responses, and queue no longer runs when playlist assign
  or Currents lookup failed.
- Replaced curl subprocess shelling with aiohttp (already a matrix-nio
  dependency): single helper for API calls, real multipart upload,
  timeouts everywhere, no temp JSON/pickle files in DATA_DIR.

Functional test (mock AzuraCast mirroring verified API shapes):
15/15 checks pass - upload contract, pagination, batch ops, and age
bucketing (fresh/6d/10d/30d -> Currents/Currents/Recurrents/Gold,
directories and non-media excluded).

main.py

@@ -12,12 +12,12 @@ import pickle
 import re
 import sqlite3
 import time
-import urllib.parse
 
+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", "./")
@@ -45,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(
@@ -468,7 +551,6 @@ async def process_radio_track(video_link, video_id, title):
     if not AZURACAST_API_KEY:
         return
 
-    base_url = AZURACAST_URL.rstrip("/")
     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)
@@ -508,171 +590,45 @@ async def process_radio_track(video_link, video_id, title):
     filename = os.path.basename(filepath)
 
     print(f"📻 Uploading to AzuraCast: {title}")
-
+    if not await azura_upload_file(filepath, filename):
-    encoded_path = urllib.parse.quote(filename)
+        return
-    upload_cmd = [
+    print(f"✅ Uploaded to AzuraCast: {filename}")
-        "curl",
-        "-s",
-        "-X",
-        "POST",
-        f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/upload?path={encoded_path}",
-        "-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
-    )
-    stdout, _ = await curl_proc.communicate()
-
-    unique_id = None
-    azura_path = None
-
-    try:
-        response = json.loads(stdout.decode())
-        unique_id = response.get("unique_id")
-        azura_path = response.get("path") or response.get("file")
-    except json.JSONDecodeError:
-        print("❌ Failed to parse AzuraCast upload response")
-
-    if not unique_id or not azura_path:
-        try:
-            search_cmd = [
-                "curl",
-                "-s",
-                f"{base_url}/api/station/{AZURACAST_STATION_ID}/files",
-                "-H",
-                f"X-API-Key: {AZURACAST_API_KEY}",
-            ]
-            search_proc = await asyncio.create_subprocess_exec(
-                *search_cmd,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.DEVNULL,
-            )
-            search_out, _ = await search_proc.communicate()
-            search_data = json.loads(search_out.decode())
-            file_list = (
-                search_data
-                if isinstance(search_data, list)
-                else search_data.get("rows", [])
-            )
-            target_suffix = f"_{video_id}."
-            for item in file_list:
-                item_path = item.get("path", "")
-                if target_suffix.lower() in item_path.lower():
-                    unique_id = item.get("unique_id")
-                    azura_path = item.get("path")
-                    break
-        except Exception as e:
-            print(f"❌ Failed to search for uploaded track on AzuraCast: {e}")
 
     # 1. Add the track to the "Currents" playlist FIRST
     # AzuraCast rejects requests for files that aren't assigned to any active playlist.
-    target_path = azura_path if azura_path else filename
+    playlists_status, playlists = await azura_request(
-
+        "GET", f"/api/station/{AZURACAST_STATION_ID}/playlists"
-    try:
+    )
-        pl_cmd = [
+    currents_id = None
-            "curl",
+    if playlists_status == 200 and isinstance(playlists, list):
-            "-s",
-            "-X",
-            "GET",
-            f"{base_url}/api/station/{AZURACAST_STATION_ID}/playlists",
-            "-H",
-            f"X-API-Key: {AZURACAST_API_KEY}",
-        ]
-        pl_proc = await asyncio.create_subprocess_exec(
-            *pl_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL
-        )
-        pl_stdout, _ = await pl_proc.communicate()
-
-        playlists = json.loads(pl_stdout.decode())
         currents_id = next(
             (p.get("id") for p in playlists if p.get("name", "").lower() == "currents"),
             None,
         )
 
-        if currents_id:
+    if currents_id:
-            payload = {
+        assigned = await azura_batch(
+            {
                 "do": "playlist",
                 "playlists": [currents_id],
-                "files": [target_path],
+                "files": [filename],
             }
-            json_path = os.path.join(DATA_DIR, f"assign_{filename}.json")
+        )
-            with open(json_path, "w") as f:
+        if assigned:
-                json.dump(payload, f)
+            print(f"✅ Auto-assigned to 'Currents' (Path: {filename})")
-
+    else:
-            batch_cmd = [
+        print("⚠️ 'Currents' playlist not found on AzuraCast.")
-                "curl",
-                "-s",
-                "-X",
-                "PUT",
-                f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/batch",
-                "-H",
-                f"X-API-Key: {AZURACAST_API_KEY}",
-                "-H",
-                "Content-Type: application/json",
-                "-d",
-                f"@{json_path}",
-            ]
-            b_proc = await asyncio.create_subprocess_exec(
-                *batch_cmd, stdout=asyncio.subprocess.DEVNULL
-            )
-            await b_proc.communicate()
-
-            if os.path.exists(json_path):
-                os.remove(json_path)
-            print(f"✅ Auto-assigned to 'Currents' (Path: {target_path})")
-    except Exception as e:
-        print(f"❌ Failed to auto-assign playlist for {title}: {e}")
 
     # 2. Issue the Play Next Request SECOND
     # Now that the file is legally in a playlist, AzuraCast will accept the request.
-    if target_path:
+    queued = await azura_batch(
-        q_payload = {
+        {
             "do": "queue",
-            "files": [target_path],
+            "files": [filename],
         }
-        q_json_path = os.path.join(DATA_DIR, f"queue_{filename}.json")
+    )
-        with open(q_json_path, "w") as f:
+    if queued:
-            json.dump(q_payload, f)
+        print(f"✅ Queued on radio to play next: {title}")
-
-        q_batch_cmd = [
-            "curl",
-            "-s",
-            "-X",
-            "PUT",
-            f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/batch",
-            "-H",
-            f"X-API-Key: {AZURACAST_API_KEY}",
-            "-H",
-            "Content-Type: application/json",
-            "-d",
-            f"@{q_json_path}",
-        ]
-        q_proc = await asyncio.create_subprocess_exec(
-            *q_batch_cmd,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        q_stdout, _ = await q_proc.communicate()
-
-        if os.path.exists(q_json_path):
-            os.remove(q_json_path)
-
-        try:
-            q_resp = json.loads(q_stdout.decode())
-            if q_resp.get("success") or q_resp.get("code") == 200:
-                print(f"✅ Queued on radio to play next: {title}")
-            else:
-                print(
-                    f"⚠️ AzuraCast rejected the queue request for {title}: {q_stdout.decode()}"
-                )
-        except json.JSONDecodeError:
-            print(f"⚠️ Failed to queue {title}. API returned: {q_stdout.decode()}")
-    else:
-        print("⚠️ Uploaded, but no path returned to make the request.")
 
     if os.path.exists(filepath):
         os.remove(filepath)
@@ -686,31 +642,15 @@ async def automated_playlist_rotation():
                 await asyncio.sleep(3600)
                 continue
 
-            base_url = AZURACAST_URL.rstrip("/")
+            pl_status, playlists = await azura_request(
-
+                "GET", f"/api/station/{AZURACAST_STATION_ID}/playlists"
-            pl_cmd = [
-                "curl",
-                "-s",
-                "-X",
-                "GET",
-                f"{base_url}/api/station/{AZURACAST_STATION_ID}/playlists",
-                "-H",
-                f"X-API-Key: {AZURACAST_API_KEY}",
-            ]
-            pl_proc = await asyncio.create_subprocess_exec(
-                *pl_cmd,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.DEVNULL,
             )
-            pl_stdout, _ = await pl_proc.communicate()
+            if pl_status != 200 or not isinstance(playlists, list):
-
-            try:
-                playlists = json.loads(pl_stdout.decode())
-                pl_map = {p.get("name", "").lower(): p.get("id") for p in playlists}
-            except json.JSONDecodeError:
                 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(
@@ -723,30 +663,10 @@ async def automated_playlist_rotation():
             recurrents_id = pl_map["recurrents"]
             gold_id = pl_map["gold"]
 
-            files_cmd = [
+            file_list = await azura_get_paginated_files()
-                "curl",
+            if file_list is None:
-                "-s",
+                await asyncio.sleep(3600)
-                f"{base_url}/api/station/{AZURACAST_STATION_ID}/files",
+                continue
-                "-H",
-                f"X-API-Key: {AZURACAST_API_KEY}",
-            ]
-            f_proc = await asyncio.create_subprocess_exec(
-                *files_cmd,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.DEVNULL,
-            )
-            f_stdout, _ = await f_proc.communicate()
-
-            try:
-                files_data = json.loads(f_stdout.decode())
-            except json.JSONDecodeError:
-                files_data = []
-
-            file_list = (
-                files_data.get("rows", [])
-                if isinstance(files_data, dict)
-                else files_data
-            )
 
             currents = []
             recurrents = []
@@ -754,11 +674,14 @@ async def automated_playlist_rotation():
             current_time = time.time()
 
             for item in file_list:
-                if item.get("type") == "dir":
+                if item.get("type") != "media":
                     continue
 
                 path = item.get("path")
-                mtime = item.get("mtime", current_time)
+                # 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
 
@@ -771,43 +694,29 @@ async def automated_playlist_rotation():
 
             async def batch_assign(files, pl_id):
                 if not files:
-                    return
+                    return True
+                ok = True
                 for i in range(0, len(files), 50):
                     chunk = files[i : i + 50]
-                    payload = {"do": "playlist", "playlists": [pl_id], "files": chunk}
+                    if not await azura_batch(
-                    json_path = os.path.join(DATA_DIR, f"batch_{pl_id}_{i}.json")
+                        {"do": "playlist", "playlists": [pl_id], "files": chunk}
-                    with open(json_path, "w") as f:
+                    ):
-                        json.dump(payload, f)
+                        ok = False
+                return ok
 
-                    batch_cmd = [
+            results = await asyncio.gather(
-                        "curl",
+                batch_assign(currents, currents_id),
-                        "-s",
+                batch_assign(recurrents, recurrents_id),
-                        "-X",
+                batch_assign(gold, gold_id),
-                        "PUT",
-                        f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/batch",
-                        "-H",
-                        f"X-API-Key: {AZURACAST_API_KEY}",
-                        "-H",
-                        "Content-Type: application/json",
-                        "-d",
-                        f"@{json_path}",
-                    ]
-                    b_proc = await asyncio.create_subprocess_exec(
-                        *batch_cmd, stdout=asyncio.subprocess.DEVNULL
-                    )
-                    await b_proc.communicate()
-
-                    if os.path.exists(json_path):
-                        os.remove(json_path)
-
-            await batch_assign(currents, currents_id)
-            await batch_assign(recurrents, recurrents_id)
-            await batch_assign(gold, gold_id)
-
-            print(
-                f"✅ Playlist rotation: {len(currents)} Currents, {len(recurrents)} Recurrents, {len(gold)} Gold."
             )
 
+            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}")