Caine
c7956ae9b2
Overnight transcoding, music discovery/import, system health reports, stats page generator, and bookmark management. Secrets stored in /etc/automation/ — not in repo.
1039 lines
40 KiB
Python
Executable File
1039 lines
40 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Music Recommender - Finds music you might like based on Last.fm history,
|
|
cross-referenced with your Navidrome collection. Can auto-download via slskd.
|
|
|
|
Usage:
|
|
python music_recommender.py recommend # Get recommendations
|
|
python music_recommender.py gaps # Find library gaps
|
|
python music_recommender.py recent # Recent scrobbles
|
|
python music_recommender.py search "query" # Search Navidrome
|
|
python music_recommender.py surprise # Pick & download an album overnight
|
|
python music_recommender.py slskd "query" # Search Soulseek
|
|
python music_recommender.py download "artist - album" # Download from Soulseek
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import string
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import urlencode
|
|
|
|
import requests
|
|
|
|
# Default config path
|
|
CONFIG_PATH = Path(__file__).parent / "music_config.json"
|
|
STATE_PATH = Path(__file__).parent / "music_state.json"
|
|
|
|
|
|
class LastFM:
|
|
"""Last.fm API client."""
|
|
|
|
BASE_URL = "https://ws.audioscrobbler.com/2.0/"
|
|
|
|
def __init__(self, api_key: str, username: str):
|
|
self.api_key = api_key
|
|
self.username = username
|
|
|
|
def _request(self, method: str, **params) -> dict:
|
|
params.update({
|
|
"method": method,
|
|
"api_key": self.api_key,
|
|
"format": "json",
|
|
})
|
|
resp = requests.get(self.BASE_URL, params=params)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def get_recent_tracks(self, limit: int = 50) -> list[dict]:
|
|
"""Get recently played tracks."""
|
|
data = self._request("user.getrecenttracks", user=self.username, limit=limit)
|
|
tracks = data.get("recenttracks", {}).get("track", [])
|
|
return [
|
|
{
|
|
"artist": t.get("artist", {}).get("#text", ""),
|
|
"album": t.get("album", {}).get("#text", ""),
|
|
"track": t.get("name", ""),
|
|
"now_playing": "@attr" in t and t["@attr"].get("nowplaying") == "true",
|
|
}
|
|
for t in tracks
|
|
]
|
|
|
|
def get_top_artists(self, period: str = "3month", limit: int = 50) -> list[dict]:
|
|
"""Get top artists. Period: overall, 7day, 1month, 3month, 6month, 12month."""
|
|
data = self._request("user.gettopartists", user=self.username, period=period, limit=limit)
|
|
artists = data.get("topartists", {}).get("artist", [])
|
|
return [
|
|
{
|
|
"name": a.get("name", ""),
|
|
"playcount": int(a.get("playcount", 0)),
|
|
"url": a.get("url", ""),
|
|
}
|
|
for a in artists
|
|
]
|
|
|
|
def get_similar_artists(self, artist: str, limit: int = 10) -> list[dict]:
|
|
"""Get artists similar to the given artist."""
|
|
try:
|
|
data = self._request("artist.getsimilar", artist=artist, limit=limit)
|
|
similar = data.get("similarartists", {}).get("artist", [])
|
|
return [
|
|
{
|
|
"name": a.get("name", ""),
|
|
"match": float(a.get("match", 0)),
|
|
"url": a.get("url", ""),
|
|
}
|
|
for a in similar
|
|
]
|
|
except Exception:
|
|
return []
|
|
|
|
def get_artist_top_albums(self, artist: str, limit: int = 5) -> list[dict]:
|
|
"""Get top albums for an artist."""
|
|
try:
|
|
data = self._request("artist.gettopalbums", artist=artist, limit=limit)
|
|
albums = data.get("topalbums", {}).get("album", [])
|
|
return [
|
|
{
|
|
"name": a.get("name", ""),
|
|
"artist": artist,
|
|
"playcount": int(a.get("playcount", 0)),
|
|
"url": a.get("url", ""),
|
|
}
|
|
for a in albums
|
|
if a.get("name") and a.get("name") != "(null)"
|
|
]
|
|
except Exception:
|
|
return []
|
|
|
|
def get_recommendations(self, limit: int = 20) -> list[dict]:
|
|
"""Get recommended artists based on listening history."""
|
|
top = self.get_top_artists(period="3month", limit=10)
|
|
|
|
recommendations = {}
|
|
for artist in top:
|
|
similar = self.get_similar_artists(artist["name"], limit=5)
|
|
for s in similar:
|
|
name = s["name"]
|
|
if name not in recommendations and name not in [a["name"] for a in top]:
|
|
recommendations[name] = {
|
|
"name": name,
|
|
"because_of": artist["name"],
|
|
"match": s["match"],
|
|
}
|
|
|
|
sorted_recs = sorted(recommendations.values(), key=lambda x: x["match"], reverse=True)
|
|
return sorted_recs[:limit]
|
|
|
|
|
|
class Navidrome:
|
|
"""Navidrome/Subsonic API client."""
|
|
|
|
def __init__(self, url: str, username: str, password: str):
|
|
self.base_url = url.rstrip("/")
|
|
self.username = username
|
|
self.password = password
|
|
|
|
def _auth_params(self) -> dict:
|
|
salt = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
|
|
token = hashlib.md5((self.password + salt).encode()).hexdigest()
|
|
return {
|
|
"u": self.username,
|
|
"t": token,
|
|
"s": salt,
|
|
"v": "1.16.1",
|
|
"c": "music_recommender",
|
|
"f": "json",
|
|
}
|
|
|
|
def _request(self, endpoint: str, **params) -> dict:
|
|
params.update(self._auth_params())
|
|
url = f"{self.base_url}/rest/{endpoint}"
|
|
resp = requests.get(url, params=params)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
sr = data.get("subsonic-response", {})
|
|
if sr.get("status") == "failed":
|
|
error = sr.get("error", {})
|
|
raise Exception(f"Subsonic error {error.get('code')}: {error.get('message')}")
|
|
|
|
return sr
|
|
|
|
def ping(self) -> bool:
|
|
try:
|
|
self._request("ping")
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def get_artists(self) -> list[dict]:
|
|
data = self._request("getArtists")
|
|
artists = []
|
|
for index in data.get("artists", {}).get("index", []):
|
|
for artist in index.get("artist", []):
|
|
artists.append({
|
|
"id": artist.get("id"),
|
|
"name": artist.get("name"),
|
|
"album_count": artist.get("albumCount", 0),
|
|
})
|
|
return artists
|
|
|
|
def get_artist(self, artist_id: str) -> dict:
|
|
data = self._request("getArtist", id=artist_id)
|
|
artist = data.get("artist", {})
|
|
return {
|
|
"id": artist.get("id"),
|
|
"name": artist.get("name"),
|
|
"album_count": artist.get("albumCount", 0),
|
|
"albums": [
|
|
{
|
|
"id": a.get("id"),
|
|
"name": a.get("name"),
|
|
"year": a.get("year"),
|
|
"song_count": a.get("songCount", 0),
|
|
}
|
|
for a in artist.get("album", [])
|
|
],
|
|
}
|
|
|
|
def search(self, query: str) -> dict:
|
|
data = self._request("search3", query=query)
|
|
result = data.get("searchResult3", {})
|
|
return {
|
|
"artists": [
|
|
{"id": a.get("id"), "name": a.get("name")}
|
|
for a in result.get("artist", [])
|
|
],
|
|
"albums": [
|
|
{"id": a.get("id"), "name": a.get("name"), "artist": a.get("artist")}
|
|
for a in result.get("album", [])
|
|
],
|
|
"songs": [
|
|
{"id": s.get("id"), "title": s.get("title"), "artist": s.get("artist"), "album": s.get("album")}
|
|
for s in result.get("song", [])
|
|
],
|
|
}
|
|
|
|
|
|
class Slskd:
|
|
"""slskd API client for Soulseek searches and downloads."""
|
|
|
|
def __init__(self, url: str, api_key: str):
|
|
self.base_url = url.rstrip("/")
|
|
self.api_key = api_key
|
|
self.headers = {"X-API-Key": api_key}
|
|
|
|
def _request(self, method: str, endpoint: str, **kwargs) -> dict:
|
|
url = f"{self.base_url}/api/v0/{endpoint}"
|
|
kwargs.setdefault("headers", {}).update(self.headers)
|
|
resp = requests.request(method, url, **kwargs)
|
|
resp.raise_for_status()
|
|
if resp.content:
|
|
return resp.json()
|
|
return {}
|
|
|
|
def search(self, query: str, timeout: int = 30) -> str:
|
|
"""Start a search and return the search ID."""
|
|
data = self._request("POST", "searches", json={"searchText": query})
|
|
return data.get("id")
|
|
|
|
def get_search_state(self, search_id: str) -> dict:
|
|
"""Get search state."""
|
|
return self._request("GET", f"searches/{search_id}")
|
|
|
|
def wait_for_search(self, search_id: str, timeout: int = 60, min_results: int = 5, min_wait: int = 30) -> bool:
|
|
"""Wait for search to complete or have enough results.
|
|
|
|
Args:
|
|
search_id: The search ID to wait for
|
|
timeout: Maximum time to wait
|
|
min_results: Return early if we have this many files
|
|
min_wait: Minimum time to wait before honoring isComplete with 0 results
|
|
"""
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
state = self.get_search_state(search_id)
|
|
file_count = state.get("fileCount", 0)
|
|
elapsed = time.time() - start
|
|
|
|
# Got enough results - done
|
|
if file_count >= min_results:
|
|
return True
|
|
|
|
# Search complete with results - done
|
|
if state.get("isComplete") and file_count > 0:
|
|
return True
|
|
|
|
# Search complete with NO results - only trust after min_wait
|
|
# (Soulseek can mark complete before results arrive)
|
|
if state.get("isComplete") and elapsed >= min_wait:
|
|
return True
|
|
|
|
time.sleep(2)
|
|
return False
|
|
|
|
def get_search_responses(self, search_id: str, max_retries: int = 15, retry_delay: float = 2) -> list[dict]:
|
|
"""Get search responses (results grouped by user).
|
|
|
|
Retries if response list is empty but state shows results exist.
|
|
slskd has a race condition where /responses can lag significantly
|
|
behind state.responseCount - sometimes 10+ seconds.
|
|
"""
|
|
for attempt in range(max_retries):
|
|
responses = self._request("GET", f"searches/{search_id}/responses")
|
|
if responses:
|
|
return responses
|
|
|
|
# Check if state says there should be responses
|
|
state = self.get_search_state(search_id)
|
|
if state.get("responseCount", 0) == 0:
|
|
# Genuinely no results
|
|
return []
|
|
|
|
# State has responses but endpoint returned empty - wait and retry
|
|
time.sleep(retry_delay)
|
|
|
|
return []
|
|
|
|
def download_files(self, username: str, files: list[dict]) -> dict:
|
|
"""Queue files for download from a user."""
|
|
return self._request("POST", f"transfers/downloads/{username}", json=files)
|
|
|
|
def get_downloads(self) -> list[dict]:
|
|
"""Get current downloads."""
|
|
return self._request("GET", "transfers/downloads")
|
|
|
|
def search_and_find_album(self, artist: str, album: str, timeout: int = 60) -> Optional[dict]:
|
|
"""Search for an album and return best result."""
|
|
query = f"{artist} {album}"
|
|
print(f" Searching Soulseek for: {query}")
|
|
|
|
search_id = self.search(query)
|
|
if not search_id:
|
|
return None
|
|
|
|
# Wait for results
|
|
self.wait_for_search(search_id, timeout=timeout, min_results=3)
|
|
|
|
responses = self.get_search_responses(search_id)
|
|
if not responses:
|
|
return None
|
|
|
|
# Score and rank results
|
|
best_match = None
|
|
best_score = 0
|
|
|
|
for response in responses:
|
|
username = response.get("username", "")
|
|
files = response.get("files", [])
|
|
|
|
# Skip users with no free slots
|
|
if not response.get("hasFreeUploadSlot", False):
|
|
continue
|
|
|
|
# Group files by directory
|
|
directories = {}
|
|
for f in files:
|
|
filename = f.get("filename", "")
|
|
# Extract directory
|
|
parts = filename.replace("\\", "/").rsplit("/", 1)
|
|
if len(parts) == 2:
|
|
dir_path, fname = parts
|
|
else:
|
|
dir_path, fname = "", parts[0]
|
|
|
|
if dir_path not in directories:
|
|
directories[dir_path] = []
|
|
directories[dir_path].append(f)
|
|
|
|
# Find directories that look like albums (multiple audio files)
|
|
for dir_path, dir_files in directories.items():
|
|
audio_files = [f for f in dir_files if any(
|
|
f.get("filename", "").lower().endswith(ext)
|
|
for ext in [".flac", ".mp3", ".m4a", ".ogg", ".opus"]
|
|
)]
|
|
|
|
if len(audio_files) < 3:
|
|
continue
|
|
|
|
# Score this directory
|
|
score = 0
|
|
dir_lower = dir_path.lower()
|
|
|
|
# Prefer FLAC
|
|
flac_count = sum(1 for f in audio_files if f.get("filename", "").lower().endswith(".flac"))
|
|
if flac_count == len(audio_files):
|
|
score += 100
|
|
elif flac_count > 0:
|
|
score += 50
|
|
|
|
# Check artist/album in path
|
|
if artist.lower() in dir_lower:
|
|
score += 30
|
|
if album.lower() in dir_lower:
|
|
score += 30
|
|
|
|
# Prefer more complete albums
|
|
score += min(len(audio_files), 20)
|
|
|
|
# Prefer faster users (higher upload speed)
|
|
score += min(response.get("uploadSpeed", 0) // 100000, 20)
|
|
|
|
if score > best_score:
|
|
best_score = score
|
|
best_match = {
|
|
"username": username,
|
|
"directory": dir_path,
|
|
"files": audio_files,
|
|
"file_count": len(audio_files),
|
|
"score": score,
|
|
"has_flac": flac_count > 0,
|
|
}
|
|
|
|
return best_match
|
|
|
|
def download_album(self, match: dict) -> bool:
|
|
"""Download an album match."""
|
|
if not match:
|
|
return False
|
|
|
|
username = match["username"]
|
|
files = match["files"]
|
|
|
|
# Format files for download API
|
|
download_files = [
|
|
{"filename": f["filename"], "size": f.get("size", 0)}
|
|
for f in files
|
|
]
|
|
|
|
try:
|
|
self.download_files(username, download_files)
|
|
return True
|
|
except Exception as e:
|
|
print(f" Download failed: {e}")
|
|
return False
|
|
|
|
def get_user_downloads(self, username: str) -> list[dict]:
|
|
"""Get downloads from a specific user."""
|
|
try:
|
|
all_downloads = self._request("GET", "transfers/downloads")
|
|
# Filter by username - strip invisible Unicode chars for robust matching
|
|
clean = lambda s: ''.join(c for c in (s or '') if c.isprintable())
|
|
target = clean(username)
|
|
return [d for d in all_downloads if clean(d.get("username", "")) == target]
|
|
except Exception:
|
|
return []
|
|
|
|
def cancel_user_downloads(self, username: str) -> bool:
|
|
"""Cancel all downloads from a user."""
|
|
try:
|
|
self._request("DELETE", f"transfers/downloads/{username}")
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def wait_for_download(self, username: str, expected_files: int,
|
|
timeout_minutes: int = 30, check_interval: int = 60) -> bool:
|
|
"""Wait for downloads from a user to complete."""
|
|
import time
|
|
|
|
start = time.time()
|
|
timeout_seconds = timeout_minutes * 60
|
|
|
|
while time.time() - start < timeout_seconds:
|
|
downloads = self.get_user_downloads(username)
|
|
|
|
if not downloads:
|
|
# No downloads found - might have completed and been removed
|
|
# Or never started
|
|
return False
|
|
|
|
# Check states
|
|
completed = sum(1 for d in downloads
|
|
for f in d.get("files", [])
|
|
if f.get("state", "").startswith("Completed"))
|
|
failed = sum(1 for d in downloads
|
|
for f in d.get("files", [])
|
|
if any(s in (f.get("state") or "") for s in ["Errored", "Cancelled", "TimedOut", "Rejected", "Aborted"]))
|
|
in_progress = sum(1 for d in downloads
|
|
for f in d.get("files", [])
|
|
if any(s in (f.get("state") or "") for s in ["Queued", "Initializing", "InProgress"]))
|
|
|
|
total = completed + failed + in_progress
|
|
|
|
print(f" Progress: {completed}/{total} complete, {failed} failed, {in_progress} in progress")
|
|
|
|
if completed >= expected_files:
|
|
return True
|
|
|
|
if failed > 0 and in_progress == 0:
|
|
# All either completed or failed, nothing in progress
|
|
return completed > 0
|
|
|
|
time.sleep(check_interval)
|
|
|
|
# Timeout
|
|
return False
|
|
|
|
|
|
class MusicRecommender:
|
|
"""Main recommender combining all sources."""
|
|
|
|
def __init__(self, config: dict):
|
|
self.config = config
|
|
|
|
self.lastfm = LastFM(
|
|
api_key=config["lastfm"]["api_key"],
|
|
username=config["lastfm"]["username"],
|
|
)
|
|
|
|
self.navidrome = None
|
|
if "navidrome" in config:
|
|
self.navidrome = Navidrome(
|
|
url=config["navidrome"]["url"],
|
|
username=config["navidrome"]["username"],
|
|
password=config["navidrome"]["password"],
|
|
)
|
|
|
|
self.slskd = None
|
|
if "slskd" in config:
|
|
self.slskd = Slskd(
|
|
url=config["slskd"]["url"],
|
|
api_key=config["slskd"]["api_key"],
|
|
)
|
|
|
|
def get_library_artists(self) -> set[str]:
|
|
if not self.navidrome:
|
|
return set()
|
|
artists = self.navidrome.get_artists()
|
|
return {a["name"].lower() for a in artists}
|
|
|
|
def get_library_albums(self) -> set[str]:
|
|
"""Get set of 'artist - album' strings in library."""
|
|
if not self.navidrome:
|
|
return set()
|
|
|
|
albums = set()
|
|
artists = self.navidrome.get_artists()
|
|
for artist in artists:
|
|
try:
|
|
details = self.navidrome.get_artist(artist["id"])
|
|
for album in details.get("albums", []):
|
|
key = f"{artist['name'].lower()} - {album['name'].lower()}"
|
|
albums.add(key)
|
|
except Exception:
|
|
pass
|
|
return albums
|
|
|
|
def find_gaps(self) -> list[dict]:
|
|
"""Find artists you listen to but don't have in your library."""
|
|
top_artists = self.lastfm.get_top_artists(period="6month", limit=50)
|
|
library = self.get_library_artists()
|
|
|
|
# Filter out YouTube auto-channels and other noise
|
|
noise_patterns = [" - topic", "official", "vevo", "cyber gardens"]
|
|
|
|
gaps = []
|
|
for artist in top_artists:
|
|
name_lower = artist["name"].lower()
|
|
|
|
# Skip noise
|
|
if any(p in name_lower for p in noise_patterns):
|
|
continue
|
|
|
|
if name_lower not in library:
|
|
gaps.append({
|
|
"name": artist["name"],
|
|
"playcount": artist["playcount"],
|
|
"reason": "listened to but not in library",
|
|
})
|
|
|
|
return gaps
|
|
|
|
def get_disliked_artists(self) -> set[str]:
|
|
"""Get set of disliked artist names (lowercase) from config feedback."""
|
|
disliked = set()
|
|
feedback = self.config.get("feedback", {})
|
|
for entry in feedback.get("disliked", []):
|
|
if "artist" in entry:
|
|
disliked.add(entry["artist"].lower())
|
|
return disliked
|
|
|
|
def get_recommendations(self) -> list[dict]:
|
|
recs = self.lastfm.get_recommendations(limit=30)
|
|
library = self.get_library_artists()
|
|
disliked = self.get_disliked_artists()
|
|
filtered = [r for r in recs if r["name"].lower() not in library and r["name"].lower() not in disliked]
|
|
return filtered[:20]
|
|
|
|
def pick_surprise_album(self) -> Optional[dict]:
|
|
"""Pick an album to download as a surprise."""
|
|
# Load state to avoid repeats
|
|
state = self._load_state()
|
|
downloaded = set(state.get("downloaded_albums", []))
|
|
|
|
# Strategy 1: Album from a gap artist (artist you listen to but don't own)
|
|
gaps = self.find_gaps()[:10]
|
|
|
|
# Strategy 2: Album from a recommended artist
|
|
recs = self.get_recommendations()[:10]
|
|
|
|
# Combine and shuffle
|
|
candidates = []
|
|
|
|
for gap in gaps:
|
|
albums = self.lastfm.get_artist_top_albums(gap["name"], limit=3)
|
|
for album in albums:
|
|
key = f"{gap['name'].lower()} - {album['name'].lower()}"
|
|
if key not in downloaded:
|
|
candidates.append({
|
|
"artist": gap["name"],
|
|
"album": album["name"],
|
|
"reason": f"You've played {gap['name']} {gap['playcount']} times but don't own this",
|
|
"priority": 2, # Higher priority for gaps
|
|
})
|
|
|
|
for rec in recs:
|
|
albums = self.lastfm.get_artist_top_albums(rec["name"], limit=2)
|
|
for album in albums:
|
|
key = f"{rec['name'].lower()} - {album['name'].lower()}"
|
|
if key not in downloaded:
|
|
candidates.append({
|
|
"artist": rec["name"],
|
|
"album": album["name"],
|
|
"reason": f"Similar to {rec['because_of']}",
|
|
"priority": 1,
|
|
})
|
|
|
|
if not candidates:
|
|
return None
|
|
|
|
# Sort by priority, then shuffle within priority
|
|
candidates.sort(key=lambda x: -x["priority"])
|
|
top_priority = candidates[0]["priority"]
|
|
top_candidates = [c for c in candidates if c["priority"] == top_priority]
|
|
|
|
return random.choice(top_candidates)
|
|
|
|
def surprise_download(self, max_attempts: int = 3, wait_for_completion: bool = False,
|
|
timeout_minutes: int = 30) -> dict:
|
|
"""Pick and download a surprise album. Tries multiple albums if needed.
|
|
|
|
Args:
|
|
max_attempts: Maximum albums to try
|
|
wait_for_completion: If True, wait for download to complete (for overnight job)
|
|
timeout_minutes: How long to wait for each download before giving up
|
|
"""
|
|
if not self.slskd:
|
|
return {"success": False, "error": "slskd not configured"}
|
|
|
|
# Get all candidates
|
|
state = self._load_state()
|
|
downloaded = set(state.get("downloaded_albums", []))
|
|
disliked = self.get_disliked_artists()
|
|
|
|
candidates = []
|
|
|
|
# From gaps (higher priority) - skip disliked artists
|
|
for gap in self.find_gaps()[:15]:
|
|
if gap["name"].lower() in disliked:
|
|
continue
|
|
albums = self.lastfm.get_artist_top_albums(gap["name"], limit=3)
|
|
for album in albums:
|
|
key = f"{gap['name'].lower()} - {album['name'].lower()}"
|
|
if key not in downloaded:
|
|
candidates.append({
|
|
"artist": gap["name"],
|
|
"album": album["name"],
|
|
"reason": f"You've played {gap['name']} {gap['playcount']} times but don't own this",
|
|
"priority": 2,
|
|
})
|
|
|
|
# From recommendations
|
|
for rec in self.get_recommendations()[:10]:
|
|
albums = self.lastfm.get_artist_top_albums(rec["name"], limit=2)
|
|
for album in albums:
|
|
key = f"{rec['name'].lower()} - {album['name'].lower()}"
|
|
if key not in downloaded:
|
|
candidates.append({
|
|
"artist": rec["name"],
|
|
"album": album["name"],
|
|
"reason": f"Similar to {rec['because_of']}",
|
|
"priority": 1,
|
|
})
|
|
|
|
# From Discogs labels (curated 90s labels)
|
|
discogs_path = Path(__file__).parent / "discogs_labels.json"
|
|
if discogs_path.exists():
|
|
try:
|
|
with open(discogs_path) as f:
|
|
discogs_data = json.load(f)
|
|
# Pick random releases from the pool
|
|
discogs_releases = discogs_data.get("releases", [])
|
|
random.shuffle(discogs_releases)
|
|
for release in discogs_releases[:20]:
|
|
if release['artist'].lower() in disliked:
|
|
continue
|
|
key = f"{release['artist'].lower()} - {release['album'].lower()}"
|
|
if key not in downloaded:
|
|
year_str = f" ({release['year']})" if release.get('year') else ""
|
|
candidates.append({
|
|
"artist": release["artist"],
|
|
"album": release["album"],
|
|
"reason": f"From {release['label']}{year_str}",
|
|
"priority": 1, # Same priority as recommendations
|
|
})
|
|
except Exception:
|
|
pass # Silently skip if file is corrupt
|
|
|
|
if not candidates:
|
|
return {"success": False, "error": "No suitable albums found"}
|
|
|
|
# Shuffle within priority groups
|
|
random.shuffle(candidates)
|
|
candidates.sort(key=lambda x: -x["priority"])
|
|
|
|
# Try up to max_attempts albums
|
|
tried = []
|
|
for pick in candidates[:max_attempts]:
|
|
print(f"\n🎲 Trying: {pick['artist']} - {pick['album']}")
|
|
print(f" Reason: {pick['reason']}\n")
|
|
|
|
match = self.slskd.search_and_find_album(pick["artist"], pick["album"])
|
|
|
|
if not match:
|
|
print(f" ❌ Not found on Soulseek, trying next...")
|
|
tried.append(f"{pick['artist']} - {pick['album']} (not found)")
|
|
continue
|
|
|
|
print(f" Found: {match['file_count']} files from {match['username']}")
|
|
print(f" FLAC: {'Yes' if match['has_flac'] else 'No'}")
|
|
|
|
if not self.slskd.download_album(match):
|
|
tried.append(f"{pick['artist']} - {pick['album']} (queue failed)")
|
|
continue
|
|
|
|
print(f" ✓ Download queued!")
|
|
|
|
# If not waiting for completion, return success after queuing
|
|
if not wait_for_completion:
|
|
self._save_downloaded(pick["artist"], pick["album"], pick.get("reason"))
|
|
return {
|
|
"success": True,
|
|
"artist": pick["artist"],
|
|
"album": pick["album"],
|
|
"reason": pick["reason"],
|
|
"files": match["file_count"],
|
|
"source": match["username"],
|
|
"has_flac": match["has_flac"],
|
|
"attempts": len(tried) + 1,
|
|
"status": "queued",
|
|
}
|
|
|
|
# Wait for download to complete
|
|
print(f" ⏳ Waiting up to {timeout_minutes} minutes for download...")
|
|
|
|
if self.slskd.wait_for_download(match["username"], match["file_count"],
|
|
timeout_minutes=timeout_minutes):
|
|
self._save_downloaded(pick["artist"], pick["album"], pick.get("reason"))
|
|
return {
|
|
"success": True,
|
|
"artist": pick["artist"],
|
|
"album": pick["album"],
|
|
"reason": pick["reason"],
|
|
"files": match["file_count"],
|
|
"source": match["username"],
|
|
"has_flac": match["has_flac"],
|
|
"attempts": len(tried) + 1,
|
|
"status": "completed",
|
|
}
|
|
else:
|
|
# Download timed out - cancel and try next
|
|
print(f" ⏱️ Timeout! Cancelling and trying next album...")
|
|
self.slskd.cancel_user_downloads(match["username"])
|
|
tried.append(f"{pick['artist']} - {pick['album']} (timeout after {timeout_minutes}min)")
|
|
continue
|
|
|
|
# Save failed attempts to wishlist for Discogs bookmarking
|
|
self._save_wishlist(tried)
|
|
|
|
return {
|
|
"success": False,
|
|
"error": f"Tried {len(tried)} albums, none worked",
|
|
"tried": tried,
|
|
}
|
|
|
|
def _save_wishlist(self, albums: list[str]):
|
|
"""Save albums that couldn't be found to wishlist + bookmarks."""
|
|
wishlist_path = Path(__file__).parent / "music_wishlist.md"
|
|
|
|
# Read existing wishlist
|
|
existing = set()
|
|
if wishlist_path.exists():
|
|
with open(wishlist_path) as f:
|
|
for line in f:
|
|
if line.startswith("- [ ] "):
|
|
existing.add(line.strip()[6:].split(" — ")[0])
|
|
|
|
# Add new entries to markdown wishlist
|
|
with open(wishlist_path, "a") as f:
|
|
if not wishlist_path.exists() or wishlist_path.stat().st_size == 0:
|
|
f.write("# Music Wishlist\n\n")
|
|
f.write("Albums that couldn't be found on Soulseek.\n")
|
|
f.write("Search Discogs/Bandcamp manually or check back later.\n\n")
|
|
|
|
for album in albums:
|
|
clean = album.replace(" (download failed)", "").replace(" (not found)", "").replace(" (queue failed)", "")
|
|
clean = clean.split(" (timeout")[0] # Remove timeout messages
|
|
if clean not in existing:
|
|
query = clean.replace(" - ", " ").replace(" ", "+")
|
|
discogs_url = f"https://www.discogs.com/search/?q={query}&type=release"
|
|
f.write(f"- [ ] {clean} — [Discogs]({discogs_url})\n")
|
|
|
|
# Also add to Floccus bookmarks
|
|
self._add_to_bookmarks(clean, discogs_url)
|
|
|
|
def _add_to_bookmarks(self, title: str, url: str):
|
|
"""Add album to Floccus bookmarks in Music/Wishlist folder."""
|
|
import subprocess
|
|
|
|
add_bookmark_script = Path(__file__).parent / "add_bookmark_to_wishlist.js"
|
|
if add_bookmark_script.exists():
|
|
try:
|
|
subprocess.run(
|
|
["node", str(add_bookmark_script), url, title],
|
|
capture_output=True,
|
|
timeout=30
|
|
)
|
|
print(f" 📌 Added to bookmarks: {title}")
|
|
except Exception as e:
|
|
print(f" ⚠ Bookmark add failed: {e}")
|
|
|
|
def _load_state(self) -> dict:
|
|
if STATE_PATH.exists():
|
|
with open(STATE_PATH) as f:
|
|
return json.load(f)
|
|
return {"downloaded_albums": []}
|
|
|
|
def _save_downloaded(self, artist: str, album: str, reason: str = None):
|
|
state = self._load_state()
|
|
key = f"{artist.lower()} - {album.lower()}"
|
|
if key not in state["downloaded_albums"]:
|
|
state["downloaded_albums"].append(key)
|
|
# Keep last 100
|
|
state["downloaded_albums"] = state["downloaded_albums"][-100:]
|
|
state["last_download"] = {
|
|
"artist": artist,
|
|
"album": album,
|
|
"reason": reason,
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
with open(STATE_PATH, "w") as f:
|
|
json.dump(state, f, indent=2)
|
|
|
|
def format_recommendations(self, recs: list[dict]) -> str:
|
|
lines = ["# Music Recommendations\n"]
|
|
lines.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n")
|
|
|
|
for i, rec in enumerate(recs, 1):
|
|
lines.append(f"{i}. **{rec['name']}**")
|
|
if "because_of" in rec:
|
|
lines.append(f" Similar to: {rec['because_of']}")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def format_gaps(self, gaps: list[dict]) -> str:
|
|
lines = ["# Library Gaps\n"]
|
|
lines.append("Artists you listen to but don't have in Navidrome:\n")
|
|
|
|
for i, gap in enumerate(gaps, 1):
|
|
lines.append(f"{i}. **{gap['name']}** ({gap['playcount']} plays)")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def load_config(path: Path) -> dict:
|
|
if not path.exists():
|
|
print(f"Config not found: {path}")
|
|
print("Create it with your API keys. See music_config.example.json")
|
|
sys.exit(1)
|
|
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Music recommendation tool")
|
|
parser.add_argument("command", nargs="?", default="recommend",
|
|
choices=["recommend", "gaps", "recent", "search", "test",
|
|
"surprise", "slskd", "download"],
|
|
help="Command to run")
|
|
parser.add_argument("query", nargs="?", help="Search query")
|
|
parser.add_argument("--config", type=Path, default=CONFIG_PATH,
|
|
help="Path to config file")
|
|
parser.add_argument("--json", action="store_true", help="Output as JSON")
|
|
parser.add_argument("--wait", action="store_true",
|
|
help="Wait for download to complete (for overnight jobs)")
|
|
parser.add_argument("--timeout", type=int, default=30,
|
|
help="Timeout in minutes for download completion (default: 30)")
|
|
|
|
args = parser.parse_args()
|
|
config = load_config(args.config)
|
|
recommender = MusicRecommender(config)
|
|
|
|
if args.command == "test":
|
|
print("Testing Last.fm...", end=" ")
|
|
try:
|
|
recent = recommender.lastfm.get_recent_tracks(limit=1)
|
|
print(f"OK (last track: {recent[0]['track'][:30] if recent else 'none'}...)")
|
|
except Exception as e:
|
|
print(f"FAILED: {e}")
|
|
|
|
if recommender.navidrome:
|
|
print("Testing Navidrome...", end=" ")
|
|
if recommender.navidrome.ping():
|
|
artists = recommender.navidrome.get_artists()
|
|
print(f"OK ({len(artists)} artists)")
|
|
else:
|
|
print("FAILED")
|
|
|
|
if recommender.slskd:
|
|
print("Testing slskd...", end=" ")
|
|
try:
|
|
recommender.slskd._request("GET", "server")
|
|
print("OK")
|
|
except Exception as e:
|
|
print(f"FAILED: {e}")
|
|
|
|
elif args.command == "recommend":
|
|
recs = recommender.get_recommendations()
|
|
if args.json:
|
|
print(json.dumps(recs, indent=2))
|
|
else:
|
|
print(recommender.format_recommendations(recs))
|
|
|
|
elif args.command == "gaps":
|
|
gaps = recommender.find_gaps()
|
|
if args.json:
|
|
print(json.dumps(gaps, indent=2))
|
|
else:
|
|
print(recommender.format_gaps(gaps))
|
|
|
|
elif args.command == "recent":
|
|
recent = recommender.lastfm.get_recent_tracks(limit=20)
|
|
if args.json:
|
|
print(json.dumps(recent, indent=2))
|
|
else:
|
|
print("# Recent Tracks\n")
|
|
for t in recent:
|
|
status = "🎵 NOW" if t["now_playing"] else " "
|
|
print(f"{status} {t['artist']} - {t['track']}")
|
|
|
|
elif args.command == "search":
|
|
if not args.query:
|
|
print("Search requires a query")
|
|
sys.exit(1)
|
|
|
|
if recommender.navidrome:
|
|
print(f"Searching Navidrome for '{args.query}'...")
|
|
results = recommender.navidrome.search(args.query)
|
|
if results["artists"]:
|
|
print("\nArtists in library:")
|
|
for a in results["artists"][:5]:
|
|
print(f" - {a['name']}")
|
|
if results["albums"]:
|
|
print("\nAlbums in library:")
|
|
for a in results["albums"][:5]:
|
|
print(f" - {a['artist']} - {a['name']}")
|
|
|
|
elif args.command == "surprise":
|
|
result = recommender.surprise_download(
|
|
wait_for_completion=args.wait,
|
|
timeout_minutes=args.timeout,
|
|
)
|
|
if args.json:
|
|
print(json.dumps(result, indent=2))
|
|
else:
|
|
if result["success"]:
|
|
status = result.get("status", "queued")
|
|
print(f"\n✅ {'Download complete!' if status == 'completed' else 'Queued for download!'}")
|
|
print(f" {result['artist']} - {result['album']}")
|
|
print(f" {result['files']} files {'(FLAC)' if result['has_flac'] else '(MP3)'}")
|
|
print(f" From: {result['source']}")
|
|
if result.get("attempts", 1) > 1:
|
|
print(f" (took {result['attempts']} attempts)")
|
|
else:
|
|
print(f"\n❌ {result['error']}")
|
|
if "tried" in result:
|
|
print(" Tried:")
|
|
for t in result["tried"]:
|
|
print(f" - {t}")
|
|
|
|
elif args.command == "slskd":
|
|
if not args.query:
|
|
print("slskd search requires a query")
|
|
sys.exit(1)
|
|
|
|
if not recommender.slskd:
|
|
print("slskd not configured")
|
|
sys.exit(1)
|
|
|
|
print(f"Searching Soulseek for '{args.query}'...")
|
|
search_id = recommender.slskd.search(args.query)
|
|
recommender.slskd.wait_for_search(search_id, timeout=30)
|
|
# Brief pause to ensure /responses endpoint is populated
|
|
time.sleep(2)
|
|
responses = recommender.slskd.get_search_responses(search_id)
|
|
|
|
print(f"\nFound {len(responses)} users with results:\n")
|
|
for resp in responses[:10]:
|
|
username = resp.get("username", "?")
|
|
files = resp.get("files", [])
|
|
has_slot = "✓" if resp.get("hasFreeUploadSlot") else "✗"
|
|
print(f" {has_slot} {username}: {len(files)} files")
|
|
|
|
elif args.command == "download":
|
|
if not args.query:
|
|
print("download requires 'artist - album'")
|
|
sys.exit(1)
|
|
|
|
if not recommender.slskd:
|
|
print("slskd not configured")
|
|
sys.exit(1)
|
|
|
|
# Parse "artist - album"
|
|
if " - " in args.query:
|
|
artist, album = args.query.split(" - ", 1)
|
|
else:
|
|
print("Format: 'Artist - Album'")
|
|
sys.exit(1)
|
|
|
|
print(f"Searching for: {artist} - {album}")
|
|
match = recommender.slskd.search_and_find_album(artist, album)
|
|
|
|
if match:
|
|
print(f"\nBest match: {match['file_count']} files from {match['username']}")
|
|
print(f"FLAC: {'Yes' if match['has_flac'] else 'No'}")
|
|
print(f"Directory: {match['directory'][:60]}...")
|
|
|
|
if recommender.slskd.download_album(match):
|
|
print("\n✅ Download queued!")
|
|
else:
|
|
print("\n❌ Download failed")
|
|
else:
|
|
print("\n❌ No suitable results found")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|