soulsync/core/artist_source_lookup.py
Broque Thomas 51e00d4ebf Fix Amazon Music search quality: images, dedup, explicit stripping, album/artist clicks
- All search_raw calls switched from single-type to types="track,album" — T2Tunes only
  returns results when both types are requested together
- _fetch_album_metas: parallel fetch (up to 5 workers) of album cover art via
  album_metadata(asin) — T2Tunes search results carry no image URLs
- search_tracks: populates image_url, release_date, total_tracks from album meta
- search_artists: strips feat. credits via _primary_artist() so "Artist feat. X" and
  "Artist ft. Y" collapse to one "Artist" entry; uses album cover as artist image
  stand-in (same approach as iTunes — T2Tunes has no artist images)
- search_albums: name-based dedup (display_name + artist key) instead of ASIN-based;
  populates image_url, release_date, total_tracks from album meta (cap 10 ASIN fetches)
- _strip_edition(): strips [Explicit]/(Explicit) from track/album names — explicit is
  the default version; Clean/Edited/Censored labels kept as-is so they stay distinct
- get_album(): applies _strip_edition to name and _primary_artist to artist so
  MusicBrainz preflight matching doesn't fail on "[Explicit]" album names
- get_album_tracks(): populates track_number and disc_number from T2TunesStreamInfo
  instead of hardcoding None — fixes track ordering in multi-track album downloads
- get_artist() / get_artist_albums(): _unslugify() converts slug artist IDs back to
  search names; _primary_artist() in comparison handles feat-annotated results
- SOURCE_ONLY_ARTIST_SOURCES: added "amazon" so artist detail page doesn't 404
- build_source_only_artist_detail: added amazon_client param + dispatch branch
- web_server.py: resolve amazon_client in _build_source_only_artist_detail wrapper;
  add source_override=="amazon" branch in get_spotify_album_tracks endpoint
- 77 tests covering all above paths; all pass
2026-05-16 15:55:15 -07:00

88 lines
2.9 KiB
Python

"""Source-artist → library lookup helpers.
Extracted from `web_server.py` so the logic can be imported and unit-tested
without booting the Flask app, Spotify client, Soulseek connection, etc.
Two concepts live here:
* ``SOURCE_ID_FIELD`` — the per-source column on the ``artists`` table that
stores the external service ID (Spotify track ID, Deezer artist ID, …).
This map is what ties a result clicked in the source-aware Search results
back to a library record so we can serve the richer library view.
* ``find_library_artist_for_source`` — given a source-aware click (e.g.
``deezer:525046``), try to locate a matching library artist. First by
direct column match against the source's ID column, then by case-
insensitive name match scoped to the active media server.
"""
from __future__ import annotations
import logging
from typing import Optional
logger = logging.getLogger("artist_source_lookup")
SOURCE_ONLY_ARTIST_SOURCES = frozenset({
"spotify", "itunes", "deezer", "discogs", "hydrabase", "musicbrainz", "amazon",
})
SOURCE_ID_FIELD = {
"spotify": "spotify_artist_id",
"itunes": "itunes_artist_id",
"deezer": "deezer_id",
"discogs": "discogs_id",
"hydrabase": "soul_id",
"musicbrainz": "musicbrainz_id",
}
def find_library_artist_for_source(
database,
source: str,
source_artist_id: str,
artist_name: Optional[str] = None,
active_server: Optional[str] = None,
) -> Optional[str]:
"""Return the library PK of an artist matching the source-aware click.
Lookup order:
1. Direct match on the source-specific ID column (server-agnostic — any
library record with the right external ID is a hit).
2. Case-insensitive name match within ``active_server`` (defaults to the
active media server when not provided), so we don't jump the user
across server contexts on a name collision.
Returns ``None`` on miss or on any database error.
"""
column = SOURCE_ID_FIELD.get(source)
if not column:
return None
try:
with database._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
f"SELECT id, name FROM artists WHERE {column} = ? LIMIT 1",
(str(source_artist_id),),
)
row = cursor.fetchone()
if row:
return row[0]
if artist_name and active_server:
cursor.execute(
"SELECT id FROM artists "
"WHERE LOWER(name) = LOWER(?) AND server_source = ? LIMIT 1",
(artist_name, active_server),
)
row = cursor.fetchone()
if row:
return row[0]
except Exception as e:
logger.debug(
f"Library upgrade lookup failed for {source}:{source_artist_id}: {e}"
)
return None