diff --git a/main.py b/main.py index 8008814..221e2b5 100755 --- a/main.py +++ b/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}") - - encoded_path = urllib.parse.quote(filename) - upload_cmd = [ - "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}") + 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. - target_path = azura_path if azura_path else filename - - try: - 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() - - playlists = json.loads(pl_stdout.decode()) + 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: - payload = { + if currents_id: + 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: - json.dump(payload, f) - - 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"@{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}") + ) + 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. - if target_path: - q_payload = { + queued = await azura_batch( + { "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: - json.dump(q_payload, f) - - 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 queued: + print(f"✅ Queued on radio to play next: {title}") 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_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_status, playlists = await azura_request( + "GET", f"/api/station/{AZURACAST_STATION_ID}/playlists" ) - pl_stdout, _ = await pl_proc.communicate() - - try: - playlists = json.loads(pl_stdout.decode()) - pl_map = {p.get("name", "").lower(): p.get("id") for p in playlists} - except json.JSONDecodeError: + 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( @@ -723,30 +663,10 @@ async def automated_playlist_rotation(): recurrents_id = pl_map["recurrents"] gold_id = pl_map["gold"] - files_cmd = [ - "curl", - "-s", - f"{base_url}/api/station/{AZURACAST_STATION_ID}/files", - "-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 - ) + file_list = await azura_get_paginated_files() + if file_list is None: + await asyncio.sleep(3600) + continue 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} - json_path = os.path.join(DATA_DIR, f"batch_{pl_id}_{i}.json") - with open(json_path, "w") as f: - json.dump(payload, f) + if not await azura_batch( + {"do": "playlist", "playlists": [pl_id], "files": chunk} + ): + ok = False + return ok - 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"@{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." + 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}")