361 lines
16 KiB
Python
361 lines
16 KiB
Python
import os
|
|
import sys
|
|
import random
|
|
import time
|
|
import json
|
|
import sqlite3
|
|
import hashlib
|
|
import requests
|
|
from google import genai
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
# --- CONFIGURATION ---
|
|
GONIC_URL = os.getenv("GONIC_URL", "http://gonic:8080/rest")
|
|
GONIC_USER = os.getenv("GONIC_USER")
|
|
GONIC_PASS = os.getenv("GONIC_PASS")
|
|
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
|
|
CURATION_COOLDOWN = 7 * 86400 # Re-curate existing playlists every 7 days for fresh rotation
|
|
|
|
DB_DIR = os.getenv("DB_DIR", ".")
|
|
DB_PATH = os.path.join(DB_DIR, "library.db")
|
|
HASH_PATH = os.path.join(DB_DIR, "library.hash")
|
|
|
|
# Declare your AI playlists and prompts here.
|
|
# The script will automatically create and own them under the script's user credentials.
|
|
AI_PLAYLIST_DEFINITIONS = [
|
|
{
|
|
"name": "AI: Christmas Bangers",
|
|
"prompt": "Upbeat Christmas classics and energetic winter anthems to get things moving. STRICT EXCLUSIONS: No traditional classical orchestral pieces or solo choral carols (keep those for the classical playlist)."
|
|
},
|
|
{
|
|
"name": "AI: Reflective Misery",
|
|
"prompt": "Melancholic, atmospheric tracks with deep emotional weight, great for late-night listening. STRICT EXCLUSIONS: Strictly skip classical tracks, acoustic chamber music, and long orchestral suites—focus entirely on modern alternative, dark indie, and sad electronic/pop."
|
|
},
|
|
{
|
|
"name": "AI: Classical Christmas",
|
|
"prompt": "Classical Christmas orchestral arrangements, traditional choral carols, solemn holiday brass, and festive symphonic masterpieces."
|
|
},
|
|
{
|
|
"name": "AI: Pure Dopamine",
|
|
"prompt": "Unapologetically joyful, high-energy, uplifting tracks with infectious grooves, bright melodies, and an undeniable feel-good factor to instantly boost the mood. STRICT EXCLUSIONS: Exclude classical compositions, slow movements, and traditional orchestral pieces."
|
|
},
|
|
{
|
|
"name": "AI: Midnight Transit",
|
|
"prompt": "Hypnotic, rhythm-driven electronic, dark synth, driving indie rock, and propulsive bass-heavy tracks built for moving through the dark. Features a steady forward momentum, cinematic atmospheres, and immersive nocturnal energy. STRICT EXCLUSIONS: Exclude traditional classical/orchestral pieces, slow ambient tracks with no percussion, high-tempo EDM club bangers, and generic acoustic pop."
|
|
},
|
|
{
|
|
"name": "AI: Halloween & Horror",
|
|
"prompt": "Spooky, eerie, and Halloween-party-ready tracks with high nocturnal energy. Includes classic Halloween bops (e.g. Thriller, Time Warp), dark pop (e.g. Disturbia), creepy alt-rock with supernatural themes (e.g. Zombie), and punchy, dramatic classical pieces or dark film scores (standalone tone poems up to 8 minutes like Danse Macabre or Fossils). STRICT EXCLUSIONS: Exclude multi-movement symphonies, extended concertos, or suites over 8 minutes (no 15+ minute works like Totentanz), standard emo/pop-punk songs (no blink-182), and generic alt-rock songs that merely have dark titles without an authentic Halloween or supernatural vibe."
|
|
},
|
|
{
|
|
"name": "AI: Lazy Sunday Morning",
|
|
"prompt": "Warm, acoustic, gentle indie folk and stripped-back singer-songwriter tracks with an organic, cozy, and laid-back feel. Perfect for a slow weekend morning. STRICT EXCLUSIONS: Exclude heavy percussion, electronic club beats, aggressive rock, and complex orchestral classical music."
|
|
},
|
|
{
|
|
"name": "AI: Deep Focus Lab",
|
|
"prompt": "Hypnotic, steady-tempo instrumental electronic, ambient techno, IDM, and minimal synth beats designed for deep concentration, coding, and technical workflows. STRICT EXCLUSIONS: Absolutely zero vocals or spoken word, no acoustic singer-songwriter tracks, and no classical or orchestral arrangements."
|
|
}
|
|
]
|
|
|
|
jitter = random.randint(0, 300)
|
|
print(f"Jitter active: sleeping for {jitter} seconds...", flush=True)
|
|
time.sleep(jitter)
|
|
|
|
# Initialize Gemini Client
|
|
gemini_client = genai.Client(api_key=GEMINI_API_KEY)
|
|
|
|
def get_auth_params():
|
|
return {
|
|
"u": GONIC_USER,
|
|
"p": GONIC_PASS,
|
|
"v": "1.16.1",
|
|
"c": "AI-Playlist-Agent",
|
|
"f": "json"
|
|
}
|
|
|
|
def init_db():
|
|
if DB_DIR and DB_DIR != ".":
|
|
os.makedirs(DB_DIR, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS tracks (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT,
|
|
artist TEXT,
|
|
album TEXT,
|
|
genre TEXT,
|
|
year INTEGER
|
|
)
|
|
""")
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS playlist_state (
|
|
name TEXT PRIMARY KEY,
|
|
last_curated INTEGER
|
|
)
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def sync_library_to_sqlite():
|
|
print("Fetching full track catalog from Gonic API...", flush=True)
|
|
auth = get_auth_params()
|
|
|
|
res = requests.get(f"{GONIC_URL}/getArtists.view", params=auth).json()
|
|
indexes = res.get("subsonic-response", {}).get("artists", {}).get("index", [])
|
|
|
|
all_tracks = []
|
|
for index_group in indexes:
|
|
for artist in index_group.get("artist", []):
|
|
artist_id = artist["id"]
|
|
artist_res = requests.get(f"{GONIC_URL}/getArtist.view", params={**auth, "id": artist_id}).json()
|
|
albums = artist_res.get("subsonic-response", {}).get("artist", {}).get("album", [])
|
|
|
|
for album in albums:
|
|
album_id = album["id"]
|
|
album_res = requests.get(f"{GONIC_URL}/getAlbum.view", params={**auth, "id": album_id}).json()
|
|
songs = album_res.get("subsonic-response", {}).get("album", {}).get("song", [])
|
|
|
|
for song in songs:
|
|
all_tracks.append((
|
|
str(song.get("id")),
|
|
song.get("title", "Unknown"),
|
|
song.get("artist", "Unknown"),
|
|
song.get("album", "Unknown"),
|
|
song.get("genre", "Unknown"),
|
|
song.get("year", 0)
|
|
))
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM tracks") # Full refresh
|
|
cursor.executemany("INSERT OR REPLACE INTO tracks VALUES (?, ?, ?, ?, ?, ?)", all_tracks)
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"Synced {len(all_tracks)} tracks to local SQLite index.", flush=True)
|
|
|
|
def check_library_changed():
|
|
auth = get_auth_params()
|
|
res = requests.get(f"{GONIC_URL}/getIndexes.view", params=auth).json()
|
|
indexes_data = str(res.get("subsonic-response", {}).get("indexes", {}))
|
|
current_hash = hashlib.md5(indexes_data.encode()).hexdigest()
|
|
|
|
if os.path.exists(HASH_PATH):
|
|
with open(HASH_PATH, "r") as f:
|
|
stored_hash = f.read()
|
|
if stored_hash == current_hash:
|
|
return False
|
|
|
|
with open(HASH_PATH, "w") as f:
|
|
f.write(current_hash)
|
|
return True
|
|
|
|
def query_candidate_tracks(prompt):
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
|
|
words = [w for w in prompt.lower().split() if len(w) > 3]
|
|
rows = []
|
|
|
|
# 1. Grab a focused subset of keyword matches (capped at 200 so it doesn't drown out the random pool)
|
|
if words:
|
|
query_clauses = " OR ".join([f"genre LIKE ? OR title LIKE ? OR artist LIKE ? OR album LIKE ?" for _ in words])
|
|
params = []
|
|
for w in words:
|
|
params.extend([f"%{w}%", f"%{w}%", f"%{w}%", f"%{w}%"])
|
|
|
|
cursor.execute(f"SELECT id, title, artist, album, genre, year FROM tracks WHERE {query_clauses} LIMIT 200", params)
|
|
rows = cursor.fetchall()
|
|
|
|
# 2. Pad aggressively with random tracks to reach a target pool of 350
|
|
# This ensures high serendipity and unearths unexpected gems from across the library.
|
|
target_pool_size = 350
|
|
existing_ids = {r[0] for r in rows}
|
|
|
|
if len(rows) < target_pool_size:
|
|
needed = target_pool_size - len(rows)
|
|
cursor.execute("SELECT id, title, artist, album, genre, year FROM tracks ORDER BY RANDOM() LIMIT ?", (needed,))
|
|
extra_rows = cursor.fetchall()
|
|
|
|
for r in extra_rows:
|
|
if r[0] not in existing_ids:
|
|
rows.append(r)
|
|
existing_ids.add(r[0])
|
|
|
|
conn.close()
|
|
|
|
return [{"id": r[0], "title": r[1], "artist": r[2], "album": r[3], "genre": r[4], "year": r[5]} for r in rows]
|
|
|
|
def ask_gemini_for_tracks(prompt, candidates):
|
|
system_instruction = (
|
|
"You are an expert music archivist, tastemaker, and DJ. "
|
|
"Your task is to curate tracks from a candidate pool that match a given thematic prompt or vibe. "
|
|
"CRITICAL INSTRUCTIONS:\n"
|
|
"1. Leverage your deep internal knowledge of music, artist styles, moods, and instrumentation. Do not rely solely on database tags or genres, as metadata can be messy or absent; use your understanding of the artist and song.\n"
|
|
"2. Prioritize absolute stylistic and emotional fit. Quality and vibe alignment matter far more than quantity: do not pad the selection with weak or unrelated matches.\n"
|
|
"3. You have complete freedom over the count: select anywhere from zero up to every single track that genuinely fits the criteria. There are no quotas, minimums, or maximum limits.\n"
|
|
"4. Return ONLY a raw JSON array containing the string IDs of the chosen tracks.\n"
|
|
"5. Output NO markdown, NO code blocks (do not use ```json ... ```), and NO conversational text whatsoever."
|
|
)
|
|
|
|
user_content = f"Prompt/Theme: {prompt}\n\nCandidate Tracks:\n{json.dumps(candidates, indent=2)}"
|
|
|
|
max_retries = 10
|
|
delay = 5
|
|
response = None
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
response = gemini_client.models.generate_content(
|
|
model="gemini-3.8-flash",
|
|
contents=user_content,
|
|
config=dict(
|
|
system_instruction=system_instruction,
|
|
temperature=0.3,
|
|
response_mime_type="application/json"
|
|
)
|
|
)
|
|
break
|
|
except Exception as e:
|
|
if "503" in str(e) or "UNAVAILABLE" in str(e):
|
|
if attempt < max_retries - 1:
|
|
print(f"Gemini capacity spike (503). Retrying in {delay}s (Attempt {attempt + 1}/{max_retries})...", flush=True)
|
|
time.sleep(delay)
|
|
delay *= 2
|
|
continue
|
|
raise e
|
|
|
|
if not response:
|
|
return []
|
|
|
|
try:
|
|
return json.loads(response.text)
|
|
except Exception as e:
|
|
print(f"Error parsing Gemini response: {e}\nRaw output: {response.text}", flush=True)
|
|
return []
|
|
|
|
def ensure_and_process_ai_playlists(library_has_changed):
|
|
auth = get_auth_params()
|
|
now = int(time.time())
|
|
|
|
# Fetch existing playlists owned by our script user
|
|
res = requests.get(f"{GONIC_URL}/getPlaylists.view", params=auth).json()
|
|
existing_playlists = {
|
|
pl.get("name"): pl.get("id")
|
|
for pl in res.get("subsonic-response", {}).get("playlists", {}).get("playlist", [])
|
|
}
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
|
|
for defn in AI_PLAYLIST_DEFINITIONS:
|
|
name = defn["name"]
|
|
prompt = defn["prompt"]
|
|
playlist_id = existing_playlists.get(name)
|
|
|
|
# 1. Create playlist automatically if it doesn't exist for this user account
|
|
if not playlist_id:
|
|
print(f"Playlist '{name}' not found for user. Creating it...", flush=True)
|
|
requests.get(f"{GONIC_URL}/createPlaylist.view", params={
|
|
**auth,
|
|
"name": name,
|
|
"public": "true"
|
|
})
|
|
|
|
# Re-fetch to grab the newly generated ID
|
|
res = requests.get(f"{GONIC_URL}/getPlaylists.view", params=auth).json()
|
|
for pl in res.get("subsonic-response", {}).get("playlists", {}).get("playlist", []):
|
|
if pl.get("name") == name:
|
|
playlist_id = pl.get("id")
|
|
break
|
|
|
|
if not playlist_id:
|
|
print(f"Failed to create or locate playlist '{name}'. Skipping.", flush=True)
|
|
continue
|
|
|
|
# 2. Fetch current playlist contents from Gonic
|
|
pl_detail_res = requests.get(f"{GONIC_URL}/getPlaylist.view", params={**auth, "id": playlist_id}).json()
|
|
existing_entries = pl_detail_res.get("subsonic-response", {}).get("playlist", {}).get("entry", [])
|
|
|
|
# 3. Check last curation timestamp from SQLite state table
|
|
cursor.execute("SELECT last_curated FROM playlist_state WHERE name = ?", (name,))
|
|
row = cursor.fetchone()
|
|
last_curated = row[0] if row else 0
|
|
|
|
# Decision Matrix for running Gemini:
|
|
# - Run if empty/new
|
|
# - Run if weekly cooldown (7 days) has expired (periodic refresh for variety)
|
|
# - Run if library changed AND it's been at least 24 hours since last curation
|
|
is_empty = not existing_entries
|
|
cooldown_expired = (now - last_curated) > CURATION_COOLDOWN
|
|
library_refresh_due = library_has_changed and (now - last_curated) > 86400
|
|
|
|
if not (is_empty or cooldown_expired or library_refresh_due):
|
|
print(f"Playlist '{name}' is up to date (Last curated {(now - last_curated) // 3600}h ago). Skipping Gemini.", flush=True)
|
|
continue
|
|
|
|
reason = "Empty/New" if is_empty else ("Periodic refresh" if cooldown_expired else "Library update")
|
|
print(f"Processing AI Playlist ({reason}): '{name}' with prompt: '{prompt}'", flush=True)
|
|
|
|
candidates = query_candidate_tracks(prompt)
|
|
if not candidates:
|
|
print("No matching candidate tracks found in local index.", flush=True)
|
|
continue
|
|
|
|
selected_ids = ask_gemini_for_tracks(prompt, candidates)
|
|
print(f"Gemini selected {len(selected_ids)} tracks.", flush=True)
|
|
|
|
if selected_ids:
|
|
existing_song_ids = {str(song.get("id")) for song in existing_entries}
|
|
tracks_to_add = [sid for sid in selected_ids if sid not in existing_song_ids]
|
|
|
|
if tracks_to_add:
|
|
query_list = [
|
|
("u", auth["u"]), ("p", auth["p"]), ("v", auth["v"]),
|
|
("c", auth["c"]), ("f", auth["f"]),
|
|
("playlistId", playlist_id), ("name", name),
|
|
("comment", prompt),
|
|
("public", "true")
|
|
]
|
|
for sid in tracks_to_add:
|
|
query_list.append(("songIdToAdd", sid))
|
|
|
|
update_res = requests.get(f"{GONIC_URL}/updatePlaylist.view", params=query_list).json()
|
|
if update_res.get("subsonic-response", {}).get("status") == "ok":
|
|
print(f"Successfully added {len(tracks_to_add)} new tracks to '{name}'!", flush=True)
|
|
else:
|
|
print(f"Failed to update playlist: {update_res}")
|
|
else:
|
|
print(f"Playlist '{name}' has no new unique tracks to add from this selection.", flush=True)
|
|
|
|
# Update last curated timestamp in SQLite state table
|
|
cursor.execute("INSERT OR REPLACE INTO playlist_state (name, last_curated) VALUES (?, ?)", (name, now))
|
|
conn.commit()
|
|
|
|
conn.close()
|
|
|
|
def main():
|
|
init_db()
|
|
print("AI Playlist Sync one-shot execution started.", flush=True)
|
|
|
|
try:
|
|
# 1. Sync library catalog only if changes are detected or local DB is missing
|
|
lib_changed = check_library_changed()
|
|
if lib_changed or not os.path.exists(DB_PATH):
|
|
print("Library changes detected (or missing local index). Updating local SQLite index...", flush=True)
|
|
sync_library_to_sqlite()
|
|
else:
|
|
print("Library catalog unchanged. Using existing local SQLite index.", flush=True)
|
|
|
|
# 2. Check and process playlists using the time-based curation cooldown matrix
|
|
ensure_and_process_ai_playlists(lib_changed)
|
|
print("AI Playlist Sync execution completed successfully.", flush=True)
|
|
|
|
except Exception as e:
|
|
print(f"Error during execution: {e}", flush=True)
|
|
raise e
|
|
|
|
if __name__ == "__main__":
|
|
main() |