NovaSonic/backend/spotify_resolve.py

#!/usr/bin/env python3
"""
Resolves Spotify URLs via the Spotify partner GraphQL API (api-partner.spotify.com).
This is the same internal API the Spotify web player uses — separate rate-limit bucket
from the public api.spotify.com/v1/ which gets throttled on server IPs.
"""
import sys, json, re, time

def eprint(*a): print(*a, file=sys.stderr)

# ─── RESPONSE PARSING ────────────────────────────────────────────────────────
# Partner API track items look like:
# { "itemV2": { "data": { "trackUnion": { "name":…, "artists":…, "albumOfTrack":…,
#                                         "duration":…, "trackNumber":…, "contentRating":… } } } }

def _parse_track(item, default_album='', default_art='', default_artist=''):
    try:
        inner = (item.get('itemV2') or item.get('track') or {}).get('data') or {}
        t = inner.get('trackUnion') or inner
        if not t.get('name'): return None

        artist_items = (t.get('artists') or {}).get('items') or []
        artists = ', '.join(
            a.get('profile', {}).get('name', '') for a in artist_items
        ) or default_artist

        alb       = t.get('albumOfTrack') or {}
        alb_name  = alb.get('name', '') or default_album
        alb_imgs  = (alb.get('images') or {}).get('items') or []
        art       = alb_imgs[0].get('url', '') if alb_imgs else default_art
        release   = ((alb.get('date') or {}).get('isoString') or '')[:10]

        dur_ms    = (t.get('duration') or {}).get('totalMilliseconds')
        rating    = (t.get('contentRating') or {}).get('label', '')

        return {
            'trackName':         t.get('name', ''),
            'artistName':        artists,
            'collectionName':    alb_name,
            'artworkUrl100':     art,
            'releaseDate':       release,
            'trackNumber':       t.get('trackNumber'),
            'trackExplicitness': 'explicit' if rating == 'EXPLICIT' else 'notExplicit',
            'durationMs':        dur_ms,
        }
    except Exception as e:
        eprint(f'[parse] error on item: {e}')
        return None

def _parse_images(imgs_obj):
    items = (imgs_obj or {}).get('items') or []
    return items[0].get('url', '') if items else ''

# ─── RESOLVER ────────────────────────────────────────────────────────────────

def resolve(sp_type, sp_id):
    from spotapi import PublicPlaylist, PublicAlbum, PublicSong

    url = f'https://open.spotify.com/{sp_type}/{sp_id}'

    if sp_type == 'track':
        song = PublicSong(url)
        raw = song.get_song_info()
        eprint(f'[spotapi] track response keys: {list(raw.keys()) if isinstance(raw, dict) else type(raw)}')
        tr = _parse_track(raw)
        if not tr:
            # Fallback: raw might already be a flat dict
            tr = {
                'trackName':         raw.get('name', ''),
                'artistName':        raw.get('artist', ''),
                'collectionName':    raw.get('album', ''),
                'artworkUrl100':     raw.get('image', ''),
                'releaseDate':       raw.get('release_date', ''),
                'trackNumber':       raw.get('track_number'),
                'trackExplicitness': 'notExplicit',
                'durationMs':        raw.get('duration_ms'),
            }
        name   = tr.get('trackName', '')
        artist = tr.get('artistName', '')
        art    = tr.get('artworkUrl100', '')
        return {'type': 'track', 'name': name, 'artist': artist, 'artworkUrl': art, 'tracks': [tr]}

    if sp_type == 'album':
        alb      = PublicAlbum(url)
        alb_info = alb.get_album_info()
        eprint(f'[spotapi] album response type: {type(alb_info)}, keys: {list(alb_info.keys()) if isinstance(alb_info, dict) else "?"}')

        # alb_info may be a dict with metadata + items, or just items
        meta   = alb_info if isinstance(alb_info, dict) else {}
        name   = meta.get('name', '')
        artist = meta.get('artist', '') or meta.get('artists', '')
        art    = meta.get('image', '') or meta.get('cover', '')

        raw_tracks = meta.get('tracks') or meta.get('items') or (alb_info if isinstance(alb_info, list) else [])
        tracks = [t for t in (_parse_track(i) for i in raw_tracks) if t]
        eprint(f'[spotapi] album ok — {len(tracks)} tracks')
        return {'type': 'album', 'name': name, 'artist': artist, 'artworkUrl': art, 'tracks': tracks}

    # playlist
    pl = PublicPlaylist(url)
    pl_meta = pl.get_playlist_info(limit=1)  # first call often returns metadata too
    eprint(f'[spotapi] playlist meta type: {type(pl_meta)}')

    tracks, skipped = [], 0
    offset = 0
    while True:
        page = pl.get_playlist_info(limit=100, offset=offset)
        items = page if isinstance(page, list) else (page.get('items') or page.get('tracks') or [])
        if not items: break
        for item in items:
            if (item.get('itemV2') or item.get('track', {}).get('data', {})).get('__typename') == 'LocalTrackResponseWrapper':
                skipped += 1; continue
            tr = _parse_track(item)
            if tr: tracks.append(tr)
        if len(items) < 100: break
        offset += len(items)
        time.sleep(0.3)

    # Extract playlist name/art from first page if available
    name  = ''
    owner = ''
    art   = ''
    if isinstance(pl_meta, dict):
        name  = pl_meta.get('name', '')
        owner = pl_meta.get('owner', '') or pl_meta.get('owner_name', '')
        art   = pl_meta.get('image', '') or pl_meta.get('cover', '')

    eprint(f'[spotapi] playlist ok — {len(tracks)} tracks')
    return {'type': 'playlist', 'name': name, 'artist': owner,
            'artworkUrl': art, 'skippedLocal': skipped, 'tracks': tracks}

# ─── MAIN ────────────────────────────────────────────────────────────────────

def main():
    if len(sys.argv) < 2: eprint('No URL'); sys.exit(1)
    url = sys.argv[1]
    m = re.search(r'open\.spotify\.com/(?:.*/)?(track|album|playlist)/([A-Za-z0-9]+)', url)
    if not m: eprint('Invalid URL'); sys.exit(1)
    sp_type, sp_id = m.group(1), m.group(2)
    eprint(f'[spotify] resolving {sp_type} {sp_id}')
    try:
        result = resolve(sp_type, sp_id)
        print(json.dumps(result, ensure_ascii=False))
    except SystemExit: raise
    except Exception as e:
        eprint(f'[spotapi] failed: {type(e).__name__}: {e}')
        sys.exit(1)

if __name__ == '__main__':
    try: main()
    except SystemExit: raise
    except Exception as e: eprint(f'fatal: {e}'); sys.exit(1)