Move MusicMap similar artist matching into metadata service
- Relocate the streamed MusicMap similar-artist flow out of web_server.py and into core.metadata_service. - Match similar artists through the configured source-priority chain instead of assuming Spotify first. - Add iTunes artwork fallback so streamed artist payloads still carry image_url when search results are sparse. - Cover the new service behavior with tests.
This commit is contained in:
Antti Kettunen
parent
9863c947dc
commit
b022a90997
3 changed files with 583 additions and 272 deletions
|
|
@ -10,6 +10,7 @@ auth checks, or source-fallback behavior.
|
|||
import threading
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Dict, Any, Literal
|
||||
import requests
|
||||
from core.spotify_client import SpotifyClient
|
||||
from core.itunes_client import iTunesClient
|
||||
from utils.logging_config import get_logger
|
||||
|
|
@ -1415,6 +1416,337 @@ def check_artist_discography_completion(
|
|||
}
|
||||
|
||||
|
||||
def _fetch_musicmap_similar_artist_names(artist_name: str) -> List[str]:
|
||||
"""Fetch similar artist names from MusicMap."""
|
||||
if not (artist_name or '').strip():
|
||||
raise ValueError('Artist name is required')
|
||||
|
||||
from bs4 import BeautifulSoup
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
url_artist = quote_plus(artist_name.strip())
|
||||
musicmap_url = f'https://www.music-map.com/{url_artist}'
|
||||
|
||||
headers = {
|
||||
'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',
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
|
||||
'Accept-Language': 'en-US,en;q=0.5',
|
||||
}
|
||||
|
||||
logger.debug("Fetching MusicMap: %s", musicmap_url)
|
||||
response = requests.get(musicmap_url, headers=headers, timeout=10)
|
||||
response.raise_for_status()
|
||||
|
||||
soup = BeautifulSoup(response.text, 'html.parser')
|
||||
gnod_map = soup.find(id='gnodMap')
|
||||
if not gnod_map:
|
||||
raise ValueError('Could not find artist map on MusicMap')
|
||||
|
||||
searched_artist_lower = _normalize_artist_name(artist_name)
|
||||
similar_artist_names: List[str] = []
|
||||
seen_names = set()
|
||||
|
||||
for anchor in gnod_map.find_all('a'):
|
||||
artist_text = anchor.get_text(strip=True)
|
||||
normalized_name = _normalize_artist_name(artist_text)
|
||||
if not normalized_name or normalized_name == searched_artist_lower or normalized_name in seen_names:
|
||||
continue
|
||||
seen_names.add(normalized_name)
|
||||
similar_artist_names.append(artist_text)
|
||||
|
||||
logger.debug("Found %s similar artists from MusicMap", len(similar_artist_names))
|
||||
return similar_artist_names
|
||||
|
||||
|
||||
def _extract_artist_image_url(artist_data: Any) -> Optional[str]:
|
||||
if not artist_data:
|
||||
return None
|
||||
|
||||
images = _extract_lookup_value(artist_data, 'images', default=[]) or []
|
||||
if not isinstance(images, list):
|
||||
try:
|
||||
images = list(images)
|
||||
except TypeError:
|
||||
images = []
|
||||
|
||||
if images:
|
||||
first_image = images[0]
|
||||
image_url = _extract_lookup_value(first_image, 'url')
|
||||
if image_url:
|
||||
return image_url
|
||||
|
||||
return _extract_lookup_value(
|
||||
artist_data,
|
||||
'image_url',
|
||||
'thumb_url',
|
||||
'cover_image',
|
||||
'picture_xl',
|
||||
'picture_big',
|
||||
'picture_medium',
|
||||
)
|
||||
|
||||
|
||||
def _build_similar_artist_payload(artist_data: Any, source: str) -> Optional[Dict[str, Any]]:
|
||||
artist_id = _extract_lookup_value(artist_data, 'id', 'artist_id', 'spotify_id', 'itunes_id', 'deezer_id')
|
||||
if not artist_id:
|
||||
return None
|
||||
|
||||
if isinstance(artist_data, dict):
|
||||
name = artist_data.get('name') or artist_data.get('artist_name') or artist_data.get('title')
|
||||
genres = artist_data.get('genres') or []
|
||||
popularity = artist_data.get('popularity') or artist_data.get('rank') or 0
|
||||
else:
|
||||
name = (
|
||||
getattr(artist_data, 'name', None)
|
||||
or getattr(artist_data, 'artist_name', None)
|
||||
or getattr(artist_data, 'title', None)
|
||||
)
|
||||
genres = getattr(artist_data, 'genres', None) or []
|
||||
popularity = getattr(artist_data, 'popularity', None) or getattr(artist_data, 'rank', None) or 0
|
||||
|
||||
if isinstance(genres, str):
|
||||
genres = [genres]
|
||||
elif not isinstance(genres, list):
|
||||
try:
|
||||
genres = list(genres)
|
||||
except TypeError:
|
||||
genres = []
|
||||
|
||||
try:
|
||||
popularity = int(popularity or 0)
|
||||
except Exception:
|
||||
popularity = 0
|
||||
|
||||
return {
|
||||
'id': str(artist_id),
|
||||
'name': str(name or artist_id),
|
||||
'image_url': _extract_artist_image_url(artist_data),
|
||||
'genres': genres,
|
||||
'popularity': popularity,
|
||||
'source': source,
|
||||
}
|
||||
|
||||
|
||||
def _resolve_musicmap_artist_source_ids(artist_name: str, source_chain: List[str]) -> Dict[str, Optional[str]]:
|
||||
searched_source_ids: Dict[str, Optional[str]] = {}
|
||||
|
||||
for source in source_chain:
|
||||
client = get_client_for_source(source)
|
||||
if not client:
|
||||
searched_source_ids[source] = None
|
||||
continue
|
||||
|
||||
search_results = _search_artists_for_source(source, client, artist_name, limit=1)
|
||||
searched_source_ids[source] = _extract_lookup_value(search_results[0], 'id', 'artist_id') if search_results else None
|
||||
|
||||
return searched_source_ids
|
||||
|
||||
|
||||
def _match_musicmap_similar_artist(
|
||||
candidate_name: str,
|
||||
source_chain: List[str],
|
||||
searched_artist_name: str,
|
||||
searched_source_ids: Dict[str, Optional[str]],
|
||||
) -> tuple[Optional[str], Optional[Dict[str, Any]]]:
|
||||
target_name = _normalize_artist_name(candidate_name)
|
||||
searched_name = _normalize_artist_name(searched_artist_name)
|
||||
|
||||
for source in source_chain:
|
||||
client = get_client_for_source(source)
|
||||
if not client:
|
||||
continue
|
||||
|
||||
search_results = _search_artists_for_source(source, client, candidate_name, limit=1)
|
||||
if not search_results:
|
||||
continue
|
||||
|
||||
matched_artist = _pick_best_artist_match(search_results, candidate_name)
|
||||
if not matched_artist:
|
||||
continue
|
||||
|
||||
matched_name = _normalize_artist_name(
|
||||
_extract_lookup_value(matched_artist, 'name', 'artist_name', 'title')
|
||||
)
|
||||
if matched_name and matched_name == searched_name:
|
||||
continue
|
||||
|
||||
matched_id = _extract_lookup_value(matched_artist, 'id', 'artist_id')
|
||||
if not matched_id:
|
||||
continue
|
||||
|
||||
if str(matched_id) == str(searched_source_ids.get(source) or ''):
|
||||
continue
|
||||
|
||||
payload = _build_similar_artist_payload(matched_artist, source)
|
||||
if not payload:
|
||||
continue
|
||||
|
||||
if source == 'itunes' and not payload.get('image_url') and hasattr(client, 'get_artist'):
|
||||
try:
|
||||
full_artist = client.get_artist(str(matched_id))
|
||||
image_url = _extract_artist_image_url(full_artist)
|
||||
if image_url:
|
||||
payload['image_url'] = image_url
|
||||
elif hasattr(client, '_get_artist_image_from_albums'):
|
||||
album_image_url = client._get_artist_image_from_albums(str(matched_id))
|
||||
if album_image_url:
|
||||
payload['image_url'] = album_image_url
|
||||
except Exception as exc:
|
||||
logger.debug("Could not enrich iTunes image for %s: %s", matched_id, exc)
|
||||
|
||||
if target_name and _normalize_artist_name(payload['name']) == searched_name:
|
||||
continue
|
||||
|
||||
return source, payload
|
||||
|
||||
return None, None
|
||||
|
||||
|
||||
def iter_musicmap_similar_artist_events(
|
||||
artist_name: str,
|
||||
limit: int = 20,
|
||||
source_override: Optional[str] = None,
|
||||
):
|
||||
"""Yield MusicMap similar-artist events using source priority."""
|
||||
try:
|
||||
source_chain = _get_source_chain_for_lookup(
|
||||
MetadataLookupOptions(source_override=source_override, allow_fallback=True)
|
||||
)
|
||||
available_sources = [source for source in source_chain if get_client_for_source(source)]
|
||||
if not available_sources:
|
||||
yield {
|
||||
'type': 'error',
|
||||
'error': 'No metadata providers available for similar artist matching',
|
||||
'status_code': 503,
|
||||
}
|
||||
return
|
||||
|
||||
similar_artist_names = _fetch_musicmap_similar_artist_names(artist_name)
|
||||
searched_source_ids = _resolve_musicmap_artist_source_ids(artist_name, source_chain)
|
||||
|
||||
yield {
|
||||
'type': 'start',
|
||||
'artist_name': artist_name,
|
||||
'total_found': len(similar_artist_names),
|
||||
'source_priority': source_chain,
|
||||
}
|
||||
|
||||
matched_count = 0
|
||||
seen_names = set()
|
||||
seen_ids = set()
|
||||
|
||||
for candidate_name in similar_artist_names[:limit]:
|
||||
normalized_candidate = _normalize_artist_name(candidate_name)
|
||||
if not normalized_candidate or normalized_candidate in seen_names:
|
||||
continue
|
||||
|
||||
source, payload = _match_musicmap_similar_artist(
|
||||
candidate_name,
|
||||
source_chain,
|
||||
artist_name,
|
||||
searched_source_ids,
|
||||
)
|
||||
if not payload:
|
||||
continue
|
||||
|
||||
payload_id = str(payload.get('id') or '')
|
||||
if payload_id in seen_ids:
|
||||
continue
|
||||
|
||||
seen_names.add(normalized_candidate)
|
||||
seen_ids.add(payload_id)
|
||||
matched_count += 1
|
||||
|
||||
yield {
|
||||
'type': 'artist',
|
||||
'artist': payload,
|
||||
'source': source,
|
||||
}
|
||||
|
||||
yield {
|
||||
'type': 'complete',
|
||||
'complete': True,
|
||||
'total': matched_count,
|
||||
'total_found': len(similar_artist_names),
|
||||
'artist_name': artist_name,
|
||||
'source_priority': source_chain,
|
||||
}
|
||||
|
||||
except requests.exceptions.RequestException as exc:
|
||||
logger.debug("Error fetching MusicMap for %s: %s", artist_name, exc)
|
||||
yield {
|
||||
'type': 'error',
|
||||
'error': f'Failed to fetch from MusicMap: {exc}',
|
||||
'status_code': 502,
|
||||
}
|
||||
except ValueError as exc:
|
||||
status_code = 404 if 'Could not find artist map on MusicMap' in str(exc) else 400
|
||||
yield {
|
||||
'type': 'error',
|
||||
'error': str(exc),
|
||||
'status_code': status_code,
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.error("Error streaming similar artists for %s: %s", artist_name, exc)
|
||||
yield {
|
||||
'type': 'error',
|
||||
'error': str(exc),
|
||||
'status_code': 500,
|
||||
}
|
||||
|
||||
|
||||
def get_musicmap_similar_artists(
|
||||
artist_name: str,
|
||||
limit: int = 20,
|
||||
source_override: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Return matched MusicMap similar artists as a single payload."""
|
||||
artists: List[Dict[str, Any]] = []
|
||||
total_found = 0
|
||||
error_message = None
|
||||
status_code = 500
|
||||
source_priority: List[str] = []
|
||||
|
||||
for event in iter_musicmap_similar_artist_events(
|
||||
artist_name,
|
||||
limit=limit,
|
||||
source_override=source_override,
|
||||
):
|
||||
if event.get('type') == 'start':
|
||||
total_found = event.get('total_found', 0)
|
||||
source_priority = event.get('source_priority', [])
|
||||
elif event.get('type') == 'artist' and event.get('artist'):
|
||||
artists.append(event['artist'])
|
||||
elif event.get('type') == 'complete':
|
||||
total_found = event.get('total_found', total_found)
|
||||
source_priority = event.get('source_priority', source_priority)
|
||||
elif event.get('type') == 'error':
|
||||
error_message = event.get('error', 'Unknown error')
|
||||
status_code = int(event.get('status_code') or status_code or 500)
|
||||
break
|
||||
|
||||
if error_message:
|
||||
return {
|
||||
'success': False,
|
||||
'error': error_message,
|
||||
'status_code': status_code,
|
||||
'artist': artist_name,
|
||||
'similar_artists': [],
|
||||
'total_found': total_found,
|
||||
'total_matched': 0,
|
||||
'source_priority': source_priority,
|
||||
}
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'artist': artist_name,
|
||||
'similar_artists': artists,
|
||||
'total_found': total_found,
|
||||
'total_matched': len(artists),
|
||||
'source_priority': source_priority,
|
||||
}
|
||||
|
||||
|
||||
def get_deezer_client():
|
||||
"""Get cached Deezer client.
|
||||
|
||||
|
|
|
|||
232
tests/test_metadata_service_musicmap.py
Normal file
232
tests/test_metadata_service_musicmap.py
Normal file
|
|
@ -0,0 +1,232 @@
|
|||
import sys
|
||||
import types
|
||||
|
||||
|
||||
if "spotipy" not in sys.modules:
|
||||
spotipy = types.ModuleType("spotipy")
|
||||
|
||||
class _DummySpotify:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
oauth2 = types.ModuleType("spotipy.oauth2")
|
||||
|
||||
class _DummyOAuth:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
spotipy.Spotify = _DummySpotify
|
||||
oauth2.SpotifyOAuth = _DummyOAuth
|
||||
oauth2.SpotifyClientCredentials = _DummyOAuth
|
||||
spotipy.oauth2 = oauth2
|
||||
sys.modules["spotipy"] = spotipy
|
||||
sys.modules["spotipy.oauth2"] = oauth2
|
||||
|
||||
if "config.settings" not in sys.modules:
|
||||
config_pkg = types.ModuleType("config")
|
||||
settings_mod = types.ModuleType("config.settings")
|
||||
|
||||
class _DummyConfigManager:
|
||||
def get(self, key, default=None):
|
||||
return default
|
||||
|
||||
settings_mod.config_manager = _DummyConfigManager()
|
||||
config_pkg.settings = settings_mod
|
||||
sys.modules["config"] = config_pkg
|
||||
sys.modules["config.settings"] = settings_mod
|
||||
|
||||
import types as pytypes
|
||||
|
||||
from core import metadata_service
|
||||
|
||||
|
||||
class _FakeMusicMapResponse:
|
||||
def __init__(self, text):
|
||||
self.text = text
|
||||
|
||||
def raise_for_status(self):
|
||||
return None
|
||||
|
||||
|
||||
class _FakeSourceClient:
|
||||
def __init__(self, results_by_query):
|
||||
self.results_by_query = results_by_query
|
||||
self.search_calls = []
|
||||
|
||||
def search_artists(self, query, **kwargs):
|
||||
self.search_calls.append((query, dict(kwargs)))
|
||||
return list(self.results_by_query.get(query, []))
|
||||
|
||||
|
||||
def test_iter_musicmap_similar_artist_events_uses_source_priority(monkeypatch):
|
||||
html = """
|
||||
<html>
|
||||
<body>
|
||||
<div id="gnodMap">
|
||||
<a href="/artist/seed">Artist One</a>
|
||||
<a href="/artist/similar">Similar Artist</a>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
deezer = _FakeSourceClient({
|
||||
"Artist One": [pytypes.SimpleNamespace(id="dz-seed", name="Artist One")],
|
||||
"Similar Artist": [],
|
||||
})
|
||||
itunes = _FakeSourceClient({
|
||||
"Artist One": [pytypes.SimpleNamespace(id="it-seed", name="Artist One")],
|
||||
"Similar Artist": [
|
||||
pytypes.SimpleNamespace(
|
||||
id="it-match",
|
||||
name="iTunes Canonical",
|
||||
image_url="https://itunes.example/it-match.jpg",
|
||||
genres=["indie", "alt"],
|
||||
popularity=77,
|
||||
)
|
||||
],
|
||||
})
|
||||
spotify = _FakeSourceClient({})
|
||||
|
||||
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
|
||||
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
|
||||
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "itunes", "spotify"])
|
||||
monkeypatch.setattr(
|
||||
metadata_service,
|
||||
"get_client_for_source",
|
||||
lambda source: {"deezer": deezer, "itunes": itunes, "spotify": spotify}.get(source),
|
||||
)
|
||||
|
||||
events = list(metadata_service.iter_musicmap_similar_artist_events("Artist One", limit=5))
|
||||
|
||||
assert events[0]["type"] == "start"
|
||||
assert events[0]["source_priority"] == ["deezer", "itunes", "spotify"]
|
||||
assert events[1]["type"] == "artist"
|
||||
assert events[1]["artist"]["id"] == "it-match"
|
||||
assert events[1]["artist"]["name"] == "iTunes Canonical"
|
||||
assert events[1]["artist"]["image_url"] == "https://itunes.example/it-match.jpg"
|
||||
assert events[1]["artist"]["genres"] == ["indie", "alt"]
|
||||
assert events[1]["artist"]["popularity"] == 77
|
||||
assert events[1]["artist"]["source"] == "itunes"
|
||||
assert events[-1]["type"] == "complete"
|
||||
assert events[-1]["complete"] is True
|
||||
assert events[-1]["total"] == 1
|
||||
assert events[-1]["total_found"] == 1
|
||||
|
||||
assert [call[0] for call in deezer.search_calls] == ["Artist One", "Similar Artist"]
|
||||
assert [call[0] for call in itunes.search_calls] == ["Artist One", "Similar Artist"]
|
||||
assert spotify.search_calls == [("Artist One", {"limit": 1, "allow_fallback": False})]
|
||||
|
||||
|
||||
def test_iter_musicmap_similar_artist_events_enriches_itunes_images(monkeypatch):
|
||||
html = """
|
||||
<html>
|
||||
<body>
|
||||
<div id="gnodMap">
|
||||
<a href="/artist/similar">Similar Artist</a>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
class _ItunesClient(_FakeSourceClient):
|
||||
def __init__(self):
|
||||
super().__init__({
|
||||
"Artist One": [pytypes.SimpleNamespace(id="it-seed", name="Artist One")],
|
||||
"Similar Artist": [pytypes.SimpleNamespace(id="it-match", name="iTunes Canonical")],
|
||||
})
|
||||
self.get_artist_calls = []
|
||||
|
||||
def get_artist(self, artist_id):
|
||||
self.get_artist_calls.append(artist_id)
|
||||
return {
|
||||
"id": artist_id,
|
||||
"name": "iTunes Canonical",
|
||||
"images": [{"url": "https://itunes.example/full-art.jpg"}],
|
||||
"genres": ["indie"],
|
||||
"popularity": 0,
|
||||
}
|
||||
|
||||
itunes = _ItunesClient()
|
||||
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
|
||||
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "itunes")
|
||||
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary])
|
||||
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: itunes if source == "itunes" else None)
|
||||
|
||||
events = list(metadata_service.iter_musicmap_similar_artist_events("Artist One", limit=5))
|
||||
|
||||
artist_events = [event for event in events if event.get("type") == "artist"]
|
||||
assert len(artist_events) == 1
|
||||
assert artist_events[0]["artist"]["image_url"] == "https://itunes.example/full-art.jpg"
|
||||
assert itunes.get_artist_calls == ["it-match"]
|
||||
|
||||
|
||||
def test_iter_musicmap_similar_artist_events_falls_back_to_itunes_album_art(monkeypatch):
|
||||
html = """
|
||||
<html>
|
||||
<body>
|
||||
<div id="gnodMap">
|
||||
<a href="/artist/similar">Similar Artist</a>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
class _ItunesClient(_FakeSourceClient):
|
||||
def __init__(self):
|
||||
super().__init__({
|
||||
"Artist One": [pytypes.SimpleNamespace(id="it-seed", name="Artist One")],
|
||||
"Similar Artist": [pytypes.SimpleNamespace(id="it-match", name="iTunes Canonical")],
|
||||
})
|
||||
self.get_artist_calls = []
|
||||
self.album_art_calls = []
|
||||
|
||||
def get_artist(self, artist_id):
|
||||
self.get_artist_calls.append(artist_id)
|
||||
return {
|
||||
"id": artist_id,
|
||||
"name": "iTunes Canonical",
|
||||
"images": [],
|
||||
"genres": ["indie"],
|
||||
"popularity": 0,
|
||||
}
|
||||
|
||||
def _get_artist_image_from_albums(self, artist_id):
|
||||
self.album_art_calls.append(artist_id)
|
||||
return "https://itunes.example/album-art.jpg"
|
||||
|
||||
itunes = _ItunesClient()
|
||||
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
|
||||
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "itunes")
|
||||
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary])
|
||||
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: itunes if source == "itunes" else None)
|
||||
|
||||
events = list(metadata_service.iter_musicmap_similar_artist_events("Artist One", limit=5))
|
||||
|
||||
artist_events = [event for event in events if event.get("type") == "artist"]
|
||||
assert len(artist_events) == 1
|
||||
assert artist_events[0]["artist"]["image_url"] == "https://itunes.example/album-art.jpg"
|
||||
assert itunes.get_artist_calls == ["it-match"]
|
||||
assert itunes.album_art_calls == ["it-match"]
|
||||
|
||||
|
||||
def test_get_musicmap_similar_artists_returns_not_found_when_musicmap_missing(monkeypatch):
|
||||
html = """
|
||||
<html>
|
||||
<body>
|
||||
<div class="no-map">Nothing here</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
|
||||
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
|
||||
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "itunes"])
|
||||
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: object())
|
||||
|
||||
result = metadata_service.get_musicmap_similar_artists("Artist One", limit=5)
|
||||
|
||||
assert result["success"] is False
|
||||
assert result["status_code"] == 404
|
||||
assert "Could not find artist map" in result["error"]
|
||||
assert result["similar_artists"] == []
|
||||
291
web_server.py
291
web_server.py
|
|
@ -11019,296 +11019,43 @@ def debug_library_photos():
|
|||
|
||||
@app.route('/api/artist/similar/<path:artist_name>/stream')
|
||||
def get_similar_artists_stream(artist_name):
|
||||
"""
|
||||
Stream similar artists from MusicMap and match them to Spotify one by one
|
||||
"""Stream MusicMap similar artists using source-priority metadata matching."""
|
||||
from core.metadata_service import iter_musicmap_similar_artist_events
|
||||
|
||||
Args:
|
||||
artist_name: The artist name to find similar artists for
|
||||
|
||||
Returns:
|
||||
Server-Sent Events stream with each matched artist
|
||||
"""
|
||||
def generate():
|
||||
try:
|
||||
logger.info(f"Streaming similar artists for: {artist_name}")
|
||||
|
||||
# Import required libraries
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
# Construct MusicMap URL
|
||||
url_artist = artist_name.lower().replace(' ', '+')
|
||||
musicmap_url = f'https://www.music-map.com/{url_artist}'
|
||||
|
||||
logger.debug(f"Fetching MusicMap: {musicmap_url}")
|
||||
|
||||
# Set headers to mimic a browser
|
||||
headers = {
|
||||
'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',
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
|
||||
'Accept-Language': 'en-US,en;q=0.5',
|
||||
}
|
||||
|
||||
# Fetch MusicMap page
|
||||
response = requests.get(musicmap_url, headers=headers, timeout=10)
|
||||
response.raise_for_status()
|
||||
|
||||
# Parse HTML
|
||||
soup = BeautifulSoup(response.text, 'html.parser')
|
||||
gnod_map = soup.find(id='gnodMap')
|
||||
|
||||
if not gnod_map:
|
||||
yield f"data: {json.dumps({'error': 'Could not find artist map on MusicMap'})}\n\n"
|
||||
return
|
||||
|
||||
# Extract similar artist names
|
||||
all_anchors = gnod_map.find_all('a')
|
||||
searched_artist_lower = artist_name.lower().strip()
|
||||
|
||||
similar_artist_names = []
|
||||
for anchor in all_anchors:
|
||||
artist_text = anchor.get_text(strip=True)
|
||||
|
||||
# Skip if this is the searched artist
|
||||
if artist_text.lower() == searched_artist_lower:
|
||||
continue
|
||||
|
||||
similar_artist_names.append(artist_text)
|
||||
|
||||
logger.debug(f"Found {len(similar_artist_names)} similar artists from MusicMap")
|
||||
|
||||
# Determine metadata source
|
||||
use_hydrabase = _is_hydrabase_active()
|
||||
|
||||
if not use_hydrabase:
|
||||
if not spotify_client or not spotify_client.is_authenticated():
|
||||
yield f"data: {json.dumps({'error': 'Spotify not authenticated'})}\n\n"
|
||||
return
|
||||
|
||||
# Get the searched artist's ID to exclude them
|
||||
searched_artist_id = None
|
||||
try:
|
||||
if use_hydrabase:
|
||||
searched_results = hydrabase_client.search_artists(artist_name, limit=1)
|
||||
else:
|
||||
searched_results = spotify_client.search_artists(artist_name, limit=1)
|
||||
if searched_results and len(searched_results) > 0:
|
||||
searched_artist_id = searched_results[0].id
|
||||
logger.info(f"Searched artist ID: {searched_artist_id}")
|
||||
except Exception as e:
|
||||
logger.error(f"Could not get searched artist ID: {e}")
|
||||
|
||||
# Match each artist one by one and stream results
|
||||
max_artists = 20
|
||||
matched_count = 0
|
||||
seen_artist_ids = set() # Track seen artist IDs to prevent duplicates
|
||||
|
||||
for artist_name_to_match in similar_artist_names[:max_artists]:
|
||||
try:
|
||||
logger.info(f"Matching: {artist_name_to_match}")
|
||||
|
||||
# Search for the artist via active metadata source
|
||||
if use_hydrabase:
|
||||
results = hydrabase_client.search_artists(artist_name_to_match, limit=1)
|
||||
else:
|
||||
results = spotify_client.search_artists(artist_name_to_match, limit=1)
|
||||
|
||||
if results and len(results) > 0:
|
||||
spotify_artist = results[0]
|
||||
|
||||
# Skip if this is the searched artist
|
||||
if spotify_artist.id == searched_artist_id:
|
||||
logger.info(f"Skipping searched artist: {spotify_artist.name}")
|
||||
continue
|
||||
|
||||
# Skip if we've already seen this artist ID (deduplication)
|
||||
if spotify_artist.id in seen_artist_ids:
|
||||
logger.warning(f"Skipping duplicate artist: {spotify_artist.name}")
|
||||
continue
|
||||
|
||||
seen_artist_ids.add(spotify_artist.id)
|
||||
|
||||
artist_data = {
|
||||
'id': spotify_artist.id,
|
||||
'name': spotify_artist.name,
|
||||
'image_url': spotify_artist.image_url if hasattr(spotify_artist, 'image_url') else None,
|
||||
'genres': spotify_artist.genres if hasattr(spotify_artist, 'genres') else [],
|
||||
'popularity': spotify_artist.popularity if hasattr(spotify_artist, 'popularity') else 0
|
||||
}
|
||||
|
||||
# Stream this matched artist immediately
|
||||
yield f"data: {json.dumps({'artist': artist_data})}\n\n"
|
||||
matched_count += 1
|
||||
|
||||
logger.info(f"Matched and streamed: {spotify_artist.name}")
|
||||
else:
|
||||
logger.warning(f"No Spotify match found for: {artist_name_to_match}")
|
||||
|
||||
except Exception as match_error:
|
||||
logger.error(f"Error matching {artist_name_to_match}: {match_error}")
|
||||
continue
|
||||
|
||||
# Send completion message
|
||||
yield f"data: {json.dumps({'complete': True, 'total': matched_count})}\n\n"
|
||||
logger.info(f"Streaming complete: {matched_count} artists matched")
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.debug(f"Error fetching MusicMap: {e}")
|
||||
yield f"data: {json.dumps({'error': f'Failed to fetch from MusicMap: {str(e)}'})}\n\n"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error streaming similar artists: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
yield f"data: {json.dumps({'error': str(e)})}\n\n"
|
||||
logger.info(f"Streaming similar artists for: {artist_name}")
|
||||
for event in iter_musicmap_similar_artist_events(artist_name, limit=20):
|
||||
yield f"data: {json.dumps(event)}\n\n"
|
||||
if event.get('artist'):
|
||||
time.sleep(0.1)
|
||||
|
||||
return Response(generate(), mimetype='text/event-stream')
|
||||
|
||||
@app.route('/api/artist/similar/<path:artist_name>')
|
||||
def get_similar_artists(artist_name):
|
||||
"""
|
||||
Get similar artists from MusicMap and match them to Spotify (legacy batch endpoint)
|
||||
"""Get MusicMap similar artists using source-priority metadata matching."""
|
||||
from core.metadata_service import get_musicmap_similar_artists
|
||||
|
||||
Args:
|
||||
artist_name: The artist name to find similar artists for
|
||||
|
||||
Returns:
|
||||
JSON with similar artists matched to Spotify data
|
||||
"""
|
||||
try:
|
||||
logger.info(f"Getting similar artists for: {artist_name}")
|
||||
|
||||
# Import required libraries
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
# Construct MusicMap URL
|
||||
url_artist = artist_name.lower().replace(' ', '+')
|
||||
musicmap_url = f'https://www.music-map.com/{url_artist}'
|
||||
|
||||
logger.debug(f"Fetching MusicMap: {musicmap_url}")
|
||||
|
||||
# Set headers to mimic a browser
|
||||
headers = {
|
||||
'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',
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
|
||||
'Accept-Language': 'en-US,en;q=0.5',
|
||||
}
|
||||
|
||||
# Fetch MusicMap page
|
||||
response = requests.get(musicmap_url, headers=headers, timeout=10)
|
||||
response.raise_for_status()
|
||||
|
||||
# Parse HTML
|
||||
soup = BeautifulSoup(response.text, 'html.parser')
|
||||
gnod_map = soup.find(id='gnodMap')
|
||||
|
||||
if not gnod_map:
|
||||
result = get_musicmap_similar_artists(artist_name, limit=20)
|
||||
if not result.get('success'):
|
||||
error = result.get('error', 'Failed to fetch similar artists')
|
||||
status_code = int(result.get('status_code') or 500)
|
||||
return jsonify({
|
||||
"success": False,
|
||||
"error": "Could not find artist map on MusicMap"
|
||||
}), 404
|
||||
|
||||
# Extract similar artist names
|
||||
all_anchors = gnod_map.find_all('a')
|
||||
searched_artist_lower = artist_name.lower().strip()
|
||||
|
||||
similar_artist_names = []
|
||||
for anchor in all_anchors:
|
||||
artist_text = anchor.get_text(strip=True)
|
||||
|
||||
# Skip if this is the searched artist
|
||||
if artist_text.lower() == searched_artist_lower:
|
||||
continue
|
||||
|
||||
similar_artist_names.append(artist_text)
|
||||
|
||||
logger.debug(f"Found {len(similar_artist_names)} similar artists from MusicMap")
|
||||
|
||||
# Determine metadata source
|
||||
use_hydrabase = _is_hydrabase_active()
|
||||
|
||||
if not use_hydrabase:
|
||||
if not spotify_client or not spotify_client.is_authenticated():
|
||||
return jsonify({
|
||||
"success": False,
|
||||
"error": "Spotify not authenticated"
|
||||
}), 401
|
||||
|
||||
# Get the searched artist's ID to exclude them
|
||||
searched_artist_id = None
|
||||
try:
|
||||
if use_hydrabase:
|
||||
searched_results = hydrabase_client.search_artists(artist_name, limit=1)
|
||||
else:
|
||||
searched_results = spotify_client.search_artists(artist_name, limit=1)
|
||||
if searched_results and len(searched_results) > 0:
|
||||
searched_artist_id = searched_results[0].id
|
||||
logger.info(f"Searched artist ID: {searched_artist_id}")
|
||||
except Exception as e:
|
||||
logger.error(f"Could not get searched artist ID: {e}")
|
||||
|
||||
# Match each artist (limit to first 20 for performance)
|
||||
matched_artists = []
|
||||
max_artists = 20
|
||||
seen_artist_ids = set() # Track seen artist IDs to prevent duplicates
|
||||
|
||||
for artist_name_to_match in similar_artist_names[:max_artists]:
|
||||
try:
|
||||
logger.info(f"Matching: {artist_name_to_match}")
|
||||
|
||||
# Search for the artist via active metadata source
|
||||
if use_hydrabase:
|
||||
results = hydrabase_client.search_artists(artist_name_to_match, limit=1)
|
||||
else:
|
||||
results = spotify_client.search_artists(artist_name_to_match, limit=1)
|
||||
|
||||
if results and len(results) > 0:
|
||||
spotify_artist = results[0]
|
||||
|
||||
# Skip if this is the searched artist
|
||||
if spotify_artist.id == searched_artist_id:
|
||||
logger.info(f"Skipping searched artist: {spotify_artist.name}")
|
||||
continue
|
||||
|
||||
# Skip if we've already seen this artist ID (deduplication)
|
||||
if spotify_artist.id in seen_artist_ids:
|
||||
logger.warning(f"Skipping duplicate artist: {spotify_artist.name}")
|
||||
continue
|
||||
|
||||
seen_artist_ids.add(spotify_artist.id)
|
||||
|
||||
matched_artists.append({
|
||||
'id': spotify_artist.id,
|
||||
'name': spotify_artist.name,
|
||||
'image_url': spotify_artist.image_url if hasattr(spotify_artist, 'image_url') else None,
|
||||
'genres': spotify_artist.genres if hasattr(spotify_artist, 'genres') else [],
|
||||
'popularity': spotify_artist.popularity if hasattr(spotify_artist, 'popularity') else 0
|
||||
})
|
||||
|
||||
logger.info(f"Matched: {spotify_artist.name}")
|
||||
else:
|
||||
logger.warning(f"No Spotify match found for: {artist_name_to_match}")
|
||||
|
||||
except Exception as match_error:
|
||||
logger.error(f"Error matching {artist_name_to_match}: {match_error}")
|
||||
continue
|
||||
|
||||
logger.info(f"Successfully matched {len(matched_artists)} artists to Spotify")
|
||||
"error": error
|
||||
}), status_code
|
||||
|
||||
return jsonify({
|
||||
"success": True,
|
||||
"artist": artist_name,
|
||||
"similar_artists": matched_artists,
|
||||
"total_found": len(similar_artist_names),
|
||||
"total_matched": len(matched_artists)
|
||||
"similar_artists": result.get('similar_artists', []),
|
||||
"total_found": result.get('total_found', 0),
|
||||
"total_matched": result.get('total_matched', 0),
|
||||
"source_priority": result.get('source_priority', []),
|
||||
})
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.debug(f"Error fetching MusicMap: {e}")
|
||||
return jsonify({
|
||||
"success": False,
|
||||
"error": f"Failed to fetch from MusicMap: {str(e)}"
|
||||
}), 500
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting similar artists: {e}")
|
||||
import traceback
|
||||
|
|
|
|||
Loading…
Reference in a new issue