Add Last.fm and Genius API clients with settings integration
This commit is contained in:
Broque Thomas
parent
1bd66cf5b4
commit
f8d23ec37c
7 changed files with 798 additions and 1 deletions
|
|
@ -209,6 +209,12 @@ class ConfigManager:
|
|||
"api_key": "",
|
||||
"enabled": False # Disabled by default - requires API key and fpcalc
|
||||
},
|
||||
"lastfm": {
|
||||
"api_key": ""
|
||||
},
|
||||
"genius": {
|
||||
"access_token": ""
|
||||
},
|
||||
"logging": {
|
||||
"path": "logs/app.log",
|
||||
"level": "INFO"
|
||||
|
|
|
|||
372
core/genius_client.py
Normal file
372
core/genius_client.py
Normal file
|
|
@ -0,0 +1,372 @@
|
|||
import requests
|
||||
import re
|
||||
import time
|
||||
import threading
|
||||
from typing import Dict, Optional, Any, List
|
||||
from functools import wraps
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
logger = get_logger("genius_client")
|
||||
|
||||
# Global rate limiting variables
|
||||
_last_api_call_time = 0
|
||||
_api_call_lock = threading.Lock()
|
||||
MIN_API_INTERVAL = 0.5 # 500ms between calls (Genius doesn't publish limits, be conservative)
|
||||
|
||||
|
||||
def rate_limited(func):
|
||||
"""Decorator to enforce rate limiting on Genius API calls"""
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
global _last_api_call_time
|
||||
|
||||
with _api_call_lock:
|
||||
current_time = time.time()
|
||||
time_since_last_call = current_time - _last_api_call_time
|
||||
|
||||
if time_since_last_call < MIN_API_INTERVAL:
|
||||
sleep_time = MIN_API_INTERVAL - time_since_last_call
|
||||
time.sleep(sleep_time)
|
||||
|
||||
_last_api_call_time = time.time()
|
||||
|
||||
try:
|
||||
result = func(*args, **kwargs)
|
||||
return result
|
||||
except Exception as e:
|
||||
if "rate limit" in str(e).lower() or "429" in str(e):
|
||||
logger.warning(f"Genius rate limit hit, implementing backoff: {e}")
|
||||
time.sleep(10.0)
|
||||
raise e
|
||||
return wrapper
|
||||
|
||||
|
||||
class GeniusClient:
|
||||
"""Client for interacting with the Genius API (metadata + lyrics scraping)"""
|
||||
|
||||
BASE_URL = "https://api.genius.com"
|
||||
|
||||
def __init__(self, access_token: str = ""):
|
||||
self.access_token = access_token
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({
|
||||
'User-Agent': 'SoulSync/1.0',
|
||||
'Accept': 'application/json'
|
||||
})
|
||||
# Separate session for web scraping (no auth header)
|
||||
self.scrape_session = requests.Session()
|
||||
self.scrape_session.headers.update({
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||||
})
|
||||
logger.info("Genius client initialized")
|
||||
|
||||
def _make_request(self, endpoint: str, params: Dict = None, timeout: int = 10) -> Optional[Dict]:
|
||||
"""Make an authenticated request to the Genius API"""
|
||||
if not self.access_token:
|
||||
logger.warning("Genius access token not configured")
|
||||
return None
|
||||
|
||||
headers = {
|
||||
'Authorization': f'Bearer {self.access_token}'
|
||||
}
|
||||
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.BASE_URL}{endpoint}",
|
||||
params=params or {},
|
||||
headers=headers,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
if response.status_code == 401:
|
||||
logger.error("Genius API: Invalid access token")
|
||||
return None
|
||||
if response.status_code == 404:
|
||||
return None
|
||||
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
meta = data.get('meta', {})
|
||||
if meta.get('status') != 200:
|
||||
logger.error(f"Genius API error: {meta}")
|
||||
return None
|
||||
|
||||
return data.get('response')
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
logger.warning(f"Genius API timeout for endpoint: {endpoint}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Genius API request error ({endpoint}): {e}")
|
||||
return None
|
||||
|
||||
# ── Search Methods ──
|
||||
|
||||
@rate_limited
|
||||
def search(self, query: str, per_page: int = 5) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Search Genius for songs matching a query.
|
||||
|
||||
Returns:
|
||||
List of hit dicts, each containing a 'result' with:
|
||||
id, title, artist_names, url, song_art_image_url, lyrics_state
|
||||
"""
|
||||
data = self._make_request('/search', {
|
||||
'q': query,
|
||||
'per_page': per_page
|
||||
})
|
||||
if not data:
|
||||
return []
|
||||
|
||||
hits = data.get('hits', [])
|
||||
return [h for h in hits if h.get('type') == 'song']
|
||||
|
||||
@rate_limited
|
||||
def search_song(self, artist_name: str, track_title: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Search for a specific song by artist and title.
|
||||
Returns the best matching song result.
|
||||
|
||||
Returns:
|
||||
Song dict with: id, title, artist_names, url, song_art_image_url,
|
||||
primary_artist (id, name, url, image_url), album (id, name, url)
|
||||
"""
|
||||
query = f"{artist_name} {track_title}"
|
||||
hits = self.search(query, per_page=5)
|
||||
|
||||
if not hits:
|
||||
logger.debug(f"No results for: {query}")
|
||||
return None
|
||||
|
||||
# Try to find best match by checking artist name
|
||||
artist_lower = artist_name.lower().strip()
|
||||
title_lower = track_title.lower().strip()
|
||||
|
||||
for hit in hits:
|
||||
result = hit.get('result', {})
|
||||
result_artist = (result.get('artist_names') or '').lower()
|
||||
result_title = (result.get('title') or '').lower()
|
||||
|
||||
# Check if artist and title match reasonably
|
||||
if artist_lower in result_artist or result_artist in artist_lower:
|
||||
if title_lower in result_title or result_title in title_lower:
|
||||
logger.debug(f"Found song match: {result.get('title')} by {result.get('artist_names')}")
|
||||
return result
|
||||
|
||||
# Fall back to first result
|
||||
first = hits[0].get('result')
|
||||
if first:
|
||||
logger.debug(f"Using first result: {first.get('title')} by {first.get('artist_names')}")
|
||||
return first
|
||||
|
||||
return None
|
||||
|
||||
# ── Song Methods ──
|
||||
|
||||
@rate_limited
|
||||
def get_song(self, song_id: int) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get detailed song info by Genius song ID.
|
||||
|
||||
Returns:
|
||||
Song dict with: id, title, artist_names, url, song_art_image_url,
|
||||
description (dom object), album, media, custom_performances,
|
||||
producer_artists, writer_artists, featured_artists
|
||||
"""
|
||||
data = self._make_request(f'/songs/{song_id}', {
|
||||
'text_format': 'plain'
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
song = data.get('song')
|
||||
if song:
|
||||
logger.debug(f"Got song info for ID: {song_id}")
|
||||
return song
|
||||
|
||||
return None
|
||||
|
||||
# ── Artist Methods ──
|
||||
|
||||
@rate_limited
|
||||
def search_artist(self, artist_name: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Search for an artist by name (via song search, extract primary_artist).
|
||||
|
||||
Returns:
|
||||
Artist dict with: id, name, url, image_url
|
||||
"""
|
||||
hits = self.search(artist_name, per_page=5)
|
||||
if not hits:
|
||||
return None
|
||||
|
||||
artist_lower = artist_name.lower().strip()
|
||||
|
||||
for hit in hits:
|
||||
result = hit.get('result', {})
|
||||
primary = result.get('primary_artist', {})
|
||||
if primary and artist_lower in (primary.get('name') or '').lower():
|
||||
logger.debug(f"Found artist: {primary.get('name')}")
|
||||
return primary
|
||||
|
||||
# Fall back to first result's primary artist
|
||||
first_artist = hits[0].get('result', {}).get('primary_artist')
|
||||
if first_artist:
|
||||
return first_artist
|
||||
|
||||
return None
|
||||
|
||||
@rate_limited
|
||||
def get_artist(self, artist_id: int) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get detailed artist info by Genius artist ID.
|
||||
|
||||
Returns:
|
||||
Artist dict with: id, name, url, image_url, description,
|
||||
alternate_names, facebook_name, twitter_name
|
||||
"""
|
||||
data = self._make_request(f'/artists/{artist_id}', {
|
||||
'text_format': 'plain'
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
artist = data.get('artist')
|
||||
if artist:
|
||||
logger.debug(f"Got artist info for ID: {artist_id}")
|
||||
return artist
|
||||
|
||||
return None
|
||||
|
||||
@rate_limited
|
||||
def get_artist_songs(self, artist_id: int, sort: str = 'popularity', per_page: int = 20) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get songs by an artist.
|
||||
|
||||
Args:
|
||||
artist_id: Genius artist ID
|
||||
sort: Sort order ('popularity', 'title')
|
||||
per_page: Results per page
|
||||
|
||||
Returns:
|
||||
List of song dicts
|
||||
"""
|
||||
data = self._make_request(f'/artists/{artist_id}/songs', {
|
||||
'sort': sort,
|
||||
'per_page': per_page
|
||||
})
|
||||
if not data:
|
||||
return []
|
||||
|
||||
return data.get('songs', [])
|
||||
|
||||
# ── Lyrics Scraping ──
|
||||
|
||||
@rate_limited
|
||||
def get_lyrics(self, song_url: str) -> Optional[str]:
|
||||
"""
|
||||
Scrape lyrics from a Genius song page.
|
||||
The Genius API doesn't provide lyrics directly — they must be scraped from the web page.
|
||||
|
||||
Args:
|
||||
song_url: Full Genius URL (e.g. https://genius.com/Artist-song-lyrics)
|
||||
|
||||
Returns:
|
||||
Lyrics text or None
|
||||
"""
|
||||
if not song_url:
|
||||
return None
|
||||
|
||||
try:
|
||||
response = self.scrape_session.get(song_url, timeout=15)
|
||||
if response.status_code != 200:
|
||||
logger.warning(f"Failed to fetch lyrics page: {response.status_code}")
|
||||
return None
|
||||
|
||||
html = response.text
|
||||
|
||||
# Extract lyrics from the page
|
||||
# Genius wraps lyrics in <div data-lyrics-container="true">
|
||||
lyrics_parts = []
|
||||
pattern = r'<div[^>]*data-lyrics-container="true"[^>]*>(.*?)</div>'
|
||||
matches = re.findall(pattern, html, re.DOTALL)
|
||||
|
||||
if not matches:
|
||||
logger.debug(f"No lyrics containers found on page: {song_url}")
|
||||
return None
|
||||
|
||||
for match in matches:
|
||||
# Clean HTML tags
|
||||
text = re.sub(r'<br\s*/?>', '\n', match)
|
||||
text = re.sub(r'<[^>]+>', '', text)
|
||||
# Decode HTML entities
|
||||
text = text.replace('&', '&')
|
||||
text = text.replace('<', '<')
|
||||
text = text.replace('>', '>')
|
||||
text = text.replace(''', "'")
|
||||
text = text.replace('"', '"')
|
||||
text = text.replace(''', "'")
|
||||
lyrics_parts.append(text.strip())
|
||||
|
||||
lyrics = '\n'.join(lyrics_parts).strip()
|
||||
if lyrics:
|
||||
logger.debug(f"Scraped {len(lyrics)} chars of lyrics from: {song_url}")
|
||||
return lyrics
|
||||
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error scraping lyrics from {song_url}: {e}")
|
||||
return None
|
||||
|
||||
def search_and_get_lyrics(self, artist_name: str, track_title: str) -> Optional[str]:
|
||||
"""
|
||||
Convenience method: search for a song and scrape its lyrics.
|
||||
|
||||
Returns:
|
||||
Lyrics text or None
|
||||
"""
|
||||
song = self.search_song(artist_name, track_title)
|
||||
if not song:
|
||||
return None
|
||||
|
||||
url = song.get('url')
|
||||
if not url:
|
||||
return None
|
||||
|
||||
# Check if lyrics are available
|
||||
lyrics_state = song.get('lyrics_state', '')
|
||||
if lyrics_state == 'unreleased':
|
||||
logger.debug(f"Lyrics unreleased for: {artist_name} - {track_title}")
|
||||
return None
|
||||
|
||||
return self.get_lyrics(url)
|
||||
|
||||
# ── Utility Methods ──
|
||||
|
||||
def extract_description(self, description_data) -> Optional[str]:
|
||||
"""
|
||||
Extract plain text description from Genius description object.
|
||||
When text_format=plain, description comes as {plain: "text"}.
|
||||
"""
|
||||
if not description_data:
|
||||
return None
|
||||
|
||||
if isinstance(description_data, dict):
|
||||
plain = description_data.get('plain', '')
|
||||
if plain and plain.strip() and plain.strip() != '?':
|
||||
return plain.strip()
|
||||
|
||||
if isinstance(description_data, str) and description_data.strip():
|
||||
return description_data.strip()
|
||||
|
||||
return None
|
||||
|
||||
def validate_token(self) -> bool:
|
||||
"""Test if the access token is valid by making a simple request"""
|
||||
if not self.access_token:
|
||||
return False
|
||||
|
||||
data = self._make_request('/search', {'q': 'test', 'per_page': 1})
|
||||
return data is not None
|
||||
333
core/lastfm_client.py
Normal file
333
core/lastfm_client.py
Normal file
|
|
@ -0,0 +1,333 @@
|
|||
import requests
|
||||
import time
|
||||
import threading
|
||||
from typing import Dict, Optional, Any, List
|
||||
from functools import wraps
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
logger = get_logger("lastfm_client")
|
||||
|
||||
# Global rate limiting variables
|
||||
_last_api_call_time = 0
|
||||
_api_call_lock = threading.Lock()
|
||||
MIN_API_INTERVAL = 0.2 # 200ms between calls (Last.fm allows 5 req/sec)
|
||||
|
||||
|
||||
def rate_limited(func):
|
||||
"""Decorator to enforce rate limiting on Last.fm API calls"""
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
global _last_api_call_time
|
||||
|
||||
with _api_call_lock:
|
||||
current_time = time.time()
|
||||
time_since_last_call = current_time - _last_api_call_time
|
||||
|
||||
if time_since_last_call < MIN_API_INTERVAL:
|
||||
sleep_time = MIN_API_INTERVAL - time_since_last_call
|
||||
time.sleep(sleep_time)
|
||||
|
||||
_last_api_call_time = time.time()
|
||||
|
||||
try:
|
||||
result = func(*args, **kwargs)
|
||||
return result
|
||||
except Exception as e:
|
||||
if "rate limit" in str(e).lower() or "429" in str(e):
|
||||
logger.warning(f"Last.fm rate limit hit, implementing backoff: {e}")
|
||||
time.sleep(5.0)
|
||||
raise e
|
||||
return wrapper
|
||||
|
||||
|
||||
class LastFMClient:
|
||||
"""Client for interacting with the Last.fm API (read-only metadata)"""
|
||||
|
||||
BASE_URL = "https://ws.audioscrobbler.com/2.0/"
|
||||
|
||||
def __init__(self, api_key: str = ""):
|
||||
self.api_key = api_key
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({
|
||||
'User-Agent': 'SoulSync/1.0',
|
||||
'Accept': 'application/json'
|
||||
})
|
||||
logger.info("Last.fm client initialized")
|
||||
|
||||
def _make_request(self, method: str, params: Dict = None, timeout: int = 10) -> Optional[Dict]:
|
||||
"""Make a request to the Last.fm API"""
|
||||
if not self.api_key:
|
||||
logger.warning("Last.fm API key not configured")
|
||||
return None
|
||||
|
||||
request_params = {
|
||||
'method': method,
|
||||
'api_key': self.api_key,
|
||||
'format': 'json'
|
||||
}
|
||||
if params:
|
||||
request_params.update(params)
|
||||
|
||||
try:
|
||||
response = self.session.get(
|
||||
self.BASE_URL,
|
||||
params=request_params,
|
||||
timeout=timeout
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
# Last.fm returns errors inside the JSON
|
||||
if 'error' in data:
|
||||
error_code = data.get('error')
|
||||
error_msg = data.get('message', 'Unknown error')
|
||||
# Error 6 = "Artist/Album/Track not found" — not a real error
|
||||
if error_code == 6:
|
||||
return None
|
||||
logger.error(f"Last.fm API error ({error_code}): {error_msg}")
|
||||
return None
|
||||
|
||||
return data
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
logger.warning(f"Last.fm API timeout for method: {method}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Last.fm API request error ({method}): {e}")
|
||||
return None
|
||||
|
||||
# ── Artist Methods ──
|
||||
|
||||
@rate_limited
|
||||
def search_artist(self, artist_name: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Search for an artist by name.
|
||||
|
||||
Returns:
|
||||
Artist dict with: name, mbid, url, listeners, image
|
||||
"""
|
||||
data = self._make_request('artist.search', {
|
||||
'artist': artist_name,
|
||||
'limit': 5
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
results = data.get('results', {}).get('artistmatches', {}).get('artist', [])
|
||||
if results and len(results) > 0:
|
||||
logger.debug(f"Found artist for query: {artist_name}")
|
||||
return results[0]
|
||||
|
||||
logger.debug(f"No artist found for query: {artist_name}")
|
||||
return None
|
||||
|
||||
@rate_limited
|
||||
def get_artist_info(self, artist_name: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get detailed artist info including bio, tags, stats.
|
||||
|
||||
Returns:
|
||||
Artist dict with: name, mbid, url, image, stats (listeners, playcount),
|
||||
similar (artists), tags (tag list), bio (summary, content)
|
||||
"""
|
||||
data = self._make_request('artist.getinfo', {
|
||||
'artist': artist_name,
|
||||
'autocorrect': 1
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
artist = data.get('artist')
|
||||
if artist:
|
||||
logger.debug(f"Got artist info for: {artist_name}")
|
||||
return artist
|
||||
|
||||
return None
|
||||
|
||||
@rate_limited
|
||||
def get_artist_top_tags(self, artist_name: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get top tags for an artist (genres, styles, moods).
|
||||
|
||||
Returns:
|
||||
List of tag dicts with: name, count, url
|
||||
"""
|
||||
data = self._make_request('artist.gettoptags', {
|
||||
'artist': artist_name,
|
||||
'autocorrect': 1
|
||||
})
|
||||
if not data:
|
||||
return []
|
||||
|
||||
tags = data.get('toptags', {}).get('tag', [])
|
||||
return tags if isinstance(tags, list) else [tags] if tags else []
|
||||
|
||||
@rate_limited
|
||||
def get_similar_artists(self, artist_name: str, limit: int = 10) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get similar artists.
|
||||
|
||||
Returns:
|
||||
List of artist dicts with: name, mbid, match (similarity score), url, image
|
||||
"""
|
||||
data = self._make_request('artist.getsimilar', {
|
||||
'artist': artist_name,
|
||||
'autocorrect': 1,
|
||||
'limit': limit
|
||||
})
|
||||
if not data:
|
||||
return []
|
||||
|
||||
artists = data.get('similarartists', {}).get('artist', [])
|
||||
return artists if isinstance(artists, list) else [artists] if artists else []
|
||||
|
||||
# ── Album Methods ──
|
||||
|
||||
@rate_limited
|
||||
def search_album(self, artist_name: str, album_title: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Search for an album.
|
||||
|
||||
Returns:
|
||||
Album dict with: name, artist, url, image
|
||||
"""
|
||||
data = self._make_request('album.search', {
|
||||
'album': f"{artist_name} {album_title}",
|
||||
'limit': 5
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
results = data.get('results', {}).get('albummatches', {}).get('album', [])
|
||||
if results and len(results) > 0:
|
||||
logger.debug(f"Found album for query: {artist_name} - {album_title}")
|
||||
return results[0]
|
||||
|
||||
logger.debug(f"No album found for query: {artist_name} - {album_title}")
|
||||
return None
|
||||
|
||||
@rate_limited
|
||||
def get_album_info(self, artist_name: str, album_title: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get detailed album info including tags, tracks, wiki.
|
||||
|
||||
Returns:
|
||||
Album dict with: name, artist, mbid, url, image, listeners, playcount,
|
||||
tracks (track list), tags (tag list), wiki (summary, content)
|
||||
"""
|
||||
data = self._make_request('album.getinfo', {
|
||||
'artist': artist_name,
|
||||
'album': album_title,
|
||||
'autocorrect': 1
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
album = data.get('album')
|
||||
if album:
|
||||
logger.debug(f"Got album info for: {artist_name} - {album_title}")
|
||||
return album
|
||||
|
||||
return None
|
||||
|
||||
# ── Track Methods ──
|
||||
|
||||
@rate_limited
|
||||
def search_track(self, artist_name: str, track_title: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Search for a track.
|
||||
|
||||
Returns:
|
||||
Track dict with: name, artist, url, listeners
|
||||
"""
|
||||
data = self._make_request('track.search', {
|
||||
'track': track_title,
|
||||
'artist': artist_name,
|
||||
'limit': 5
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
results = data.get('results', {}).get('trackmatches', {}).get('track', [])
|
||||
if results and len(results) > 0:
|
||||
logger.debug(f"Found track for query: {artist_name} - {track_title}")
|
||||
return results[0]
|
||||
|
||||
logger.debug(f"No track found for query: {artist_name} - {track_title}")
|
||||
return None
|
||||
|
||||
@rate_limited
|
||||
def get_track_info(self, artist_name: str, track_title: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get detailed track info including tags, wiki, play stats.
|
||||
|
||||
Returns:
|
||||
Track dict with: name, mbid, url, duration, listeners, playcount,
|
||||
artist, album, toptags (tag list), wiki (summary, content)
|
||||
"""
|
||||
data = self._make_request('track.getinfo', {
|
||||
'artist': artist_name,
|
||||
'track': track_title,
|
||||
'autocorrect': 1
|
||||
})
|
||||
if not data:
|
||||
return None
|
||||
|
||||
track = data.get('track')
|
||||
if track:
|
||||
logger.debug(f"Got track info for: {artist_name} - {track_title}")
|
||||
return track
|
||||
|
||||
return None
|
||||
|
||||
# ── Utility Methods ──
|
||||
|
||||
def get_best_image(self, images: List) -> Optional[str]:
|
||||
"""
|
||||
Extract the best quality image URL from Last.fm image array.
|
||||
Last.fm returns images as [{#text: url, size: small/medium/large/extralarge/mega}]
|
||||
"""
|
||||
if not images or not isinstance(images, list):
|
||||
return None
|
||||
|
||||
# Prefer largest
|
||||
for size in ['mega', 'extralarge', 'large', 'medium', 'small']:
|
||||
for img in images:
|
||||
if isinstance(img, dict) and img.get('size') == size:
|
||||
url = img.get('#text', '')
|
||||
if url:
|
||||
return url
|
||||
|
||||
return None
|
||||
|
||||
def extract_tags(self, tags_data, max_tags: int = 10) -> List[str]:
|
||||
"""
|
||||
Extract tag names from Last.fm tags response.
|
||||
Filters out low-count tags and normalizes.
|
||||
"""
|
||||
if not tags_data:
|
||||
return []
|
||||
|
||||
tag_list = tags_data if isinstance(tags_data, list) else tags_data.get('tag', [])
|
||||
if not isinstance(tag_list, list):
|
||||
tag_list = [tag_list] if tag_list else []
|
||||
|
||||
tags = []
|
||||
for tag in tag_list[:max_tags]:
|
||||
if isinstance(tag, dict):
|
||||
name = tag.get('name', '').strip()
|
||||
if name and len(name) > 1:
|
||||
tags.append(name)
|
||||
elif isinstance(tag, str):
|
||||
tags.append(tag.strip())
|
||||
|
||||
return tags
|
||||
|
||||
def validate_api_key(self) -> bool:
|
||||
"""Test if the API key is valid by making a simple request"""
|
||||
if not self.api_key:
|
||||
return False
|
||||
|
||||
data = self._make_request('chart.gettopartists', {'limit': 1})
|
||||
return data is not None
|
||||
|
|
@ -3188,6 +3188,36 @@ def run_service_test(service, test_config):
|
|||
return False, f"{message}. {fingerprint_status}"
|
||||
except Exception as e:
|
||||
return False, f"AcoustID test error: {str(e)}"
|
||||
elif service == "lastfm":
|
||||
api_key = test_config.get('api_key', '')
|
||||
|
||||
if not api_key:
|
||||
return False, "Missing Last.fm API key."
|
||||
|
||||
try:
|
||||
from core.lastfm_client import LastFMClient
|
||||
client = LastFMClient(api_key=api_key)
|
||||
if client.validate_api_key():
|
||||
return True, "Successfully connected to Last.fm!"
|
||||
else:
|
||||
return False, "Invalid Last.fm API key."
|
||||
except Exception as e:
|
||||
return False, f"Last.fm connection error: {str(e)}"
|
||||
elif service == "genius":
|
||||
access_token = test_config.get('access_token', '')
|
||||
|
||||
if not access_token:
|
||||
return False, "Missing Genius access token."
|
||||
|
||||
try:
|
||||
from core.genius_client import GeniusClient
|
||||
client = GeniusClient(access_token=access_token)
|
||||
if client.validate_token():
|
||||
return True, "Successfully connected to Genius!"
|
||||
else:
|
||||
return False, "Invalid Genius access token."
|
||||
except Exception as e:
|
||||
return False, f"Genius connection error: {str(e)}"
|
||||
return False, "Unknown service."
|
||||
except AttributeError as e:
|
||||
# This specifically catches the error you reported for Jellyfin
|
||||
|
|
@ -3901,7 +3931,7 @@ def handle_settings():
|
|||
if 'active_media_server' in new_settings:
|
||||
config_manager.set_active_media_server(new_settings['active_media_server'])
|
||||
|
||||
for service in ['spotify', 'plex', 'jellyfin', 'navidrome', 'soulseek', 'download_source', 'settings', 'database', 'metadata_enhancement', 'file_organization', 'playlist_sync', 'tidal', 'tidal_download', 'listenbrainz', 'acoustid', 'import', 'lossy_copy', 'ui_appearance', 'youtube', 'content_filter']:
|
||||
for service in ['spotify', 'plex', 'jellyfin', 'navidrome', 'soulseek', 'download_source', 'settings', 'database', 'metadata_enhancement', 'file_organization', 'playlist_sync', 'tidal', 'tidal_download', 'listenbrainz', 'acoustid', 'lastfm', 'genius', 'import', 'lossy_copy', 'ui_appearance', 'youtube', 'content_filter']:
|
||||
if service in new_settings:
|
||||
for key, value in new_settings[service].items():
|
||||
config_manager.set(f'{service}.{key}', value)
|
||||
|
|
|
|||
|
|
@ -3246,6 +3246,38 @@
|
|||
<button class="test-button" onclick="clearQuarantine()" style="margin-top: 10px; background: rgba(255,82,82,0.15); border-color: rgba(255,82,82,0.3); color: #ff5252;">Clear Quarantine</button>
|
||||
</div>
|
||||
|
||||
<!-- Last.fm Settings -->
|
||||
<div class="api-service-frame">
|
||||
<h4 class="service-title lastfm-title">Last.fm</h4>
|
||||
<div class="form-group">
|
||||
<label>API Key:</label>
|
||||
<input type="password" id="lastfm-api-key"
|
||||
placeholder="Last.fm API Key">
|
||||
</div>
|
||||
<div class="callback-info">
|
||||
<div class="callback-help">Get your API key from <a
|
||||
href="https://www.last.fm/api/account/create" target="_blank"
|
||||
style="color: #d51007;">Last.fm API Account</a></div>
|
||||
<div class="callback-help">Used for artist tags, play counts, and similar artists. Callback URL is not used.</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Genius Settings -->
|
||||
<div class="api-service-frame">
|
||||
<h4 class="service-title genius-title">Genius</h4>
|
||||
<div class="form-group">
|
||||
<label>Client Access Token:</label>
|
||||
<input type="password" id="genius-access-token"
|
||||
placeholder="Genius Client Access Token">
|
||||
</div>
|
||||
<div class="callback-info">
|
||||
<div class="callback-help">Get your token from <a
|
||||
href="https://genius.com/api-clients" target="_blank"
|
||||
style="color: #ffff64;">Genius API Clients</a></div>
|
||||
<div class="callback-help">Generate a "Client Access Token" — no OAuth flow needed.</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Test Connection Buttons -->
|
||||
<div class="api-test-buttons">
|
||||
<button class="test-button" onclick="testConnection('spotify')">Test
|
||||
|
|
@ -3257,6 +3289,10 @@
|
|||
ListenBrainz</button>
|
||||
<button class="test-button" onclick="testConnection('acoustid')">Test
|
||||
AcoustID</button>
|
||||
<button class="test-button" onclick="testConnection('lastfm')">Test
|
||||
Last.fm</button>
|
||||
<button class="test-button" onclick="testConnection('genius')">Test
|
||||
Genius</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
|
|
|||
|
|
@ -3905,6 +3905,12 @@ async function loadSettingsData() {
|
|||
document.getElementById('acoustid-api-key').value = settings.acoustid?.api_key || '';
|
||||
document.getElementById('acoustid-enabled').checked = settings.acoustid?.enabled || false;
|
||||
|
||||
// Populate Last.fm settings
|
||||
document.getElementById('lastfm-api-key').value = settings.lastfm?.api_key || '';
|
||||
|
||||
// Populate Genius settings
|
||||
document.getElementById('genius-access-token').value = settings.genius?.access_token || '';
|
||||
|
||||
// Populate Download settings (right column)
|
||||
document.getElementById('download-path').value = settings.soulseek?.download_path || './downloads';
|
||||
document.getElementById('transfer-path').value = settings.soulseek?.transfer_path || './Transfer';
|
||||
|
|
@ -4615,6 +4621,12 @@ async function saveSettings(quiet = false) {
|
|||
api_key: document.getElementById('acoustid-api-key').value,
|
||||
enabled: document.getElementById('acoustid-enabled').checked
|
||||
},
|
||||
lastfm: {
|
||||
api_key: document.getElementById('lastfm-api-key').value
|
||||
},
|
||||
genius: {
|
||||
access_token: document.getElementById('genius-access-token').value
|
||||
},
|
||||
download_source: {
|
||||
mode: document.getElementById('download-source-mode').value,
|
||||
hybrid_primary: document.getElementById('hybrid-primary-source').value,
|
||||
|
|
|
|||
|
|
@ -1558,6 +1558,14 @@ body {
|
|||
color: #ba55d3;
|
||||
}
|
||||
|
||||
.lastfm-title {
|
||||
color: #d51007;
|
||||
}
|
||||
|
||||
.genius-title {
|
||||
color: #ffff64;
|
||||
}
|
||||
|
||||
/* Server Toggle Buttons - upgraded */
|
||||
.server-toggle-container {
|
||||
display: flex;
|
||||
|
|
|
|||
Loading…
Reference in a new issue