Playlist rotation automation, fingers crossed.
This commit is contained in:
@@ -82,7 +82,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 +97,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 +110,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 +153,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),
|
||||
)
|
||||
@@ -190,13 +187,11 @@ def get_video_info(youtube, video_id):
|
||||
try:
|
||||
video_details = youtube.videos().list(id=video_id, part="snippet").execute()
|
||||
|
||||
# Check if the video actually exists/is accessible
|
||||
if not video_details.get("items"):
|
||||
return False, "[Video unavailable or private]", ""
|
||||
|
||||
snippet = video_details["items"][0]["snippet"]
|
||||
|
||||
# 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]")
|
||||
@@ -224,11 +219,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 +233,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 +272,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 +299,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)
|
||||
|
||||
# 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 +322,11 @@ 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:
|
||||
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:
|
||||
@@ -426,19 +410,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,19 +427,18 @@ async def process_radio_track(video_link, video_id, title):
|
||||
if not AZURACAST_API_KEY:
|
||||
return
|
||||
|
||||
# Sanitize title for the local filesystem
|
||||
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}")
|
||||
|
||||
# 1. Download with yt-dlp
|
||||
dl_cmd = [
|
||||
"yt-dlp",
|
||||
"--no-cache-dir",
|
||||
"--embed-metadata",
|
||||
"--extract-audio",
|
||||
"--embed-metadata",
|
||||
"--output",
|
||||
f"{base_filepath}.%(ext)s",
|
||||
video_link,
|
||||
@@ -472,7 +451,6 @@ async def process_radio_track(video_link, video_id, title):
|
||||
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:
|
||||
@@ -484,13 +462,12 @@ async def process_radio_track(video_link, video_id, title):
|
||||
|
||||
print(f"📻 Uploading to AzuraCast: {title}")
|
||||
|
||||
# 2. Upload to AzuraCast using curl with the correct multipart endpoint
|
||||
upload_cmd = [
|
||||
"curl",
|
||||
"-s",
|
||||
"-X",
|
||||
"POST",
|
||||
f"{AZURACAST_URL}/api/station/{AZURACAST_STATION_ID}/files/upload",
|
||||
f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/upload",
|
||||
"-H",
|
||||
f"X-API-Key: {AZURACAST_API_KEY}",
|
||||
"-F",
|
||||
@@ -504,18 +481,16 @@ async def process_radio_track(video_link, video_id, title):
|
||||
)
|
||||
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 the /upload endpoint doesn't return the ID directly, fetch it via search
|
||||
if not unique_id:
|
||||
search_cmd = [
|
||||
"curl",
|
||||
"-s",
|
||||
"-G",
|
||||
f"{AZURACAST_URL}/api/station/{AZURACAST_STATION_ID}/files",
|
||||
f"{base_url}/api/station/{AZURACAST_STATION_ID}/files",
|
||||
"--data-urlencode",
|
||||
f"search={filename}",
|
||||
"-H",
|
||||
@@ -544,7 +519,7 @@ async def process_radio_track(video_link, video_id, title):
|
||||
"-s",
|
||||
"-X",
|
||||
"POST",
|
||||
f"{AZURACAST_URL}/api/station/{AZURACAST_STATION_ID}/request/{unique_id}",
|
||||
f"{base_url}/api/station/{AZURACAST_STATION_ID}/request/{unique_id}",
|
||||
"-H",
|
||||
f"X-API-Key: {AZURACAST_API_KEY}",
|
||||
]
|
||||
@@ -556,15 +531,208 @@ async def process_radio_track(video_link, video_id, title):
|
||||
await req_proc.communicate()
|
||||
print(f"✅ Queued on radio: {title}")
|
||||
else:
|
||||
print(f"⚠️ Uploaded, but no unique_id returned. Response: {stdout.decode()}")
|
||||
print(f"⚠️ Uploaded, but no unique_id returned.")
|
||||
except json.JSONDecodeError:
|
||||
print(f"❌ Failed to parse AzuraCast response: {stdout.decode()}")
|
||||
print(f"❌ Failed to parse AzuraCast response")
|
||||
|
||||
# Add the track to the "Currents" playlist immediately
|
||||
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": [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': {title}")
|
||||
except Exception as e:
|
||||
print(f"❌ Failed to auto-assign playlist for {title}: {e}")
|
||||
|
||||
# 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 a day."""
|
||||
while True:
|
||||
try:
|
||||
if not AZURACAST_API_KEY:
|
||||
await asyncio.sleep(86400)
|
||||
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(86400)
|
||||
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(86400)
|
||||
continue
|
||||
|
||||
currents_id = pl_map["currents"]
|
||||
recurrents_id = pl_map["recurrents"]
|
||||
gold_id = pl_map["gold"]
|
||||
|
||||
files_cmd = [
|
||||
"curl",
|
||||
"-s",
|
||||
"-X",
|
||||
"GET",
|
||||
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(86400)
|
||||
|
||||
|
||||
async def main():
|
||||
"""Get DB and Matrix client ready, and start syncing."""
|
||||
args = parse_arguments()
|
||||
@@ -574,12 +742,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)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user