diff --git a/config/settings.py b/config/settings.py
index 5c1e83db..8ccc0554 100644
--- a/config/settings.py
+++ b/config/settings.py
@@ -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"
diff --git a/core/genius_client.py b/core/genius_client.py
new file mode 100644
index 00000000..0fc1093b
--- /dev/null
+++ b/core/genius_client.py
@@ -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
+ lyrics_parts = []
+ pattern = r'
]*data-lyrics-container="true"[^>]*>(.*?)
'
+ 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'
', '\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
diff --git a/core/lastfm_client.py b/core/lastfm_client.py
new file mode 100644
index 00000000..5132bed3
--- /dev/null
+++ b/core/lastfm_client.py
@@ -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
diff --git a/web_server.py b/web_server.py
index 0376bcbd..14f9d563 100644
--- a/web_server.py
+++ b/web_server.py
@@ -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)
diff --git a/webui/index.html b/webui/index.html
index 62edb177..f65bacdd 100644
--- a/webui/index.html
+++ b/webui/index.html
@@ -3246,6 +3246,38 @@
+
+