PersonalizedPlaylistsService: bake in ID-validity gate, lift selectors
User-facing bug found in the discover-page audit: multiple sections (hidden gems, discovery shuffle, popular picks, decade browser, genre browser) had no `WHERE (spotify_track_id IS NOT NULL OR itunes_track_id IS NOT NULL ...)` gate. Tracks with no source IDs in the discovery pool got displayed, the user clicked download, the download silently failed because there was nothing to look up. Lift + gate `PersonalizedPlaylistsService` had 5 selection methods that all shared the same shape — connect to DB, run a SELECT against `discovery_pool` with different WHERE clauses, optionally apply diversity, return list of track dicts. ~366 lines of business logic, ~55% of which was repeated boilerplate. Three new private helpers consolidate everything: - `_select_discovery_tracks(*, source, extra_where, extra_params, order_by, fetch_limit, extra_columns)` — shared SELECT against `discovery_pool`. The mandatory ID gate is hard-coded into the WHERE clause: no opt-out flag, every method inherits it for free. Plus the source filter and the blacklist filter — same shape every selector needs. - `_apply_diversity_filter(tracks, *, max_per_album, max_per_artist, limit)` — per-album / per-artist cap loop, returns trimmed list. Lifted from the inline duplicates in decade / genre / popular_picks. - `_compute_adaptive_diversity_limits(tracks, *, relaxed=False)` — step-function tiers based on unique-artist count. `relaxed=True` gives the slightly looser limits the genre playlist used vs the decade playlist. Re-enable 4 library methods `get_recently_added`, `get_top_tracks`, `get_forgotten_favorites`, `get_familiar_favorites` were all stubs (`return []`) because they predated the schema columns they need. Schema now has them: `tracks.created_at`, `tracks.play_count`, `tracks.last_played`, and the source ID columns added in earlier work. New `_select_library_tracks(*, where_clause, params, order_by, limit)` helper mirrors the discovery selector but targets the `tracks` table joined against `albums` + `artists`. Mandatory ID gate lives in the helper too: every library method automatically rejects rows where spotify_track_id, itunes_track_id, deezer_id, musicbrainz_recording_id, AND audiodb_id are all NULL. Selection rules: - `get_recently_added` — ORDER BY created_at DESC - `get_top_tracks` — WHERE play_count > 0 ORDER BY play_count DESC - `get_forgotten_favorites` — WHERE play_count > 5 AND last_played < (now - 90 days) ORDER BY play_count DESC - `get_familiar_favorites` — WHERE play_count BETWEEN 3 AND 15 Tests `tests/test_personalized_playlists_id_gate.py` — 17 tests pinning: - `_select_discovery_tracks` filters NULL-id rows, honors source + blacklist + extra_where - `_apply_diversity_filter` caps per-album + per-artist + stops at limit - `_compute_adaptive_diversity_limits` returns the right tier for unique-artist count + relaxed flag - All 5 discovery methods (decade, popular_picks, hidden_gems, discovery_shuffle, genre is exercised via the helper) reject NULL-id rows - All 4 library methods reject NULL-id rows + honor their play-count rules Behavior preserved Same diversity tiers, same over-fetch multipliers (10x for decade / genre, 3x for popular_picks), same `popularity DESC, RANDOM()` ordering, same `popularity >= 60` / `< 40` thresholds, same blacklist filter. Public method signatures unchanged — `web_server.py` needs zero edits. Net file: 1089 → ~1170 LOC (helpers + docstrings), but actual business logic across the 9 methods went from ~418 lines down to ~195 (-53%). 2222/2222 full suite green (was 2205 + 17 new). Ruff clean.
This commit is contained in:
Broque Thomas
parent
1f2b8f8ccd
commit
0701bcc213
2 changed files with 884 additions and 379 deletions
|
|
@ -105,8 +105,191 @@ class PersonalizedPlaylistsService:
|
|||
from core.metadata_service import get_primary_source
|
||||
return get_primary_source()
|
||||
|
||||
# Standard column set returned by every discovery_pool selector.
|
||||
# Callers can request additional columns via the `extra_columns` parameter
|
||||
# of `_select_discovery_tracks` (e.g. `release_date`, `artist_genres`).
|
||||
_STANDARD_DISCOVERY_COLUMNS: Tuple[str, ...] = (
|
||||
'spotify_track_id',
|
||||
'itunes_track_id',
|
||||
'track_name',
|
||||
'artist_name',
|
||||
'album_name',
|
||||
'album_cover_url',
|
||||
'duration_ms',
|
||||
'popularity',
|
||||
'track_data_json',
|
||||
'source',
|
||||
)
|
||||
|
||||
def _select_discovery_tracks(
|
||||
self,
|
||||
*,
|
||||
source: str,
|
||||
extra_where: str = "",
|
||||
extra_params: tuple = (),
|
||||
order_by: str = "RANDOM()",
|
||||
fetch_limit: int,
|
||||
extra_columns: tuple = (),
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Shared selector for discovery_pool playlist methods.
|
||||
|
||||
Builds and runs a SELECT against `discovery_pool` with a baked-in
|
||||
ID-validity gate so callers cannot accidentally return rows with no
|
||||
usable source IDs (which would fail downstream when the user clicks
|
||||
download).
|
||||
|
||||
The WHERE clause always includes:
|
||||
source = ?
|
||||
AND (spotify_track_id IS NOT NULL OR itunes_track_id IS NOT NULL)
|
||||
AND LOWER(artist_name) NOT IN
|
||||
(SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
|
||||
|
||||
The ID gate is mandatory and not opt-out by design — if a future
|
||||
method needs to skip it, that's a design discussion, not a flag.
|
||||
|
||||
Callers compose additional filters via `extra_where` (a SQL fragment
|
||||
beginning with "AND ...") and `extra_params` (positional bindings for
|
||||
any `?` placeholders inside `extra_where`).
|
||||
|
||||
Diversity filtering is the caller's responsibility — apply
|
||||
`_apply_diversity_filter` to the returned list if needed.
|
||||
|
||||
Args:
|
||||
source: discovery_pool.source value to filter on.
|
||||
extra_where: optional SQL fragment appended to the WHERE clause.
|
||||
Must start with "AND " if non-empty.
|
||||
extra_params: positional bindings for `?` placeholders in
|
||||
`extra_where`.
|
||||
order_by: ORDER BY expression, used as-is (e.g. "RANDOM()",
|
||||
"popularity DESC, RANDOM()").
|
||||
fetch_limit: LIMIT applied to the query. Callers that intend to
|
||||
run a diversity filter should over-fetch (e.g. `limit * 3`).
|
||||
extra_columns: additional columns to SELECT beyond
|
||||
`_STANDARD_DISCOVERY_COLUMNS` (e.g. `('release_date',)`).
|
||||
|
||||
Returns:
|
||||
List of track dicts via `_build_track_dict`. Returns `[]` on any
|
||||
error (logged at error level).
|
||||
"""
|
||||
try:
|
||||
columns = self._STANDARD_DISCOVERY_COLUMNS + tuple(extra_columns)
|
||||
select_cols = ",\n ".join(columns)
|
||||
|
||||
query = f"""
|
||||
SELECT
|
||||
{select_cols}
|
||||
FROM discovery_pool
|
||||
WHERE source = ?
|
||||
AND (spotify_track_id IS NOT NULL OR itunes_track_id IS NOT NULL)
|
||||
AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
|
||||
{extra_where}
|
||||
ORDER BY {order_by}
|
||||
LIMIT ?
|
||||
"""
|
||||
|
||||
params = (source,) + tuple(extra_params) + (fetch_limit,)
|
||||
|
||||
with self.database._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(query, params)
|
||||
rows = cursor.fetchall()
|
||||
|
||||
return [self._build_track_dict(row, source) for row in rows]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in _select_discovery_tracks (source={source}): {e}")
|
||||
return []
|
||||
|
||||
def _apply_diversity_filter(
|
||||
self,
|
||||
tracks: List[Dict],
|
||||
*,
|
||||
max_per_album: int,
|
||||
max_per_artist: int,
|
||||
limit: int,
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Apply per-album / per-artist diversity caps to a track list.
|
||||
|
||||
Iterates `tracks` in order, accepting each only if its album count
|
||||
is still under `max_per_album` AND its artist count is still under
|
||||
`max_per_artist`. Stops once `limit` tracks have been accepted.
|
||||
|
||||
Returns a new list, already trimmed to at most `limit` items.
|
||||
"""
|
||||
tracks_by_album: Dict[str, int] = {}
|
||||
tracks_by_artist: Dict[str, int] = {}
|
||||
diverse_tracks: List[Dict] = []
|
||||
|
||||
for track in tracks:
|
||||
album = track['album_name']
|
||||
artist = track['artist_name']
|
||||
|
||||
album_count = tracks_by_album.get(album, 0)
|
||||
artist_count = tracks_by_artist.get(artist, 0)
|
||||
|
||||
if album_count < max_per_album and artist_count < max_per_artist:
|
||||
diverse_tracks.append(track)
|
||||
tracks_by_album[album] = album_count + 1
|
||||
tracks_by_artist[artist] = artist_count + 1
|
||||
|
||||
if len(diverse_tracks) >= limit:
|
||||
break
|
||||
|
||||
return diverse_tracks
|
||||
|
||||
def _compute_adaptive_diversity_limits(
|
||||
self,
|
||||
tracks: List[Dict],
|
||||
*,
|
||||
relaxed: bool = False,
|
||||
) -> Tuple[int, int]:
|
||||
"""
|
||||
Pick (max_per_album, max_per_artist) caps based on artist variety.
|
||||
|
||||
Mirrors the step-functions previously inlined in the decade and
|
||||
genre playlist methods. With more unique artists we can be strict;
|
||||
with fewer we relax to still hit the requested track count.
|
||||
|
||||
When `relaxed=True` (used for genre playlists), the moderate and
|
||||
low bands use looser caps and an extra "very limited" tier kicks
|
||||
in below 5 unique artists.
|
||||
|
||||
Args:
|
||||
tracks: candidate track list to inspect.
|
||||
relaxed: True to apply the looser genre-playlist limits.
|
||||
|
||||
Returns:
|
||||
Tuple of (max_per_album, max_per_artist).
|
||||
"""
|
||||
unique_artists = len(set(t['artist_name'] for t in tracks))
|
||||
|
||||
if relaxed:
|
||||
# Genre playlist tiers
|
||||
if unique_artists >= 20:
|
||||
return 3, 5
|
||||
if unique_artists >= 10:
|
||||
return 4, 10
|
||||
if unique_artists >= 5:
|
||||
return 6, 15
|
||||
return 8, 25
|
||||
|
||||
# Decade-style strict tiers
|
||||
if unique_artists >= 20:
|
||||
return 3, 5
|
||||
if unique_artists >= 10:
|
||||
return 4, 8
|
||||
return 5, 12
|
||||
|
||||
def _build_track_dict(self, row, source: str) -> Dict:
|
||||
"""Build a standardized track dictionary from a database row."""
|
||||
"""Build a standardized track dictionary from a database row.
|
||||
|
||||
If the row carries the optional `artist_genres` column (selected via
|
||||
`_select_discovery_tracks(extra_columns=('artist_genres',))`), the raw
|
||||
JSON string is passed through under `_artist_genres_raw` so callers
|
||||
can run Python-side genre matching without re-querying the row.
|
||||
"""
|
||||
# Convert sqlite3.Row to dict if needed (Row objects don't support .get())
|
||||
if hasattr(row, 'keys'):
|
||||
row = dict(row)
|
||||
|
|
@ -118,7 +301,7 @@ class PersonalizedPlaylistsService:
|
|||
except:
|
||||
track_data = None
|
||||
|
||||
return {
|
||||
result = {
|
||||
'track_id': row.get('spotify_track_id') or row.get('itunes_track_id') or row.get('deezer_track_id'),
|
||||
'spotify_track_id': row.get('spotify_track_id'),
|
||||
'itunes_track_id': row.get('itunes_track_id'),
|
||||
|
|
@ -130,9 +313,19 @@ class PersonalizedPlaylistsService:
|
|||
'duration_ms': row.get('duration_ms', 0),
|
||||
'popularity': row.get('popularity', 0),
|
||||
'track_data_json': track_data,
|
||||
'source': source
|
||||
'source': source,
|
||||
}
|
||||
|
||||
# Pass through optional extra columns under underscore-prefixed keys
|
||||
# so callers that requested them via `extra_columns` can use them
|
||||
# without having to re-query.
|
||||
if 'artist_genres' in row:
|
||||
result['_artist_genres_raw'] = row.get('artist_genres')
|
||||
if 'release_date' in row:
|
||||
result['_release_date'] = row.get('release_date')
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def get_parent_genre(spotify_genre: str) -> str:
|
||||
"""
|
||||
|
|
@ -152,52 +345,133 @@ class PersonalizedPlaylistsService:
|
|||
# LIBRARY-BASED PLAYLISTS
|
||||
# ========================================
|
||||
|
||||
def get_recently_added(self, limit: int = 50) -> List[Dict]:
|
||||
def _select_library_tracks(
|
||||
self,
|
||||
*,
|
||||
where_clause: str = "",
|
||||
params: tuple = (),
|
||||
order_by: str = "t.created_at DESC",
|
||||
limit: int,
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Get recently added tracks from library.
|
||||
Shared selector for library-based playlist methods.
|
||||
|
||||
Returns tracks ordered by date_added DESC
|
||||
Builds and runs a SELECT against `tracks` joined to `albums` and
|
||||
`artists`, with a baked-in ID-validity gate so callers cannot
|
||||
accidentally return rows with no usable source IDs (which would fail
|
||||
downstream when the user clicks download or tries to match the track
|
||||
against an external metadata source).
|
||||
|
||||
NOTE: This requires library tracks to have Spotify metadata which may not be available.
|
||||
Returns empty list if schema incompatible.
|
||||
The WHERE clause always includes:
|
||||
(t.spotify_track_id IS NOT NULL
|
||||
OR t.itunes_track_id IS NOT NULL
|
||||
OR t.deezer_id IS NOT NULL
|
||||
OR t.musicbrainz_recording_id IS NOT NULL
|
||||
OR t.audiodb_id IS NOT NULL)
|
||||
|
||||
The ID gate is mandatory and not opt-out by design — mirrors the
|
||||
shared `_select_discovery_tracks` helper.
|
||||
|
||||
Args:
|
||||
where_clause: optional SQL fragment appended to the WHERE clause.
|
||||
Must start with "AND " if non-empty.
|
||||
params: positional bindings for `?` placeholders in `where_clause`.
|
||||
order_by: ORDER BY expression, used as-is.
|
||||
limit: LIMIT applied to the query.
|
||||
|
||||
Returns:
|
||||
List of track dicts in the standard `_build_track_dict` shape with
|
||||
`source='library'`. Returns `[]` on any error (logged at error
|
||||
level).
|
||||
"""
|
||||
try:
|
||||
logger.warning("Recently Added requires Spotify-linked library tracks - returning empty")
|
||||
return []
|
||||
query = f"""
|
||||
SELECT
|
||||
t.spotify_track_id AS spotify_track_id,
|
||||
t.itunes_track_id AS itunes_track_id,
|
||||
t.deezer_id AS deezer_track_id,
|
||||
t.title AS track_name,
|
||||
ar.name AS artist_name,
|
||||
al.title AS album_name,
|
||||
al.thumb_url AS album_cover_url,
|
||||
t.duration AS duration_ms
|
||||
FROM tracks t
|
||||
LEFT JOIN albums al ON t.album_id = al.id
|
||||
LEFT JOIN artists ar ON t.artist_id = ar.id
|
||||
WHERE (t.spotify_track_id IS NOT NULL
|
||||
OR t.itunes_track_id IS NOT NULL
|
||||
OR t.deezer_id IS NOT NULL
|
||||
OR t.musicbrainz_recording_id IS NOT NULL
|
||||
OR t.audiodb_id IS NOT NULL)
|
||||
{where_clause}
|
||||
ORDER BY {order_by}
|
||||
LIMIT ?
|
||||
"""
|
||||
|
||||
bound_params = tuple(params) + (limit,)
|
||||
|
||||
with self.database._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(query, bound_params)
|
||||
rows = cursor.fetchall()
|
||||
|
||||
results: List[Dict] = []
|
||||
for row in rows:
|
||||
row_dict = dict(row) if hasattr(row, 'keys') else row
|
||||
# Library tracks have no popularity score and no JSON blob;
|
||||
# fill the standard shape with the documented defaults.
|
||||
row_dict.setdefault('popularity', 0)
|
||||
row_dict.setdefault('track_data_json', '{}')
|
||||
# Guard against NULL artist/album joins so downstream string
|
||||
# ops (diversity filter, dedupe) don't blow up.
|
||||
if row_dict.get('artist_name') is None:
|
||||
row_dict['artist_name'] = 'Unknown'
|
||||
if row_dict.get('album_name') is None:
|
||||
row_dict['album_name'] = 'Unknown'
|
||||
if row_dict.get('track_name') is None:
|
||||
row_dict['track_name'] = 'Unknown'
|
||||
if row_dict.get('duration_ms') is None:
|
||||
row_dict['duration_ms'] = 0
|
||||
results.append(self._build_track_dict(row_dict, 'library'))
|
||||
|
||||
return results
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting recently added tracks: {e}")
|
||||
logger.error(f"Error in _select_library_tracks: {e}")
|
||||
return []
|
||||
|
||||
def get_recently_added(self, limit: int = 50) -> List[Dict]:
|
||||
"""Get recently added tracks from the local library, newest first."""
|
||||
tracks = self._select_library_tracks(
|
||||
order_by="t.created_at DESC",
|
||||
limit=limit,
|
||||
)
|
||||
logger.info(f"Recently Added: selected {len(tracks)} library tracks")
|
||||
return tracks
|
||||
|
||||
def get_top_tracks(self, limit: int = 50) -> List[Dict]:
|
||||
"""
|
||||
Get user's all-time top tracks based on play count.
|
||||
|
||||
NOTE: This requires library tracks to have Spotify metadata which may not be available.
|
||||
Returns empty list if schema incompatible.
|
||||
"""
|
||||
try:
|
||||
logger.warning("Top Tracks requires Spotify-linked library tracks - returning empty")
|
||||
return []
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting top tracks: {e}")
|
||||
return []
|
||||
"""Get the user's all-time top library tracks by play count."""
|
||||
tracks = self._select_library_tracks(
|
||||
where_clause="AND t.play_count > 0",
|
||||
order_by="t.play_count DESC",
|
||||
limit=limit,
|
||||
)
|
||||
logger.info(f"Top Tracks: selected {len(tracks)} library tracks")
|
||||
return tracks
|
||||
|
||||
def get_forgotten_favorites(self, limit: int = 50) -> List[Dict]:
|
||||
"""
|
||||
Get tracks you loved but haven't played recently.
|
||||
|
||||
NOTE: This requires library tracks to have Spotify metadata which may not be available.
|
||||
Returns empty list if schema incompatible.
|
||||
"""
|
||||
try:
|
||||
logger.warning("Forgotten Favorites requires Spotify-linked library tracks - returning empty")
|
||||
return []
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting forgotten favorites: {e}")
|
||||
return []
|
||||
"""Get tracks the user loved (>5 plays) but hasn't played in 90+ days."""
|
||||
tracks = self._select_library_tracks(
|
||||
where_clause=(
|
||||
"AND t.play_count > 5 "
|
||||
"AND t.last_played IS NOT NULL "
|
||||
"AND t.last_played < datetime('now', '-90 days')"
|
||||
),
|
||||
order_by="t.play_count DESC",
|
||||
limit=limit,
|
||||
)
|
||||
logger.info(f"Forgotten Favorites: selected {len(tracks)} library tracks")
|
||||
return tracks
|
||||
|
||||
def get_decade_playlist(self, decade: int, limit: int = 100, source: str = None) -> List[Dict]:
|
||||
"""
|
||||
|
|
@ -208,99 +482,46 @@ class PersonalizedPlaylistsService:
|
|||
limit: Maximum tracks to return
|
||||
source: Optional source filter ('spotify' or 'itunes'), auto-detects if not provided
|
||||
"""
|
||||
try:
|
||||
start_year = decade
|
||||
end_year = decade + 9
|
||||
start_year = decade
|
||||
end_year = decade + 9
|
||||
active_source = source or self._get_active_source()
|
||||
|
||||
# Determine active source if not specified
|
||||
active_source = source or self._get_active_source()
|
||||
# Over-fetch 10x for diversity filtering headroom.
|
||||
all_tracks = self._select_discovery_tracks(
|
||||
source=active_source,
|
||||
extra_where=(
|
||||
"AND release_date IS NOT NULL "
|
||||
"AND CAST(SUBSTR(release_date, 1, 4) AS INTEGER) BETWEEN ? AND ?"
|
||||
),
|
||||
extra_params=(start_year, end_year),
|
||||
order_by="RANDOM()",
|
||||
fetch_limit=limit * 10,
|
||||
extra_columns=('release_date',),
|
||||
)
|
||||
|
||||
with self.database._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Query discovery_pool - get 10x more for diversity filtering, filtered by source
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
spotify_track_id,
|
||||
itunes_track_id,
|
||||
track_name,
|
||||
artist_name,
|
||||
album_name,
|
||||
album_cover_url,
|
||||
duration_ms,
|
||||
popularity,
|
||||
release_date,
|
||||
track_data_json,
|
||||
source
|
||||
FROM discovery_pool
|
||||
WHERE release_date IS NOT NULL
|
||||
AND CAST(SUBSTR(release_date, 1, 4) AS INTEGER) BETWEEN ? AND ?
|
||||
AND source = ?
|
||||
AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
|
||||
ORDER BY RANDOM()
|
||||
LIMIT ?
|
||||
""", (start_year, end_year, active_source, limit * 10))
|
||||
|
||||
rows = cursor.fetchall()
|
||||
all_tracks = []
|
||||
for row in rows:
|
||||
all_tracks.append(self._build_track_dict(row, active_source))
|
||||
|
||||
if not all_tracks:
|
||||
logger.warning(f"No tracks found for {decade}s")
|
||||
return []
|
||||
|
||||
# Shuffle first for randomness
|
||||
import random
|
||||
random.shuffle(all_tracks)
|
||||
|
||||
# Count unique artists to determine diversity level
|
||||
unique_artists = len(set(track['artist_name'] for track in all_tracks))
|
||||
|
||||
# Adaptive diversity limits based on artist variety
|
||||
if unique_artists >= 20:
|
||||
# Good variety - apply diversity constraints
|
||||
max_per_album = 3
|
||||
max_per_artist = 5
|
||||
elif unique_artists >= 10:
|
||||
# Moderate variety - more lenient
|
||||
max_per_album = 4
|
||||
max_per_artist = 8
|
||||
else:
|
||||
# Low variety - very lenient to hit 50 tracks
|
||||
max_per_album = 5
|
||||
max_per_artist = 12
|
||||
|
||||
logger.info(f"{decade}s has {unique_artists} unique artists - using limits: {max_per_album} per album, {max_per_artist} per artist")
|
||||
|
||||
# Apply diversity constraints
|
||||
tracks_by_album = {}
|
||||
tracks_by_artist = {}
|
||||
diverse_tracks = []
|
||||
|
||||
for track in all_tracks:
|
||||
album = track['album_name']
|
||||
artist = track['artist_name']
|
||||
|
||||
# Count current tracks for this album/artist
|
||||
album_count = tracks_by_album.get(album, 0)
|
||||
artist_count = tracks_by_artist.get(artist, 0)
|
||||
|
||||
if album_count < max_per_album and artist_count < max_per_artist:
|
||||
diverse_tracks.append(track)
|
||||
tracks_by_album[album] = album_count + 1
|
||||
tracks_by_artist[artist] = artist_count + 1
|
||||
|
||||
if len(diverse_tracks) >= limit:
|
||||
break
|
||||
|
||||
logger.info(f"Found {len(diverse_tracks)} tracks from {decade}s in discovery pool (adaptive diversity)")
|
||||
return diverse_tracks[:limit]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting decade playlist for {decade}s: {e}")
|
||||
if not all_tracks:
|
||||
logger.warning(f"No tracks found for {decade}s")
|
||||
return []
|
||||
|
||||
random.shuffle(all_tracks)
|
||||
|
||||
max_per_album, max_per_artist = self._compute_adaptive_diversity_limits(all_tracks)
|
||||
unique_artists = len(set(t['artist_name'] for t in all_tracks))
|
||||
logger.info(
|
||||
f"{decade}s has {unique_artists} unique artists - using limits: "
|
||||
f"{max_per_album} per album, {max_per_artist} per artist"
|
||||
)
|
||||
|
||||
diverse_tracks = self._apply_diversity_filter(
|
||||
all_tracks,
|
||||
max_per_album=max_per_album,
|
||||
max_per_artist=max_per_artist,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
logger.info(f"Found {len(diverse_tracks)} tracks from {decade}s in discovery pool (adaptive diversity)")
|
||||
return diverse_tracks
|
||||
|
||||
def get_available_genres(self, source: str = None) -> List[Dict]:
|
||||
"""
|
||||
Get list of consolidated parent genres with track counts from discovery pool.
|
||||
|
|
@ -363,238 +584,128 @@ class PersonalizedPlaylistsService:
|
|||
logger.error(f"Error getting available genres: {e}")
|
||||
return []
|
||||
|
||||
def _genre_matches(self, artist_genres_json: Optional[str], search_keywords: List[str]) -> bool:
|
||||
"""Return True if any artist genre in the JSON-encoded column matches any keyword."""
|
||||
if not artist_genres_json:
|
||||
return False
|
||||
try:
|
||||
genres = json.loads(artist_genres_json)
|
||||
except Exception:
|
||||
return False
|
||||
for artist_genre in genres:
|
||||
artist_genre_lower = artist_genre.lower()
|
||||
for keyword in search_keywords:
|
||||
if keyword in artist_genre_lower:
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_genre_playlist(self, genre: str, limit: int = 50, source: str = None) -> List[Dict]:
|
||||
"""
|
||||
Get tracks from a specific genre with diversity filtering.
|
||||
Uses cached artist genres from database (populated during discovery scan).
|
||||
Supports both parent genres (e.g., "Electronic/Dance") and specific genres (e.g., "house").
|
||||
|
||||
The genre keyword match runs Python-side over the JSON-encoded artist_genres
|
||||
column, so this method overfetches via the shared selector then filters.
|
||||
"""
|
||||
try:
|
||||
# Determine active source if not specified
|
||||
active_source = source or self._get_active_source()
|
||||
active_source = source or self._get_active_source()
|
||||
|
||||
with self.database._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
# Build keyword list: parent genre expands to all child keywords;
|
||||
# specific genre uses its own name for partial matching.
|
||||
if genre in self.GENRE_MAPPING:
|
||||
search_keywords = [k.lower() for k in self.GENRE_MAPPING[genre]]
|
||||
logger.info(f"Matching parent genre '{genre}' with {len(search_keywords)} child keywords")
|
||||
else:
|
||||
search_keywords = [genre.lower()]
|
||||
logger.info(f"Matching specific genre '{genre}' with partial matching")
|
||||
|
||||
# Get all tracks with genres from discovery pool, filtered by source
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
spotify_track_id,
|
||||
itunes_track_id,
|
||||
track_name,
|
||||
artist_name,
|
||||
album_name,
|
||||
album_cover_url,
|
||||
duration_ms,
|
||||
popularity,
|
||||
artist_genres,
|
||||
track_data_json,
|
||||
source
|
||||
FROM discovery_pool
|
||||
WHERE artist_genres IS NOT NULL
|
||||
AND source = ?
|
||||
AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
|
||||
""", (active_source,))
|
||||
rows = cursor.fetchall()
|
||||
# Pull every source row with non-null artist_genres through the shared
|
||||
# selector (gets the ID gate + blacklist filter for free). Cap at a high
|
||||
# bound — the Python keyword filter narrows the result drastically.
|
||||
candidate_tracks = self._select_discovery_tracks(
|
||||
source=active_source,
|
||||
extra_where="AND artist_genres IS NOT NULL",
|
||||
order_by="RANDOM()",
|
||||
fetch_limit=1_000_000,
|
||||
extra_columns=('artist_genres',),
|
||||
)
|
||||
|
||||
# Determine if this is a parent genre or specific genre
|
||||
is_parent_genre = genre in self.GENRE_MAPPING
|
||||
search_keywords = []
|
||||
|
||||
if is_parent_genre:
|
||||
# Use all child genre keywords for matching
|
||||
search_keywords = self.GENRE_MAPPING[genre]
|
||||
logger.info(f"Matching parent genre '{genre}' with {len(search_keywords)} child keywords")
|
||||
else:
|
||||
# Use the genre name itself for partial matching
|
||||
search_keywords = [genre.lower()]
|
||||
logger.info(f"Matching specific genre '{genre}' with partial matching")
|
||||
|
||||
# Filter tracks that match the genre
|
||||
matching_tracks = []
|
||||
|
||||
for row in rows:
|
||||
try:
|
||||
artist_genres_json = row['artist_genres']
|
||||
if artist_genres_json:
|
||||
genres = json.loads(artist_genres_json)
|
||||
|
||||
# Check if any artist genre matches any search keyword
|
||||
genre_match = False
|
||||
for artist_genre in genres:
|
||||
artist_genre_lower = artist_genre.lower()
|
||||
for keyword in search_keywords:
|
||||
if keyword in artist_genre_lower:
|
||||
genre_match = True
|
||||
break
|
||||
if genre_match:
|
||||
break
|
||||
|
||||
if genre_match:
|
||||
matching_tracks.append(self._build_track_dict(row, active_source))
|
||||
except Exception as e:
|
||||
logger.debug(f"Error parsing genres for track: {e}")
|
||||
continue
|
||||
|
||||
if not matching_tracks:
|
||||
logger.warning(f"No tracks found for genre: {genre}")
|
||||
return []
|
||||
|
||||
# Shuffle before limiting for better variety
|
||||
random.shuffle(matching_tracks)
|
||||
|
||||
# Limit to 10x for diversity filtering
|
||||
all_tracks = matching_tracks[:limit * 10] if len(matching_tracks) > limit * 10 else matching_tracks
|
||||
|
||||
if not all_tracks:
|
||||
return []
|
||||
|
||||
# Apply adaptive diversity filtering (relaxed for genres)
|
||||
unique_artists = len(set(track['artist_name'] for track in all_tracks))
|
||||
|
||||
if unique_artists >= 20:
|
||||
max_per_album = 3
|
||||
max_per_artist = 5
|
||||
elif unique_artists >= 10:
|
||||
max_per_album = 4
|
||||
max_per_artist = 10
|
||||
elif unique_artists >= 5:
|
||||
max_per_album = 6
|
||||
max_per_artist = 15
|
||||
else:
|
||||
# Very limited artist pool - be more lenient
|
||||
max_per_album = 8
|
||||
max_per_artist = 25
|
||||
|
||||
logger.info(f"Genre '{genre}' has {unique_artists} artists, {len(all_tracks)} total tracks - limits: {max_per_album}/album, {max_per_artist}/artist")
|
||||
|
||||
# Shuffle and apply diversity
|
||||
random.shuffle(all_tracks)
|
||||
tracks_by_album = {}
|
||||
tracks_by_artist = {}
|
||||
diverse_tracks = []
|
||||
|
||||
for track in all_tracks:
|
||||
album = track['album_name']
|
||||
artist = track['artist_name']
|
||||
|
||||
album_count = tracks_by_album.get(album, 0)
|
||||
artist_count = tracks_by_artist.get(artist, 0)
|
||||
|
||||
if album_count < max_per_album and artist_count < max_per_artist:
|
||||
diverse_tracks.append(track)
|
||||
tracks_by_album[album] = album_count + 1
|
||||
tracks_by_artist[artist] = artist_count + 1
|
||||
|
||||
if len(diverse_tracks) >= limit:
|
||||
break
|
||||
|
||||
logger.info(f"Found {len(diverse_tracks)} tracks for genre '{genre}'")
|
||||
return diverse_tracks[:limit]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting genre playlist for {genre}: {e}")
|
||||
if not candidate_tracks:
|
||||
logger.warning(f"No tracks with genre data found for source: {active_source}")
|
||||
return []
|
||||
|
||||
# `_build_track_dict` stashes the raw `artist_genres` column under
|
||||
# `_artist_genres_raw` (since we requested it via `extra_columns`),
|
||||
# so the keyword match can run without re-querying.
|
||||
matching_tracks = [
|
||||
track for track in candidate_tracks
|
||||
if self._genre_matches(track.get('_artist_genres_raw'), search_keywords)
|
||||
]
|
||||
|
||||
if not matching_tracks:
|
||||
logger.warning(f"No tracks found for genre: {genre}")
|
||||
return []
|
||||
|
||||
random.shuffle(matching_tracks)
|
||||
|
||||
# Cap candidate set at 10x limit for diversity filtering headroom.
|
||||
all_tracks = matching_tracks[:limit * 10] if len(matching_tracks) > limit * 10 else matching_tracks
|
||||
|
||||
max_per_album, max_per_artist = self._compute_adaptive_diversity_limits(all_tracks, relaxed=True)
|
||||
unique_artists = len(set(t['artist_name'] for t in all_tracks))
|
||||
logger.info(
|
||||
f"Genre '{genre}' has {unique_artists} artists, {len(all_tracks)} total tracks - "
|
||||
f"limits: {max_per_album}/album, {max_per_artist}/artist"
|
||||
)
|
||||
|
||||
random.shuffle(all_tracks)
|
||||
diverse_tracks = self._apply_diversity_filter(
|
||||
all_tracks,
|
||||
max_per_album=max_per_album,
|
||||
max_per_artist=max_per_artist,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
logger.info(f"Found {len(diverse_tracks)} tracks for genre '{genre}'")
|
||||
return diverse_tracks
|
||||
|
||||
# ========================================
|
||||
# DISCOVERY POOL PLAYLISTS
|
||||
# ========================================
|
||||
|
||||
def get_popular_picks(self, limit: int = 50) -> List[Dict]:
|
||||
"""Get high popularity tracks from discovery pool with diversity (max 2 tracks per album/artist)"""
|
||||
# Determine active source
|
||||
"""Get high popularity tracks from discovery pool with diversity (max 2 per album, 3 per artist)."""
|
||||
active_source = self._get_active_source()
|
||||
|
||||
try:
|
||||
with self.database._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
# Over-fetch 3x so the diversity filter has room to spread albums/artists.
|
||||
all_tracks = self._select_discovery_tracks(
|
||||
source=active_source,
|
||||
extra_where="AND popularity >= 60",
|
||||
order_by="popularity DESC, RANDOM()",
|
||||
fetch_limit=limit * 3,
|
||||
)
|
||||
|
||||
# Get more tracks than needed to allow for filtering, filtered by source
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
spotify_track_id,
|
||||
itunes_track_id,
|
||||
track_name,
|
||||
artist_name,
|
||||
album_name,
|
||||
album_cover_url,
|
||||
duration_ms,
|
||||
popularity,
|
||||
track_data_json,
|
||||
source
|
||||
FROM discovery_pool
|
||||
WHERE popularity >= 60 AND source = ?
|
||||
AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
|
||||
ORDER BY popularity DESC, RANDOM()
|
||||
LIMIT ?
|
||||
""", (active_source, limit * 3))
|
||||
diverse_tracks = self._apply_diversity_filter(
|
||||
all_tracks,
|
||||
max_per_album=2,
|
||||
max_per_artist=3,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
rows = cursor.fetchall()
|
||||
all_tracks = [self._build_track_dict(row, active_source) for row in rows]
|
||||
|
||||
# Apply diversity constraint: max 2 tracks per album, max 3 per artist
|
||||
tracks_by_album = {}
|
||||
tracks_by_artist = {}
|
||||
diverse_tracks = []
|
||||
|
||||
for track in all_tracks:
|
||||
album = track['album_name']
|
||||
artist = track['artist_name']
|
||||
|
||||
# Count current tracks for this album/artist
|
||||
album_count = tracks_by_album.get(album, 0)
|
||||
artist_count = tracks_by_artist.get(artist, 0)
|
||||
|
||||
# Apply limits: max 2 per album, max 3 per artist
|
||||
if album_count < 2 and artist_count < 3:
|
||||
diverse_tracks.append(track)
|
||||
tracks_by_album[album] = album_count + 1
|
||||
tracks_by_artist[artist] = artist_count + 1
|
||||
|
||||
if len(diverse_tracks) >= limit:
|
||||
break
|
||||
|
||||
logger.info(f"Popular Picks ({active_source}): Selected {len(diverse_tracks)} tracks with diversity")
|
||||
return diverse_tracks[:limit]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting popular picks: {e}")
|
||||
return []
|
||||
logger.info(f"Popular Picks ({active_source}): selected {len(diverse_tracks)} tracks with diversity")
|
||||
return diverse_tracks
|
||||
|
||||
def get_hidden_gems(self, limit: int = 50) -> List[Dict]:
|
||||
"""Get low popularity (underground/indie) tracks from discovery pool"""
|
||||
# Determine active source
|
||||
"""Get low-popularity (underground/indie) tracks from discovery pool."""
|
||||
active_source = self._get_active_source()
|
||||
|
||||
try:
|
||||
with self.database._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
spotify_track_id,
|
||||
itunes_track_id,
|
||||
track_name,
|
||||
artist_name,
|
||||
album_name,
|
||||
album_cover_url,
|
||||
duration_ms,
|
||||
popularity,
|
||||
track_data_json,
|
||||
source
|
||||
FROM discovery_pool
|
||||
WHERE popularity < 40 AND source = ?
|
||||
AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
|
||||
ORDER BY RANDOM()
|
||||
LIMIT ?
|
||||
""", (active_source, limit))
|
||||
|
||||
rows = cursor.fetchall()
|
||||
return [self._build_track_dict(row, active_source) for row in rows]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting hidden gems: {e}")
|
||||
return []
|
||||
tracks = self._select_discovery_tracks(
|
||||
source=active_source,
|
||||
extra_where="AND popularity < 40",
|
||||
order_by="RANDOM()",
|
||||
fetch_limit=limit,
|
||||
)
|
||||
logger.info(f"Hidden Gems ({active_source}): selected {len(tracks)} tracks")
|
||||
return tracks
|
||||
|
||||
def get_discovery_shuffle(self, limit: int = 50) -> List[Dict]:
|
||||
"""
|
||||
|
|
@ -602,53 +713,24 @@ class PersonalizedPlaylistsService:
|
|||
|
||||
Different every time you call it!
|
||||
"""
|
||||
# Determine active source
|
||||
active_source = self._get_active_source()
|
||||
|
||||
try:
|
||||
with self.database._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
spotify_track_id,
|
||||
itunes_track_id,
|
||||
track_name,
|
||||
artist_name,
|
||||
album_name,
|
||||
album_cover_url,
|
||||
duration_ms,
|
||||
popularity,
|
||||
track_data_json,
|
||||
source
|
||||
FROM discovery_pool
|
||||
WHERE source = ?
|
||||
AND LOWER(artist_name) NOT IN (SELECT LOWER(artist_name) FROM discovery_artist_blacklist)
|
||||
ORDER BY RANDOM()
|
||||
LIMIT ?
|
||||
""", (active_source, limit))
|
||||
|
||||
rows = cursor.fetchall()
|
||||
return [self._build_track_dict(row, active_source) for row in rows]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting discovery shuffle: {e}")
|
||||
return []
|
||||
tracks = self._select_discovery_tracks(
|
||||
source=active_source,
|
||||
order_by="RANDOM()",
|
||||
fetch_limit=limit,
|
||||
)
|
||||
logger.info(f"Discovery Shuffle ({active_source}): selected {len(tracks)} tracks")
|
||||
return tracks
|
||||
|
||||
def get_familiar_favorites(self, limit: int = 50) -> List[Dict]:
|
||||
"""
|
||||
Get tracks with medium play counts (3-15 plays) - your reliable go-tos.
|
||||
|
||||
NOTE: This requires library tracks to have Spotify metadata which may not be available.
|
||||
Returns empty list if schema incompatible.
|
||||
"""
|
||||
try:
|
||||
logger.warning("Familiar Favorites requires Spotify-linked library tracks - returning empty")
|
||||
return []
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting familiar favorites: {e}")
|
||||
return []
|
||||
"""Get tracks with medium play counts (3-15 plays) - reliable go-tos."""
|
||||
tracks = self._select_library_tracks(
|
||||
where_clause="AND t.play_count BETWEEN 3 AND 15",
|
||||
order_by="t.play_count DESC",
|
||||
limit=limit,
|
||||
)
|
||||
logger.info(f"Familiar Favorites: selected {len(tracks)} library tracks")
|
||||
return tracks
|
||||
|
||||
# ========================================
|
||||
# DAILY MIX (HYBRID PLAYLISTS)
|
||||
|
|
|
|||
423
tests/test_personalized_playlists_id_gate.py
Normal file
423
tests/test_personalized_playlists_id_gate.py
Normal file
|
|
@ -0,0 +1,423 @@
|
|||
"""Tests for the lifted PersonalizedPlaylistsService selectors and the
|
||||
mandatory ID-validity gate every section must enforce.
|
||||
|
||||
Context: discover-page selection methods used to return tracks/albums
|
||||
with all source IDs NULL — the UI displayed them, the user clicked
|
||||
download, the download silently failed because there was nothing to
|
||||
look up. This test file pins the gate at the helper level so a future
|
||||
section can't accidentally bypass it.
|
||||
|
||||
Coverage:
|
||||
- `_select_discovery_tracks` filters out rows where every source ID is NULL
|
||||
- `_select_discovery_tracks` honors source filter + blacklist filter
|
||||
- `_select_library_tracks` enforces the same gate against the `tracks` table
|
||||
- `_apply_diversity_filter` caps per-album + per-artist counts
|
||||
- `_compute_adaptive_diversity_limits` returns the right tier for the
|
||||
unique-artist count + relaxed flag
|
||||
- The 5 discovery_pool methods (decade / genre / popular_picks /
|
||||
hidden_gems / discovery_shuffle) each filter NULL-id rows
|
||||
- The 4 library methods (recently_added / top_tracks /
|
||||
forgotten_favorites / familiar_favorites) each filter NULL-id rows
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
from contextlib import contextmanager
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from core.personalized_playlists import PersonalizedPlaylistsService
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fake DB
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _FakeDatabase:
|
||||
"""Wraps an in-memory sqlite connection so the service's
|
||||
`database._get_connection()` calls work the same as in production.
|
||||
|
||||
The schema mirrors the real `discovery_pool` + `tracks` + `albums`
|
||||
+ `artists` shape just enough for the selection methods to exercise.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._conn = sqlite3.connect(":memory:")
|
||||
self._conn.row_factory = sqlite3.Row
|
||||
self._setup_schema()
|
||||
|
||||
def _setup_schema(self):
|
||||
cursor = self._conn.cursor()
|
||||
cursor.executescript("""
|
||||
CREATE TABLE discovery_pool (
|
||||
id INTEGER PRIMARY KEY,
|
||||
source TEXT NOT NULL,
|
||||
spotify_track_id TEXT,
|
||||
itunes_track_id TEXT,
|
||||
deezer_track_id TEXT,
|
||||
track_name TEXT,
|
||||
artist_name TEXT,
|
||||
album_name TEXT,
|
||||
album_cover_url TEXT,
|
||||
duration_ms INTEGER,
|
||||
popularity INTEGER,
|
||||
release_date TEXT,
|
||||
artist_genres TEXT,
|
||||
track_data_json TEXT
|
||||
);
|
||||
CREATE TABLE discovery_artist_blacklist (
|
||||
artist_name TEXT PRIMARY KEY
|
||||
);
|
||||
CREATE TABLE artists (
|
||||
id INTEGER PRIMARY KEY,
|
||||
name TEXT
|
||||
);
|
||||
CREATE TABLE albums (
|
||||
id INTEGER PRIMARY KEY,
|
||||
title TEXT,
|
||||
artist_id INTEGER,
|
||||
thumb_url TEXT
|
||||
);
|
||||
CREATE TABLE tracks (
|
||||
id INTEGER PRIMARY KEY,
|
||||
album_id INTEGER,
|
||||
artist_id INTEGER,
|
||||
title TEXT,
|
||||
duration INTEGER,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
play_count INTEGER DEFAULT 0,
|
||||
last_played TIMESTAMP,
|
||||
spotify_track_id TEXT,
|
||||
itunes_track_id TEXT,
|
||||
deezer_id TEXT,
|
||||
musicbrainz_recording_id TEXT,
|
||||
audiodb_id TEXT
|
||||
);
|
||||
""")
|
||||
self._conn.commit()
|
||||
|
||||
@contextmanager
|
||||
def _get_connection(self):
|
||||
# Match the production interface (`with database._get_connection() as conn`)
|
||||
try:
|
||||
yield self._conn
|
||||
finally:
|
||||
pass
|
||||
|
||||
def insert_discovery_track(self, **kwargs):
|
||||
cols = ", ".join(kwargs.keys())
|
||||
placeholders = ", ".join(["?"] * len(kwargs))
|
||||
self._conn.execute(
|
||||
f"INSERT INTO discovery_pool ({cols}) VALUES ({placeholders})",
|
||||
tuple(kwargs.values()),
|
||||
)
|
||||
self._conn.commit()
|
||||
|
||||
def insert_library_track(self, *, artist_name='Library Artist',
|
||||
album_name='Library Album', track_name='Library Track',
|
||||
**track_kwargs):
|
||||
# Insert artist + album + track in one shot for convenience.
|
||||
cur = self._conn.cursor()
|
||||
cur.execute("INSERT INTO artists (name) VALUES (?)", (artist_name,))
|
||||
artist_id = cur.lastrowid
|
||||
cur.execute(
|
||||
"INSERT INTO albums (title, artist_id, thumb_url) VALUES (?, ?, ?)",
|
||||
(album_name, artist_id, 'http://x/cover.jpg'),
|
||||
)
|
||||
album_id = cur.lastrowid
|
||||
track_kwargs.setdefault('duration', 200000)
|
||||
cur.execute(
|
||||
f"""INSERT INTO tracks (album_id, artist_id, title, {', '.join(track_kwargs.keys())})
|
||||
VALUES (?, ?, ?, {', '.join(['?'] * len(track_kwargs))})""",
|
||||
(album_id, artist_id, track_name, *track_kwargs.values()),
|
||||
)
|
||||
self._conn.commit()
|
||||
|
||||
def blacklist(self, artist_name):
|
||||
self._conn.execute(
|
||||
"INSERT INTO discovery_artist_blacklist (artist_name) VALUES (?)",
|
||||
(artist_name,),
|
||||
)
|
||||
self._conn.commit()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service():
|
||||
"""Service with a fresh in-memory DB. `_get_active_source` patched
|
||||
to return 'spotify' so every selector targets the same source."""
|
||||
db = _FakeDatabase()
|
||||
svc = PersonalizedPlaylistsService(db)
|
||||
with patch.object(svc, '_get_active_source', return_value='spotify'):
|
||||
yield svc, db
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# `_select_discovery_tracks` — the helper everyone goes through
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_discovery_helper_filters_null_id_rows(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='Has Spotify ID',
|
||||
artist_name='A', album_name='A', popularity=50,
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id=None, itunes_track_id=None,
|
||||
track_name='No IDs', artist_name='B', album_name='B', popularity=50,
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', itunes_track_id='it1', track_name='Has iTunes ID',
|
||||
artist_name='C', album_name='C', popularity=50,
|
||||
)
|
||||
|
||||
tracks = svc._select_discovery_tracks(
|
||||
source='spotify',
|
||||
order_by='track_name',
|
||||
fetch_limit=100,
|
||||
)
|
||||
names = sorted(t['track_name'] for t in tracks)
|
||||
assert names == ['Has Spotify ID', 'Has iTunes ID']
|
||||
|
||||
|
||||
def test_discovery_helper_filters_blacklisted_artists(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='Keep',
|
||||
artist_name='Good Artist', album_name='X',
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp2', track_name='Drop',
|
||||
artist_name='Bad Artist', album_name='X',
|
||||
)
|
||||
db.blacklist('bad artist')
|
||||
|
||||
tracks = svc._select_discovery_tracks(
|
||||
source='spotify',
|
||||
order_by='track_name',
|
||||
fetch_limit=100,
|
||||
)
|
||||
assert [t['track_name'] for t in tracks] == ['Keep']
|
||||
|
||||
|
||||
def test_discovery_helper_honors_source_filter(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='SP',
|
||||
artist_name='A', album_name='X',
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='itunes', itunes_track_id='it1', track_name='IT',
|
||||
artist_name='A', album_name='X',
|
||||
)
|
||||
|
||||
tracks = svc._select_discovery_tracks(
|
||||
source='spotify',
|
||||
order_by='track_name',
|
||||
fetch_limit=100,
|
||||
)
|
||||
assert [t['track_name'] for t in tracks] == ['SP']
|
||||
|
||||
|
||||
def test_discovery_helper_honors_extra_where(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='Pop60',
|
||||
artist_name='A', album_name='X', popularity=60,
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp2', track_name='Pop20',
|
||||
artist_name='A', album_name='X', popularity=20,
|
||||
)
|
||||
|
||||
tracks = svc._select_discovery_tracks(
|
||||
source='spotify',
|
||||
extra_where='AND popularity >= 50',
|
||||
order_by='popularity DESC',
|
||||
fetch_limit=100,
|
||||
)
|
||||
assert [t['track_name'] for t in tracks] == ['Pop60']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Diversity filter
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_diversity_filter_caps_per_album():
|
||||
svc = PersonalizedPlaylistsService(_FakeDatabase())
|
||||
tracks = [
|
||||
{'track_name': f't{i}', 'artist_name': 'A', 'album_name': 'Album1'}
|
||||
for i in range(10)
|
||||
]
|
||||
out = svc._apply_diversity_filter(
|
||||
tracks, max_per_album=3, max_per_artist=10, limit=10,
|
||||
)
|
||||
assert len(out) == 3
|
||||
|
||||
|
||||
def test_diversity_filter_caps_per_artist():
|
||||
svc = PersonalizedPlaylistsService(_FakeDatabase())
|
||||
tracks = [
|
||||
{'track_name': f't{i}', 'artist_name': 'OnlyArtist', 'album_name': f'Album{i}'}
|
||||
for i in range(10)
|
||||
]
|
||||
out = svc._apply_diversity_filter(
|
||||
tracks, max_per_album=10, max_per_artist=2, limit=10,
|
||||
)
|
||||
assert len(out) == 2
|
||||
|
||||
|
||||
def test_diversity_filter_stops_at_limit():
|
||||
svc = PersonalizedPlaylistsService(_FakeDatabase())
|
||||
tracks = [
|
||||
{'track_name': f't{i}', 'artist_name': f'a{i}', 'album_name': f'b{i}'}
|
||||
for i in range(20)
|
||||
]
|
||||
out = svc._apply_diversity_filter(
|
||||
tracks, max_per_album=10, max_per_artist=10, limit=5,
|
||||
)
|
||||
assert len(out) == 5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Adaptive diversity limits
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_adaptive_limits_high_variety():
|
||||
svc = PersonalizedPlaylistsService(_FakeDatabase())
|
||||
tracks = [{'artist_name': f'a{i}'} for i in range(30)]
|
||||
max_album, max_artist = svc._compute_adaptive_diversity_limits(tracks)
|
||||
# High variety tier — strict limits
|
||||
assert (max_album, max_artist) == (3, 5)
|
||||
|
||||
|
||||
def test_adaptive_limits_low_variety():
|
||||
svc = PersonalizedPlaylistsService(_FakeDatabase())
|
||||
tracks = [{'artist_name': f'a{i % 3}'} for i in range(30)] # only 3 unique artists
|
||||
max_album, max_artist = svc._compute_adaptive_diversity_limits(tracks)
|
||||
# Low variety tier — much more lenient (matches existing decade-style limits)
|
||||
assert max_album >= 4
|
||||
assert max_artist >= 8
|
||||
|
||||
|
||||
def test_adaptive_limits_relaxed_flag_loosens_genre_tier():
|
||||
svc = PersonalizedPlaylistsService(_FakeDatabase())
|
||||
# 15 unique artists triggers the moderate tier
|
||||
tracks = [{'artist_name': f'a{i}'} for i in range(15)]
|
||||
strict = svc._compute_adaptive_diversity_limits(tracks, relaxed=False)
|
||||
relaxed = svc._compute_adaptive_diversity_limits(tracks, relaxed=True)
|
||||
assert relaxed[1] >= strict[1]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public methods enforce the gate (smoke test on each)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_hidden_gems_filters_null_id_rows(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='Gem',
|
||||
artist_name='A', album_name='X', popularity=20,
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id=None, itunes_track_id=None,
|
||||
track_name='Nogem', artist_name='A', album_name='X', popularity=20,
|
||||
)
|
||||
out = svc.get_hidden_gems(limit=10)
|
||||
assert [t['track_name'] for t in out] == ['Gem']
|
||||
|
||||
|
||||
def test_get_discovery_shuffle_filters_null_id_rows(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='Yes',
|
||||
artist_name='A', album_name='X',
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id=None, itunes_track_id=None,
|
||||
track_name='No', artist_name='A', album_name='X',
|
||||
)
|
||||
out = svc.get_discovery_shuffle(limit=10)
|
||||
assert [t['track_name'] for t in out] == ['Yes']
|
||||
|
||||
|
||||
def test_get_popular_picks_filters_null_id_rows(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='Yes',
|
||||
artist_name='A', album_name='X', popularity=80,
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id=None, itunes_track_id=None,
|
||||
track_name='No', artist_name='A', album_name='X', popularity=80,
|
||||
)
|
||||
out = svc.get_popular_picks(limit=10)
|
||||
assert [t['track_name'] for t in out] == ['Yes']
|
||||
|
||||
|
||||
def test_get_decade_playlist_filters_null_id_rows(service):
|
||||
svc, db = service
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id='sp1', track_name='Yes',
|
||||
artist_name='A', album_name='X', release_date='2024-06-01',
|
||||
)
|
||||
db.insert_discovery_track(
|
||||
source='spotify', spotify_track_id=None, itunes_track_id=None,
|
||||
track_name='No', artist_name='A', album_name='X', release_date='2024-06-01',
|
||||
)
|
||||
out = svc.get_decade_playlist(2020, limit=10)
|
||||
assert [t['track_name'] for t in out] == ['Yes']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Library methods (re-enabled)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_recently_added_filters_null_id_rows(service):
|
||||
svc, db = service
|
||||
db.insert_library_track(
|
||||
track_name='Has Spotify', spotify_track_id='sp1',
|
||||
)
|
||||
db.insert_library_track(
|
||||
track_name='No IDs', # all source ID columns NULL
|
||||
)
|
||||
db.insert_library_track(
|
||||
track_name='Has MB', musicbrainz_recording_id='mb1',
|
||||
)
|
||||
out = svc.get_recently_added(limit=10)
|
||||
names = sorted(t['track_name'] for t in out)
|
||||
assert names == ['Has MB', 'Has Spotify']
|
||||
|
||||
|
||||
def test_get_top_tracks_requires_play_count(service):
|
||||
svc, db = service
|
||||
db.insert_library_track(
|
||||
track_name='Played', spotify_track_id='sp1', play_count=5,
|
||||
)
|
||||
db.insert_library_track(
|
||||
track_name='Unplayed', spotify_track_id='sp2', play_count=0,
|
||||
)
|
||||
out = svc.get_top_tracks(limit=10)
|
||||
assert [t['track_name'] for t in out] == ['Played']
|
||||
|
||||
|
||||
def test_get_familiar_favorites_uses_play_count_band(service):
|
||||
svc, db = service
|
||||
db.insert_library_track(
|
||||
track_name='InBand', spotify_track_id='sp1', play_count=8,
|
||||
)
|
||||
db.insert_library_track(
|
||||
track_name='TooLow', spotify_track_id='sp2', play_count=2,
|
||||
)
|
||||
db.insert_library_track(
|
||||
track_name='TooHigh', spotify_track_id='sp3', play_count=20,
|
||||
)
|
||||
out = svc.get_familiar_favorites(limit=10)
|
||||
assert [t['track_name'] for t in out] == ['InBand']
|
||||
Loading…
Reference in a new issue