musicgrabber/spotify_browser.py

"""
MusicGrabber - Spotify Headless Browser Script

Standalone script executed as a subprocess by spotify.py.
Receives spotify_type and spotify_id via environment variables.
Outputs JSON to stdout: {success, tracks, playlist_name, count} or {success, error}.
"""

import json
import os
import re
import sys
import time

from playwright.sync_api import sync_playwright

spotify_type = os.environ["SPOTIFY_TYPE"]
spotify_id = os.environ["SPOTIFY_ID"]
url = f"https://open.spotify.com/{spotify_type}/{spotify_id}"
SELECTOR = '[data-testid="tracklist-row"]'
is_liked_songs = os.environ.get("SPOTIFY_IS_LIKED_SONGS", "") == "1"

tracks = []
playlist_name = f"Spotify {spotify_type.title()}"
stall_timeout_seconds = 30
stall_raw = (os.environ.get("SPOTIFY_BROWSER_STALL_SECONDS") or "").strip()
if stall_raw.isdigit():
    stall_timeout_seconds = max(5, min(300, int(stall_raw)))

sp_dc = (os.environ.get("SPOTIFY_SP_DC") or "").strip() or None

# Spotify's embed trackList tops out here, so any real total worth chasing is larger.
# We only trust a DOM-read count as a hard completion target when it clears this, so a
# misread small number can never cut a big playlist short.
EMBED_TRACK_LIMIT = 100


def _float_env(name: str, default: float) -> float:
    try:
        return float((os.environ.get(name) or "").strip())
    except (TypeError, ValueError):
        return default


def _int_env(name: str) -> int | None:
    raw = (os.environ.get(name) or "").strip()
    return int(raw) if raw.isdigit() else None


# The parent tells us when it'll pull the plug; we bow out a touch before then with
# whatever we've gathered. seconds_per_track + overhead let us guesstimate how long the
# full scroll should take from the count we read off the page.
hard_deadline_seconds = _int_env("SPOTIFY_BROWSER_DEADLINE_SECONDS")
seconds_per_track = _float_env("SPOTIFY_BROWSER_SECONDS_PER_TRACK", 0.12)
base_overhead = _float_env("SPOTIFY_BROWSER_BASE_OVERHEAD", 120.0)


def _detect_total_from_dom(page) -> int | None:
    """Read the playlist's real track count from the rendered page.

    Spotify's server HTML is now a JS shell, but the live page still shows the count
    in its header as 'N songs' / 'N items'. We take the largest such number we find in
    the header region (the total dwarfs any incidental match), so we know when we've
    got the lot and roughly how long to keep scrolling.
    """
    selectors = [
        '[data-testid="playlist-page"]',
        '[data-testid="entityTitle"]',
        '[data-testid="action-bar-row"]',
        "header",
    ]
    for sel in selectors:
        try:
            el = page.query_selector(sel)
            if not el:
                continue
            text = el.inner_text()
        except Exception:
            continue
        best = None
        for m in re.finditer(r"(\d[\d.,   ]*)\s+(?:songs?|tracks?|items?)\b", text, re.IGNORECASE):
            digits = re.sub(r"\D", "", m.group(1))
            if digits and (best is None or int(digits) > best):
                best = int(digits)
        if best:
            return best
    return None


_script_start = time.monotonic()

