170 lines
8.3 KiB
Python
170 lines
8.3 KiB
Python
import random
|
|
import re
|
|
from typing import List, Dict, Set, Any, Optional
|
|
from app.providers.yandex import YandexMusicProvider
|
|
from app.providers.base import TrackMetadata
|
|
|
|
def _normalize(s: Optional[str]) -> str:
|
|
if not s:
|
|
return ""
|
|
s = s.lower()
|
|
s = re.sub(r"[^\w\s]", "", s)
|
|
return " ".join(s.split())
|
|
|
|
class WaveRecommender:
|
|
def __init__(self, yandex_provider: Optional[YandexMusicProvider] = None):
|
|
self.yandex_provider = yandex_provider or YandexMusicProvider()
|
|
|
|
def get_candidate_tracks(
|
|
self,
|
|
anchor_profile: Dict[str, Any],
|
|
existing_keys: Set[str],
|
|
wave_keys: Set[str],
|
|
target_count: int = 20
|
|
) -> List[TrackMetadata]:
|
|
"""Discover candidate tracks based on user anchors not yet in the library."""
|
|
candidates: List[TrackMetadata] = []
|
|
seen_keys: Set[str] = set()
|
|
|
|
try:
|
|
client = self.yandex_provider.get_client()
|
|
except Exception as e:
|
|
print(f"[WaveRecommender] Yandex client unavailable: {e}")
|
|
return candidates
|
|
|
|
# 1. Select anchor artists from profile
|
|
anchor_artists: List[str] = []
|
|
|
|
# Pull from starred
|
|
for s in anchor_profile.get("starred", []):
|
|
if s.get("artist") and s["artist"] not in anchor_artists:
|
|
anchor_artists.append(s["artist"])
|
|
|
|
# Pull from top played
|
|
for p in anchor_profile.get("top_played", []):
|
|
if p.get("artist") and p["artist"] not in anchor_artists:
|
|
anchor_artists.append(p["artist"])
|
|
|
|
# Pull from top library artists
|
|
for a in anchor_profile.get("top_artists", []):
|
|
if a.get("artist") and a["artist"] not in anchor_artists:
|
|
anchor_artists.append(a["artist"])
|
|
|
|
# Fallback to recent if empty
|
|
if not anchor_artists:
|
|
for r in anchor_profile.get("recent_added", []):
|
|
if r.get("artist") and r["artist"] not in anchor_artists:
|
|
anchor_artists.append(r["artist"])
|
|
|
|
if not anchor_artists:
|
|
print("[WaveRecommender] No anchors found in user profile.")
|
|
return candidates
|
|
|
|
# Randomly sample 3-5 anchors for diversity in this run
|
|
sample_size = min(5, len(anchor_artists))
|
|
selected_anchors = random.sample(anchor_artists, sample_size)
|
|
print(f"[WaveRecommender] Sourcing candidates using anchors: {selected_anchors}")
|
|
|
|
for anchor_name in selected_anchors:
|
|
if len(candidates) >= target_count:
|
|
break
|
|
try:
|
|
# Search for anchor artist to get artist ID
|
|
search_res = client.search(anchor_name, type_="artist")
|
|
if not search_res or not search_res.artists or not search_res.artists.results:
|
|
continue
|
|
|
|
artist_obj = search_res.artists.results[0]
|
|
|
|
# Sourcing A: Similar artists
|
|
sim_res = client.artists_similar(artist_obj.id)
|
|
similar_artists = sim_res.similar_artists[:4] if sim_res and sim_res.similar_artists else []
|
|
|
|
for sim_art in similar_artists:
|
|
if len(candidates) >= target_count:
|
|
break
|
|
try:
|
|
tracks_res = client.artists_tracks(sim_art.id, page=0, page_size=6)
|
|
if not tracks_res or not tracks_res.tracks:
|
|
continue
|
|
|
|
for t in tracks_res.tracks:
|
|
if not t or not t.available:
|
|
continue
|
|
art_name = ", ".join(a.name for a in t.artists) if t.artists else sim_art.name
|
|
key = f"{_normalize(art_name)}|{_normalize(t.title)}"
|
|
|
|
# Filter out existing library or wave tracks
|
|
if key in existing_keys or key in wave_keys or key in seen_keys:
|
|
continue
|
|
|
|
meta = self.yandex_provider._convert_track(t)
|
|
if meta:
|
|
seen_keys.add(key)
|
|
candidates.append(meta)
|
|
if len(candidates) >= target_count:
|
|
break
|
|
except Exception as e:
|
|
print(f"[WaveRecommender] Error fetching tracks for {sim_art.name}: {e}")
|
|
|
|
# Sourcing B: Also get popular tracks from the anchor artist's own catalog that user doesn't have yet
|
|
if len(candidates) < target_count:
|
|
own_tracks = client.artists_tracks(artist_obj.id, page=0, page_size=6)
|
|
if own_tracks and own_tracks.tracks:
|
|
for t in own_tracks.tracks:
|
|
if not t or not t.available:
|
|
continue
|
|
art_name = ", ".join(a.name for a in t.artists) if t.artists else artist_obj.name
|
|
key = f"{_normalize(art_name)}|{_normalize(t.title)}"
|
|
if key not in existing_keys and key not in wave_keys and key not in seen_keys:
|
|
meta = self.yandex_provider._convert_track(t)
|
|
if meta:
|
|
seen_keys.add(key)
|
|
candidates.append(meta)
|
|
if len(candidates) >= target_count:
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f"[WaveRecommender] Error processing anchor {anchor_name}: {e}")
|
|
|
|
# Sourcing C: Spotify top tracks for anchor artists (User preference: Spotify)
|
|
try:
|
|
from spotify_scraper import SpotifyClient
|
|
with SpotifyClient() as sp_client:
|
|
for anchor_name in selected_anchors:
|
|
try:
|
|
sp_search = sp_client.search(anchor_name, types=["artist"], limit=1)
|
|
if hasattr(sp_search, "artists") and sp_search.artists:
|
|
sp_art = sp_client.get_artist(sp_search.artists[0].id)
|
|
if hasattr(sp_art, "top_tracks") and sp_art.top_tracks:
|
|
for st in sp_art.top_tracks[:5]:
|
|
st_art = ", ".join(a.name for a in st.artists) if st.artists else anchor_name
|
|
key = f"{_normalize(st_art)}|{_normalize(st.name)}"
|
|
if key in existing_keys or key in wave_keys or key in seen_keys:
|
|
continue
|
|
cover = None
|
|
if hasattr(st, "images") and st.images:
|
|
img = st.images[0]
|
|
cover = getattr(img, "url", None) or (img.get("url") if isinstance(img, dict) else None)
|
|
album_name = getattr(st.album, "name", "Singles") if getattr(st, "album", None) else "Singles"
|
|
meta = TrackMetadata(
|
|
service="spotify",
|
|
service_id=str(st.id),
|
|
title=st.name,
|
|
artist=st_art,
|
|
album=album_name,
|
|
duration_ms=getattr(st, "duration_ms", None),
|
|
cover_url=cover,
|
|
is_explicit=getattr(st, "explicit", False),
|
|
source_url=getattr(st, "url", None) or f"https://open.spotify.com/track/{st.id}"
|
|
)
|
|
seen_keys.add(key)
|
|
# Insert Spotify tracks at the beginning so they are preferred
|
|
candidates.insert(0, meta)
|
|
except Exception as err:
|
|
print(f"[WaveRecommender] Spotify sourcing notice for {anchor_name}: {err}")
|
|
except Exception as e:
|
|
print(f"[WaveRecommender] Spotify client notice: {e}")
|
|
|
|
return candidates[:target_count]
|