#!/usr/bin/env python3 """ParkerBot""" import argparse import asyncio import datetime import glob import html import json import os import pickle import re import sqlite3 import time from google.auth.transport.requests import Request from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient import errors from nio import AsyncClient, RoomMessageText, SyncResponse, UploadResponse DATA_DIR = os.getenv("DATA_DIR", "./") DB_PATH = os.path.join(DATA_DIR, "parkerbot.sqlite3") PICKLE_PATH = os.path.join(DATA_DIR, "token.pickle") TOKEN_PATH = os.path.join(DATA_DIR, "sync_token") MATRIX_SERVER = os.getenv("MATRIX_SERVER") MATRIX_ROOM = os.getenv("MATRIX_ROOM") MATRIX_USER = os.getenv("MATRIX_USER") 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.""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() return conn, cursor def parse_arguments(): """Parse command line arguments.""" parser = argparse.ArgumentParser( description=( "Matrix bot to generate YouTube (music) playlists from links sent " "to a channel." ) ) parser.add_argument( "--backwards-sync", action="store_true", help=( "Run backwards sync on start. This most probably will cause you to " "exceed your YouTube daily API quota, and other hidden YouTube rate" " limits." ), ) return parser.parse_args() def define_tables(conn, cursor): """Define tables for use with program.""" with conn: cursor.execute( """CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, sender TEXT, message TEXT, timestamp DATETIME, UNIQUE (sender, message, timestamp))""" ) cursor.execute( """CREATE TABLE IF NOT EXISTS playlists ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, playlist_id TEXT UNIQUE, creation_date DATE)""" ) cursor.execute( """CREATE TABLE IF NOT EXISTS playlist_tracks ( id INTEGER PRIMARY KEY AUTOINCREMENT, playlist_id INTEGER, message_id INTEGER, video_id TEXT, FOREIGN KEY (playlist_id) REFERENCES playlists(id), FOREIGN KEY (message_id) REFERENCES messages(id), UNIQUE (playlist_id, message_id))""" ) def get_authenticated_service(): """Get an authentivated YouTube service.""" credentials = None if os.path.exists(PICKLE_PATH): with open(PICKLE_PATH, "rb") as token: credentials = pickle.load(token) if not credentials or not credentials.valid: if credentials and credentials.expired and credentials.refresh_token: credentials.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file( YOUTUBE_CLIENT_SECRETS_FILE, scopes=["https://www.googleapis.com/auth/youtube.force-ssl"], ) credentials = flow.run_local_server(port=8080) with open(PICKLE_PATH, "wb") as token: pickle.dump(credentials, token) return build("youtube", "v3", credentials=credentials) def monday_date(timestamp): """Return Monday of week for given timestamp. Weeks start on Monday.""" date = datetime.datetime.fromtimestamp(timestamp / 1000, datetime.UTC) return date - datetime.timedelta(days=date.weekday()) def make_playlist(youtube, title): """Make a playlist with given title.""" response = ( youtube.playlists() .insert( part="snippet,status", body={ "snippet": { "title": title, "description": "Weekly playlist generated by ParkerBot", }, "status": {"privacyStatus": "public"}, }, ) .execute() ) return response["id"] def get_or_make_playlist(conn, cursor, youtube, playlist_date): """Get ID of playlist with given named suffix, make if doesn't exist.""" title = f"{YOUTUBE_PLAYLIST_TITLE} {playlist_date.strftime('%Y-%m-%d')}" cursor.execute("SELECT playlist_id FROM playlists WHERE title = ?", (title,)) row = cursor.fetchone() if row: return row[0] playlist_id = make_playlist(youtube, title) with conn: cursor.execute( "INSERT INTO playlists (title, playlist_id, creation_date) VALUES (?, ?, ?)", (title, playlist_id, playlist_date), ) return playlist_id def add_video_to_playlist(youtube, playlist_id, video_id, retry_count=6): """Add video to playlist.""" for attempt in range(retry_count): try: youtube.playlistItems().insert( part="snippet", body={ "snippet": { "playlistId": playlist_id, "resourceId": {"kind": "youtube#video", "videoId": video_id}, } }, ).execute() break except errors.HttpError as error: if attempt < retry_count - 1: time.sleep(2**attempt) continue raise error def parse_iso8601_duration(duration_str): """Parse ISO 8601 duration string into seconds.""" pattern = re.compile(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?') match = pattern.match(duration_str) if not match: return 0 hours = int(match.group(1)) if match.group(1) else 0 minutes = int(match.group(2)) if match.group(2) else 0 seconds = int(match.group(3)) if match.group(3) else 0 return hours * 3600 + minutes * 60 + seconds def get_video_info(youtube, video_id): """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,contentDetails").execute() if not video_details.get("items"): return False, "[Video unavailable or private]", "", 0 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) 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, duration except errors.HttpError as error: print(f"YouTube API error fetching info for {video_id}: {error}") 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]", "", 0 async def send_intro_message(client, sender, room_id): """Sends introduction message in reply to sender, in room with room_id.""" intro_message = ( f"Hi {sender}, I'm ParkerBot! I generate YouTube playlists from links " "sent to this channel. You can find my source code here: " "https://git.abdulocra.cy/abdulocracy/parkerbot" ) await client.room_send( room_id=room_id, message_type="m.room.message", content={"msgtype": "m.text", "body": intro_message}, ) with open("./parker.gif", "rb") as gif_file: response = await client.upload(gif_file, content_type="image/gif") if isinstance(response, UploadResponse): gif_uri = response.content_uri await client.room_send( room_id=room_id, message_type="m.room.message", content={ "msgtype": "m.image", "url": gif_uri, "body": "parker.gif", "info": {"mimetype": "image/gif"}, }, ) async def send_playlist_of_week(client, sender, room_id, playlist_id): """Sends playlist of the week in reply to sender, in room with room_id.""" playlist_link = f"https://www.youtube.com/playlist?list={playlist_id}" reply_msg = f"{sender}, here's the playlist of the week: {playlist_link}" await client.room_send( room_id=room_id, message_type="m.room.message", content={"msgtype": "m.text", "body": reply_msg}, ) async def send_playlist_of_all(client, sender, room_id, playlist_id): """Sends playlist of all time in reply to sender, in room with room_id.""" playlist_link = f"https://www.youtube.com/playlist?list={playlist_id}" reply_msg = f"{sender}, here's the playlist of all time: {playlist_link}" await client.room_send( room_id=room_id, message_type="m.room.message", content={"msgtype": "m.text", "body": reply_msg}, ) async def message_callback(conn, cursor, youtube, client, room, event): """Event handler for received messages.""" sender = event.sender if sender != MATRIX_USER: body = event.body.strip() timestamp = event.server_timestamp playlist_id = get_or_make_playlist( conn, cursor, youtube, monday_date(timestamp) ) all_playlist_id = get_or_make_playlist( conn, cursor, youtube, datetime.datetime.fromtimestamp(0) ) timestamp_sec = datetime.datetime.fromtimestamp( event.server_timestamp / 1000, datetime.UTC, ) current_time = datetime.datetime.now(datetime.UTC) recent = abs(current_time - timestamp_sec) < datetime.timedelta(minutes=5) if body == "!parkerbot" and recent: await send_intro_message(client, sender, room.room_id) return if body == "!week" and recent: await send_playlist_of_week(client, sender, room.room_id, playlist_id) return if body == "!all" and recent: await send_playlist_of_all(client, sender, room.room_id, all_playlist_id) return youtube_link_pattern = ( r"(https?://(?:www\.|music\.)?youtube\.com/(?!playlist\?list=)watch" r"\?v=[\w-]+|https?://youtu\.be/[\w-]+)" ) youtube_links = re.findall(youtube_link_pattern, body) for link in youtube_links: video_id = link.split("v=")[-1].split("&")[0].split("/")[-1] is_music_vid, title, channel, duration = get_video_info(youtube, video_id) if recent: plain_text = f"{title}" if channel: plain_text += f" - {channel}" escaped_text = html.escape(plain_text) html_text = ( f"{escaped_text}" ) await client.room_send( room_id=room.room_id, message_type="m.room.message", content={ "msgtype": "m.text", "body": plain_text, "format": "org.matrix.custom.html", "formatted_body": html_text, }, ) 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"{html.escape(warning_text)}", }, ) 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_playlist(youtube, playlist_id, video_id) add_video_to_playlist(youtube, all_playlist_id, video_id) with conn: cursor.execute( ( "INSERT INTO playlist_tracks (playlist_id, message_id, video_id) " "VALUES (?, ?, ?)" ), (playlist_id, message_id, video_id), ) print(f"Added track to this week's playlist: {link}") if recent: if duration <= 1200: asyncio.create_task(process_radio_track(link, video_id, title)) else: print(f"📻 Skipping radio upload for {title} (duration: {duration}s > 1200s)") def in_playlist(cursor, video_id, playlist_id): """Checks if video is in playlist.""" cursor.execute( "SELECT id FROM playlist_tracks WHERE video_id = ? AND playlist_id = ?", (video_id, playlist_id), ) if cursor.fetchone(): return True return False def record_message(conn, cursor, sender, link, timestamp): """Records message to messages table in DB, returns ID.""" try: with conn: cursor.execute( "INSERT INTO messages (sender, message, timestamp) VALUES (?, ?, ?)", (sender, link, timestamp), ) print(f"Saved message: {sender} {link} {timestamp}") except sqlite3.IntegrityError as e: if "UNIQUE constraint failed" in str(e): print(f"Entry already exists: {sender} {link} {timestamp}") else: raise e cursor.execute( "SELECT id FROM messages WHERE sender = ? AND message = ? AND timestamp = ?", (sender, link, timestamp), ) return cursor.fetchone()[0] async def sync_callback(response): """Saves Matrix sync token.""" with open(TOKEN_PATH, "w", encoding="utf-8") as f: f.write(response.next_batch) def load_sync_token(): """Gets saved Matrix sync token if it exists.""" try: with open(TOKEN_PATH, "r", encoding="utf-8") as file: return file.read().strip() except FileNotFoundError: return None async def get_client(conn, cursor, youtube): """Returns configured and logged in Matrix client.""" client = AsyncClient(MATRIX_SERVER, MATRIX_USER) client.add_event_callback( lambda room, event: message_callback( conn, cursor, youtube, client, room, event ), RoomMessageText, ) client.add_response_callback(sync_callback, SyncResponse) print(await client.login(MATRIX_PASSWORD)) return client async def backwards_sync(conn, cursor, youtube, client, room, start_token): """Fetch and process historical messages from a given room.""" print("Starting to process channel log...") from_token = start_token room_id = room.room_id while True: response = await client.room_messages(room_id, from_token, direction="b") for event in response.chunk: if isinstance(event, RoomMessageText): await message_callback(conn, cursor, youtube, client, room, event) if not response.end or response.end == from_token: break 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 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) 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", "--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}") upload_cmd = [ "curl", "-s", "-X", "POST", f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/upload", "-H", f"X-API-Key: {AZURACAST_API_KEY}", "-F", f"path={filename}", "-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 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()) currents_id = next( (p.get("id") for p in playlists if p.get("name", "").lower() == "currents"), None, ) if currents_id: payload = { "do": "playlist", "playlists": [currents_id], "files": [target_path], } 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}") # 2. Issue the Play Next Request SECOND # Now that the file is legally in a playlist, AzuraCast will accept the request. if unique_id: req_cmd = [ "curl", "-s", "-X", "POST", f"{base_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.PIPE, stderr=asyncio.subprocess.PIPE, ) req_stdout, _ = await req_proc.communicate() try: req_resp = json.loads(req_stdout.decode()) # Check the JSON response directly so we don't lie about success if req_resp.get("success") or req_resp.get("code") == 200: print(f"✅ Queued on radio to play next: {title}") else: print( f"⚠️ AzuraCast rejected the request for {title}: {req_stdout.decode()}" ) except json.JSONDecodeError: print(f"⚠️ Failed to queue {title}. API returned: {req_stdout.decode()}") else: print("⚠️ Uploaded, but no unique_id returned to make the request.") 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 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_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: await asyncio.sleep(3600) continue 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"] 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 ) currents = [] recurrents = [] gold = [] current_time = time.time() for item in file_list: if item.get("type") == "dir": continue path = item.get("path") mtime = item.get("mtime", current_time) 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 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) 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." ) 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() conn, cursor = connect_db() define_tables(conn, cursor) youtube = get_authenticated_service() client = await get_client(conn, cursor, youtube) sync_token = load_sync_token() 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) if __name__ == "__main__": asyncio.run(main())