try:
    with sync_playwright() as p:
        browser = p.chromium.launch(
            headless=True,
            args=["--disable-blink-features=AutomationControlled"],
        )
        context = browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
            viewport={"width": 1280, "height": 800},
        )
        # Stop Spotify (and other sites) detecting Playwright via navigator.webdriver
        context.add_init_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")

        if sp_dc:
            context.add_cookies([{
                "name": "sp_dc",
                "value": sp_dc,
                "domain": ".spotify.com",
                "path": "/",
                "httpOnly": True,
                "secure": True,
            }])

        page = context.new_page()
        page.goto(url, timeout=60000)

        # If Spotify redirects to the login page the cookie has expired
        if "accounts.spotify.com" in page.url or "/login" in page.url:
            print(json.dumps({"success": False, "error": "spotify_cookies_expired"}))
            sys.exit(0)
        time.sleep(3)

        expected_total = None
        expected_total_env = (os.environ.get("SPOTIFY_EXPECTED_TOTAL") or "").strip()
        if expected_total_env.isdigit():
            expected_total = int(expected_total_env)
        try:
            if expected_total is None:
                total_meta = page.locator('meta[name="music:song_count"]').first
                total_raw = total_meta.get_attribute("content")
                if total_raw and total_raw.isdigit():
                    expected_total = int(total_raw)
        except Exception:
            pass

        if expected_total is None:
            try:
                og_desc = page.locator('meta[property="og:description"]').first.get_attribute("content") or ""
                m = re.search(r"(\d[\d,]*)\s+items?", og_desc, re.IGNORECASE)
                if m:
                    expected_total = int(m.group(1).replace(",", ""))
            except Exception:
                expected_total = None

        # Accept cookie consent if present; this can block page rendering
        cookie_selectors = [
            "button:has-text('Accept cookies')",
            "button:has-text('Accept Cookies')",
            "button:has-text('ACCEPT COOKIES')",
            "[data-testid='cookie-policy-manage-dialog-accept-button']",
            "button.onetrust-close-btn-handler"
        ]
        for sel in cookie_selectors:
            try:
                btn = page.locator(sel).first
                if btn.count() == 0:
                    continue
                if not btn.is_visible(timeout=750):
                    continue
                print(f"DEBUG: Found cookie button with selector: {sel}", file=sys.stderr)
                btn.click(timeout=2000, force=True)
                time.sleep(0.5)
                break
            except Exception as e:
                print(f"DEBUG: Cookie selector {sel} failed: {e}", file=sys.stderr)

        # Wait for track list to load
        page.wait_for_selector(SELECTOR, timeout=30000)

        try:
            # Get playlist name from the page
            title_elem = page.query_selector('[data-testid="playlist-page"] h1')
            if not title_elem:
                title_elem = page.query_selector('[data-testid="entityTitle"] h1')
            if not title_elem:
                title_elem = page.query_selector('h1')
            if title_elem:
                name = title_elem.inner_text().strip()
                if name and name != "Your Library":
                    playlist_name = name
        except Exception:
            pass

        # Liked songs page uses the same row structure as a regular playlist.
        if is_liked_songs:
            playlist_name = "Liked Songs"

        # If we still don't know the count (the parent's HTTP probe comes back empty
        # now that the public page is a JS shell), read it off the rendered header.
        if not expected_total:
            expected_total = _detect_total_from_dom(page)
            if expected_total:
                print(f"DEBUG: detected {expected_total} tracks from rendered page", file=sys.stderr)

        # Guesstimate how long the full scroll should take from the count, with generous
        # slack for slow hosts, and never run past the parent's hard deadline. This stops
        # us spinning at the outer ceiling on a known-size list while still giving a huge
        # playlist plenty of room.
        working_deadline = hard_deadline_seconds
        if expected_total:
            est_with_slack = (base_overhead + expected_total * seconds_per_track) * 1.5
            working_deadline = min(hard_deadline_seconds or est_with_slack, est_with_slack)

        # Spotify uses virtualised scrolling; tracks get unloaded as you scroll.
        # Extract tracks incrementally while scrolling.
        seen_tracks_by_index = {}
        last_seen_count = 0
        last_progress_at = time.monotonic()

        def extract_visible_tracks():
            for row in page.query_selector_all(SELECTOR):
                try:
                    text = row.inner_text().strip()
                    parts = text.split(chr(10))
                    parts = [pt.strip() for pt in parts if pt.strip()]

                    # Only extract tracks that start with a numeric row index.
                    if not parts:
                        continue
                    track_index_raw = re.sub(r"[^\d]", "", parts[0])
                    if not track_index_raw:
                        continue
                    track_index = int(track_index_raw)

                    # Skip the track number
                    parts = parts[1:]

                    # Skip "E" for Explicit marker
                    if parts and parts[0] == "E":
                        parts = parts[1:]

                    if len(parts) >= 2:
                        track_name = parts[0].strip()
                        artist = parts[1].strip()
                        if artist == "E" and len(parts) >= 3:
                            artist = parts[2].strip()
                        if not track_name or artist == "E":
                            continue
                        # Music video rows show "Music Video" as the artist.
                        # The real artist follows a bullet (•) separator in the row parts,
                        # or may be extractable from the title field ("Artist - Title" pattern).
                        if artist.lower() == "music video":
                            # Scan parts for the bullet separator; artist is the element after it.
                            bullet_idx = next((i for i, p in enumerate(parts) if p == "\u2022"), None)
                            if bullet_idx is not None and bullet_idx + 1 < len(parts):
                                real_artist = parts[bullet_idx + 1].strip()
                                if real_artist and real_artist.lower() != "music video":
                                    seen_tracks_by_index[track_index] = f"{real_artist} - {track_name}"
                                    continue
                            # Fallback: try splitting title on a dash
                            dash = re.search(r"\s+[-\u2013\u2014]\s+", track_name)
                            if dash:
                                salvaged_artist = track_name[:dash.start()].strip()
                                salvaged_title = track_name[dash.end():].strip()
                                if salvaged_artist and salvaged_title:
                                    seen_tracks_by_index[track_index] = f"{salvaged_artist} - {salvaged_title}"
                            continue
                        seen_tracks_by_index[track_index] = f"{artist} - {track_name}"
                except Exception:
                    continue

        # First extraction before scrolling
        extract_visible_tracks()

        while True:
            # Out of time: hand back whatever we've got rather than let the parent
            # hard-kill us (which loses everything and drops back to the 100-track embed).
            if working_deadline and (time.monotonic() - _script_start) > working_deadline:
                if expected_total and len(seen_tracks_by_index) < expected_total:
                    print(
                        f"DEBUG: stopping at deadline ({working_deadline:.0f}s) with "
                        f"{len(seen_tracks_by_index)}/{expected_total} tracks",
                        file=sys.stderr,
                    )
                break
            # Reached the known total. Only trust counts above the embed cap as a hard
            # stop, so a misread small number can't truncate a big playlist.
            if expected_total and expected_total > EMBED_TRACK_LIMIT and len(seen_tracks_by_index) >= expected_total:
                break
            if time.monotonic() - last_progress_at > stall_timeout_seconds:
                break

            rows = page.query_selector_all(SELECTOR)
            if rows:
                rows[-1].scroll_into_view_if_needed()
            else:
                page.mouse.wheel(0, 2000)
            time.sleep(0.3)

            extract_visible_tracks()

            current_seen = len(seen_tracks_by_index)
            if current_seen > last_seen_count:
                last_seen_count = current_seen
                last_progress_at = time.monotonic()

        tracks = [seen_tracks_by_index[idx] for idx in sorted(seen_tracks_by_index)]
        if expected_total and len(tracks) < expected_total:
            print(
                f"DEBUG: Track extraction incomplete ({len(tracks)}/{expected_total})",
                file=sys.stderr,
            )
        browser.close()

    print(json.dumps({
        "success": True,
        "tracks": tracks,
        "playlist_name": playlist_name,
        "count": len(tracks),
        "expected_total": expected_total,
    }))

except Exception as e:
    print(json.dumps({"success": False, "error": str(e)}))