#!/usr/bin/env python3
"""
tracklist_count_check.py

For each playlist, takes a live Spotify snapshot (ordered list of positions
with track/ISRC/artist) and checks whether the insights tracklist tables cover
the same tracks.  Spotify is the source of truth — we are measuring gaps in
our coverage of what Spotify actually has.

Run the setup SQL first:
  tracklist_count_check_setup.sql

Usage:
  python tracklist_count_check.py
  python tracklist_count_check.py --stale-hours 48   # re-check rows older than 48h
  python tracklist_count_check.py --limit 200        # cap rows per run
  python tracklist_count_check.py --source hourly    # only check hourly playlists
  python tracklist_count_check.py --source priority  # only check priority playlists
  python tracklist_count_check.py --min-delta 3      # only report deltas >= 3 in summary
  python tracklist_count_check.py --refresh            # force-refresh insights counts (task keeps these current)
  python tracklist_count_check.py --workers 5         # parallel Spotify workers (default 5)
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import re
import sys
import time
import threading
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path

try:
    from dotenv import load_dotenv
except ImportError:
    sys.exit("Error: python-dotenv is required.\n  pip install python-dotenv")

load_dotenv(Path(__file__).parent / ".env")

try:
    from cryptography.hazmat.backends import default_backend
    from cryptography.hazmat.primitives import serialization
    import snowflake.connector
except ImportError:
    sys.exit(
        "Error: snowflake-connector-python and cryptography are required.\n"
        "  pip install snowflake-connector-python cryptography"
    )

# ── Spotify ────────────────────────────────────────────────────────────────────

TOKEN_URL       = "https://accounts.spotify.com/api/token"
API_BASE        = "https://api.spotify.com/v1"
REQUEST_TIMEOUT = 10

# ── Snowflake ──────────────────────────────────────────────────────────────────

SNOWFLAKE_TABLE = "FACTS.PROD.PLAYLIST_TRACK_COUNT_CHECK"
DETAIL_TABLE    = "FACTS.PROD.PLAYLIST_TRACK_COUNT_CHECK_DETAIL"

CREATE_TABLE = f"""
    CREATE TABLE IF NOT EXISTS {SNOWFLAKE_TABLE} (
        SOURCE                              VARCHAR(20)   NOT NULL,
        STORE_PLAYLIST_ID                   VARCHAR(100)  NOT NULL,
        INSIGHTS_TRACK_COUNT                INTEGER,
        SPOTIFY_TRACK_COUNT                 INTEGER,
        SNAPSHOT_MATCHES                    BOOLEAN,
        TRACKS_MISSING_FROM_INSIGHTS        INTEGER,   -- unique ISRCs on Spotify not found anywhere in Insights
        POSITIONS_NOT_MATCHING              INTEGER,   -- positions where status is WRONG_TRACK or MISSING
        DIFFERENT_ISRC_SAME_NAME_AND_ARTIST INTEGER,
        SPOTIFY_NOT_FOUND                   BOOLEAN,
        SPOTIFY_SNAPSHOT_ID                 VARCHAR(100),
        SPOTIFY_CHECKED_AT                  TIMESTAMP_NTZ,
        INSIGHTS_CHECKED_AT                 TIMESTAMP_NTZ,
        INSIGHTS_LATEST_TRACK_DATE          DATE,
        CHARTMETRIC_PLAYLIST_ID             INTEGER,
        PRIMARY KEY (SOURCE, STORE_PLAYLIST_ID)
    )
"""

CREATE_DETAIL_TABLE = f"""
    CREATE TABLE IF NOT EXISTS {DETAIL_TABLE} (
        SOURCE                              VARCHAR(20)   NOT NULL,
        STORE_PLAYLIST_ID                   VARCHAR(100)  NOT NULL,
        POSITION                            INTEGER       NOT NULL,
        STATUS                              VARCHAR(20)   NOT NULL,  -- MATCH / DIFFERENT_ISRC / WRONG_TRACK / MISSING
        SPOTIFY_ISRC                        VARCHAR(20),
        SPOTIFY_TRACK_ID                    VARCHAR(50),
        SPOTIFY_TRACK_NAME                  VARCHAR(500),
        SPOTIFY_ARTIST_NAME                 VARCHAR(500),
        SPOTIFY_ADDED_AT                    TIMESTAMP_NTZ,
        INSIGHTS_ISRC                       VARCHAR(20),
        INSIGHTS_TRACK_NAME                 VARCHAR(500),
        INSIGHTS_ARTIST_NAME                VARCHAR(500),
        SPOTIFY_CHECKED_AT                  TIMESTAMP_NTZ NOT NULL,
        INSIGHTS_CHECKED_AT                 TIMESTAMP_NTZ,
        PRIMARY KEY (SOURCE, STORE_PLAYLIST_ID, POSITION)
    )
"""

DELETE_DETAIL_ROWS = f"""
    DELETE FROM {DETAIL_TABLE}
    WHERE source = %s AND store_playlist_id = %s
"""

INSERT_DETAIL_ROW = f"""
    INSERT INTO {DETAIL_TABLE}
        (source, store_playlist_id, position, status,
         spotify_isrc, spotify_track_id, spotify_track_name, spotify_artist_name, spotify_added_at,
         insights_isrc, insights_track_name, insights_artist_name,
         spotify_checked_at, insights_checked_at)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP(), %s)
"""


REFRESH_INSIGHTS_COUNTS = f"""
    MERGE INTO {SNOWFLAKE_TABLE} AS tgt
    USING (
        -- Start from the full playlist universe so playlists with zero Insights tracks
        -- still get a row (insights_track_count = 0). Without this, they are invisible.
        WITH all_playlists AS (
            SELECT store_playlist_id, 'hourly' AS source, chartmetric_playlist_id
            FROM FACTS.PROD.HOURLY_PLAYLISTS
            WHERE store_id = 286
            UNION ALL
            SELECT store_playlist_id, 'priority' AS source, chartmetric_playlist_id
            FROM FACTS.PROD.PRIORITY_PLAYLISTS
            WHERE store_id = 286
        ),
        hourly_counts AS (
            SELECT
                store_playlist_id,
                -- Cap at playlist_track_count to match ows-playlist behaviour:
                -- editorial playlists are already capped by the position filter;
                -- personalized/algorithmic playlists have no positions so we cap the ISRC count.
                LEAST(
                    COUNT(DISTINCT IFF(
                        playlist_type IN ('PERSONALIZED', 'ALGORITHMIC', 'STATION', 'RADIO'),
                        isrc,
                        current_position::VARCHAR
                    )),
                    MAX(playlist_track_count)
                ) AS insights_track_count,
                MAX(last_added_on_date) AS insights_latest_track_date
            FROM FACTS.PROD.PLAYLISTS_HOURLY_BY_PLAYLIST_CURRENT_TRACKLIST
            WHERE store_id = 286
              AND last_added_on_date IS NOT NULL
              AND (removed_on IS NULL OR removed_on < last_added_on_date)
              AND (
                  playlist_type IN ('PERSONALIZED', 'ALGORITHMIC', 'STATION', 'RADIO')
                  OR (current_position IS NOT NULL AND current_position <= playlist_track_count)
              )
            GROUP BY store_playlist_id
        ),
        priority_counts AS (
            SELECT
                store_playlist_id,
                LEAST(
                    COUNT(DISTINCT IFF(
                        playlist_type IN ('PERSONALIZED', 'ALGORITHMIC', 'STATION', 'RADIO'),
                        isrc,
                        current_position::VARCHAR
                    )),
                    MAX(playlist_track_count)
                ) AS insights_track_count,
                MAX(last_added_on_date) AS insights_latest_track_date
            FROM FACTS.PROD.PLAYLISTS_PRIORITY_BY_PLAYLIST_CURRENT_TRACKLIST
            WHERE last_added_on_date IS NOT NULL
              AND (removed_on IS NULL OR removed_on < last_added_on_date)
              AND (
                  playlist_type IN ('PERSONALIZED', 'ALGORITHMIC', 'STATION', 'RADIO')
                  OR (current_position IS NOT NULL AND current_position <= playlist_track_count)
              )
            GROUP BY store_playlist_id
        )
        SELECT
            ap.source,
            ap.store_playlist_id,
            ap.chartmetric_playlist_id,
            COALESCE(
                IFF(ap.source = 'hourly', hc.insights_track_count, pc.insights_track_count),
                0
            ) AS insights_track_count,
            IFF(ap.source = 'hourly', hc.insights_latest_track_date, pc.insights_latest_track_date)
                AS insights_latest_track_date
        FROM all_playlists ap
        LEFT JOIN hourly_counts   hc ON ap.source = 'hourly'   AND hc.store_playlist_id = ap.store_playlist_id
        LEFT JOIN priority_counts pc ON ap.source = 'priority' AND pc.store_playlist_id = ap.store_playlist_id
    ) AS src
    ON  tgt.source            = src.source
    AND tgt.store_playlist_id = src.store_playlist_id
    WHEN MATCHED AND (
        tgt.insights_track_count       IS DISTINCT FROM src.insights_track_count
        OR tgt.insights_latest_track_date IS DISTINCT FROM src.insights_latest_track_date
        OR tgt.chartmetric_playlist_id IS DISTINCT FROM src.chartmetric_playlist_id
    ) THEN
        UPDATE SET insights_track_count        = src.insights_track_count,
                   insights_latest_track_date  = src.insights_latest_track_date,
                   chartmetric_playlist_id     = src.chartmetric_playlist_id
    WHEN NOT MATCHED THEN
        INSERT (source, store_playlist_id, chartmetric_playlist_id, insights_track_count, insights_latest_track_date)
        VALUES (src.source, src.store_playlist_id, src.chartmetric_playlist_id, src.insights_track_count, src.insights_latest_track_date)
"""

SELECT_PENDING = f"""
    SELECT s.source, s.store_playlist_id, s.insights_track_count
    FROM   {SNOWFLAKE_TABLE} s
    LEFT JOIN (
        SELECT store_playlist_id, 'hourly' AS source, playlist_follower_count AS followers
        FROM FACTS.PROD.HOURLY_PLAYLIST_METADATA WHERE store_id = 286
        UNION ALL
        SELECT store_playlist_id, 'priority' AS source, playlist_follower_count AS followers
        FROM FACTS.PROD.PRIORITY_PLAYLIST_METADATA WHERE store_id = 286
    ) m ON m.source = s.source AND m.store_playlist_id = s.store_playlist_id
    WHERE  (
               s.spotify_checked_at IS NULL
            OR s.spotify_checked_at < DATEADD(hour, -%s, CURRENT_TIMESTAMP())
           )
    {{source_filter}}
    {{playlist_id_filter}}
    ORDER BY s.spotify_checked_at ASC NULLS FIRST, COALESCE(m.followers, 0) DESC
    {{limit_clause}}
"""

UPDATE_ROW = f"""
    UPDATE {SNOWFLAKE_TABLE}
    SET    spotify_track_count                = %s,
           snapshot_matches                   = %s,
           tracks_missing_from_insights       = %s,
           positions_not_matching             = %s,
           different_isrc_same_name_and_artist = %s,
           spotify_not_found                  = %s,
           spotify_snapshot_id               = %s,
           spotify_checked_at                = CURRENT_TIMESTAMP(),
           insights_checked_at               = %s
    WHERE  source            = %s
      AND  store_playlist_id = %s
"""

# ── Spotify helpers ────────────────────────────────────────────────────────────

def _fetch_token() -> str:
    client_id     = os.environ.get("SPOTIFY_CLIENT_ID")
    client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET")
    if not client_id or not client_secret:
        sys.exit("Error: set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET env vars")

    credentials = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
    data = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode()
    req  = urllib.request.Request(
        TOKEN_URL,
        data=data,
        headers={"Authorization": f"Basic {credentials}"},
    )
    with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
        return json.loads(resp.read())["access_token"]


class TokenManager:
    """Thread-safe Spotify token with automatic refresh on 401."""
    def __init__(self):
        self._lock = threading.Lock()
        self._token = _fetch_token()

    @property
    def token(self) -> str:
        return self._token

    def refresh(self, old_token: str) -> str:
        with self._lock:
            if self._token == old_token:
                self._token = _fetch_token()
            return self._token



def get_spotify_tracks(
    token_mgr: TokenManager, playlist_id: str
) -> tuple[str | None, int | None, list[tuple], bool]:
    """
    Fetches snapshot_id and all tracks for a playlist from Spotify, returning:
      (snapshot_id, total, track_rows, not_found)

      snapshot_id — Spotify's opaque playlist version identifier
      total       — tracks.total from API
      track_rows  — list of (position, isrc, spotify_track_id, track_name, artist_name, added_at)
                    in playlist order (1-based); fields may be None for local files/podcasts
      not_found   — True if playlist returned 404

    Handles 401 internally via token_mgr.refresh().
    """
    # First request hits the playlist endpoint to get snapshot_id + first page of tracks
    first_fields = urllib.parse.urlencode({
        "fields": "snapshot_id,tracks(total,items(added_at,track(id,name,artists(name),external_ids(isrc))),next)",
        "limit":  100,
    })
    url        = f"{API_BASE}/playlists/{playlist_id}?{first_fields}"
    track_rows = []
    total       = None
    snapshot_id = None
    position    = 1  # 1-based
    is_first    = True
    token       = token_mgr.token

    while url:
        req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
        try:
            with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
                data = json.loads(resp.read())
        except urllib.error.HTTPError as e:
            if e.code == 404:
                return (None, None, [], True)
            if e.code == 401:
                token = token_mgr.refresh(token)
                continue
            if e.code == 429:
                wait = int(e.headers.get("Retry-After", 5))
                print(f"\n  Rate limited — waiting {wait}s...", flush=True)
                time.sleep(wait)
                continue
            if e.code >= 500:
                time.sleep(1)
                continue
            raise
        except (urllib.error.URLError, TimeoutError, OSError):
            print(f"\n  Network error on {playlist_id}, skipping.", flush=True)
            return (None, None, [], False)

        if is_first:
            snapshot_id = data.get("snapshot_id")
            page        = data.get("tracks", {})
            is_first    = False
        else:
            page = data

        if total is None:
            total = page.get("total")

        for item in page.get("items") or []:
            item       = item or {}
            track      = item.get("track") or {}
            isrc       = (track.get("external_ids") or {}).get("isrc")
            track_id   = track.get("id")
            track_name = track.get("name")
            artist_name = ", ".join(a["name"] for a in track.get("artists") or [] if a.get("name"))
            added_at   = item.get("added_at")  # ISO 8601 string or None
            track_rows.append((
                position,
                isrc.upper().replace("-", "") if isrc else None,
                track_id,
                track_name,
                artist_name or None,
                added_at,
            ))
            position += 1

        url = page.get("next")
        # Fallback: if Spotify didn't return a next URL (fields-filtered responses
        # sometimes omit it) but we know there are more tracks, construct the URL.
        if url is None and total is not None and len(track_rows) < total:
            url = f"{API_BASE}/playlists/{playlist_id}/tracks?offset={len(track_rows)}&limit=100"

    return (snapshot_id, total, track_rows, False)


# ── Snowflake helpers ──────────────────────────────────────────────────────────

def get_snowflake_connection():
    params = {
        "account":   os.getenv("SNOWFLAKE_ACCOUNT", "orchard"),
        "user":      os.getenv("SNOWFLAKE_USER"),
        "database":  "FACTS",
        "schema":    "PROD",
        "warehouse": os.getenv("SNOWFLAKE_WAREHOUSE", "DEV_PERFORMANCE_WAREHOUSE"),
    }

    key_path = os.getenv("SNOWFLAKE_PRIVATE_KEY_PATH")
    if not key_path:
        sys.exit("Error: SNOWFLAKE_PRIVATE_KEY_PATH env var is required")
    key_file = Path(key_path).expanduser()
    if not key_file.exists():
        sys.exit(f"Error: private key file not found: {key_file}")

    passphrase = os.getenv("SNOWFLAKE_KEY_PASSPHRASE")
    with open(key_file, "rb") as f:
        p_key = serialization.load_pem_private_key(
            f.read(),
            password=passphrase.encode() if passphrase else None,
            backend=default_backend(),
        )
    params["private_key"] = p_key.private_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )

    return snowflake.connector.connect(**params)


def normalize_name(s: str | None) -> str:
    return re.sub(r"\s*,\s*", ",", (s or "").lower().strip())


def load_insights_tracks(
    cursor, rows: list
) -> dict[tuple[str, str], list[tuple[int | None, str, str | None, str | None]]]:
    """
    Batch-loads tracks from the insights tracklist tables for all playlists in `rows`.

    Returns a dict keyed by (source, store_playlist_id), where each value is a list of
    (current_position, isrc, track_name, artist_name).

    current_position is None for algorithmic/personalized playlists (no ordered positions).
    For editorial playlists each position is the track's slot in the playlist.
    """
    hourly_ids   = [pid for src, pid, _ in rows if src == "hourly"]
    priority_ids = [pid for src, pid, _ in rows if src == "priority"]

    result: dict[tuple[str, str], list[tuple[int | None, str, str | None, str | None]]] = {}

    sources = [
        ("hourly",   "FACTS.PROD.PLAYLISTS_HOURLY_BY_PLAYLIST_CURRENT_TRACKLIST",   hourly_ids,   "store_id = 286 AND "),
        ("priority", "FACTS.PROD.PLAYLISTS_PRIORITY_BY_PLAYLIST_CURRENT_TRACKLIST", priority_ids, ""),
    ]

    for source, table, ids, extra_where in sources:
        if not ids:
            continue
        ph = ", ".join(["%s"] * len(ids))
        # Replicate ows-playlist deduplication:
        #   1. raw: collapse per-country rows → one row per (playlist, position, isrc)
        #   2. dedup_isrc: keep one row per (playlist, isrc), preferring lowest position then most recent
        #   3. dedup_position: keep one row per (playlist, position), preferring most recent then isrc asc
        #      (for personalized/algorithmic playlists current_position is NULL → all pass through as 1)
        #   4. capped: row-number the result and cap at playlist_track_count
        query = (
            f"WITH raw AS ("
            f"  SELECT store_playlist_id, current_position, isrc,"
            f"      MAX(chartmetric_track_name) AS track_name,"
            f"      MAX(chartmetric_artist_name) AS artist_name,"
            f"      MAX(last_added_on_date) AS last_added_on_date,"
            f"      MAX(playlist_track_count) AS playlist_track_count"
            f"  FROM {table}"
            f"  WHERE {extra_where}last_added_on_date IS NOT NULL"
            f"  AND (removed_on IS NULL OR removed_on < last_added_on_date)"
            f"  AND ("
            f"      playlist_type IN ('PERSONALIZED','ALGORITHMIC','STATION','RADIO')"
            f"      OR (current_position IS NOT NULL AND current_position <= playlist_track_count)"
            f"  )"
            f"  AND store_playlist_id IN ({ph})"
            f"  GROUP BY store_playlist_id, current_position, isrc"
            f"),"
            f"dedup_isrc AS ("
            f"  SELECT *,"
            f"      ROW_NUMBER() OVER ("
            f"          PARTITION BY store_playlist_id, isrc"
            f"          ORDER BY current_position ASC NULLS LAST, last_added_on_date DESC NULLS LAST"
            f"      ) AS rn_isrc"
            f"  FROM raw"
            f"),"
            f"dedup_position AS ("
            f"  SELECT *,"
            f"      CASE WHEN current_position IS NULL THEN 1"
            f"           ELSE ROW_NUMBER() OVER ("
            f"               PARTITION BY store_playlist_id, current_position"
            f"               ORDER BY last_added_on_date DESC NULLS LAST, isrc ASC"
            f"           )"
            f"      END AS rn_position"
            f"  FROM dedup_isrc WHERE rn_isrc = 1"
            f"),"
            f"capped AS ("
            f"  SELECT *,"
            f"      ROW_NUMBER() OVER ("
            f"          PARTITION BY store_playlist_id"
            f"          ORDER BY current_position ASC NULLS LAST, isrc ASC"
            f"      ) AS row_num,"
            f"      MAX(COALESCE(playlist_track_count, 999999)) OVER (PARTITION BY store_playlist_id) AS max_ptc"
            f"  FROM dedup_position WHERE rn_position = 1"
            f")"
            f"SELECT store_playlist_id, current_position, isrc, track_name, artist_name"
            f" FROM capped WHERE row_num <= max_ptc"
        )
        cursor.execute(query, ids)
        for pid, position, isrc, track_name, artist_name in cursor.fetchall():
            if not isrc:
                continue
            key = (source, pid)
            if key not in result:
                result[key] = []
            result[key].append((position, isrc.upper().replace("-", ""), track_name, artist_name))

    return result


def fmt_eta(seconds: float) -> str:
    m, s = divmod(int(seconds), 60)
    h, m = divmod(m, 60)
    if h:
        return f"{h}h{m:02d}m"
    if m:
        return f"{m}m{s:02d}s"
    return f"{s}s"


# ── Per-playlist comparison (pure function, no DB access) ─────────────────────

def compare_playlist(
    source: str,
    pid: str,
    track_rows: list[tuple],
    spotify_total: int,
    snapshot_id: str,
    insights_tracks: list[tuple[int | None, str, str | None, str | None]],
    insights_checked_at: datetime,
) -> dict:
    """Compare Spotify tracks against Insights data. Returns a result dict."""
    # Build Spotify position → (isrc, track_name, artist_name, track_id, added_at)
    sp_by_pos: dict[int, tuple[str | None, str | None, str | None, str | None, str | None]] = {}
    for pos, isrc, track_id, track_name, artist_name, added_at in track_rows:
        sp_by_pos[pos] = (isrc, track_name, artist_name, track_id, added_at)

    # Build insights position → (isrc, track_name, artist_name)
    ins_by_pos: dict[int, tuple[str, str | None, str | None]] = {}
    ins_isrc_set: set[str] = set()
    for pos, isrc, tname, aname in insights_tracks:
        ins_isrc_set.add(isrc)
        if pos is not None:
            ins_by_pos[pos] = (isrc, tname, aname)

    is_positional = len(ins_by_pos) > 0

    # per-position result: pos → (status, ins_isrc, ins_tname, ins_aname)
    pos_result: dict[int, tuple[str, str | None, str | None, str | None]] = {}
    diff_isrc_same_name: list[tuple[str, str]] = []
    missing_n = 0

    if is_positional:
        for pos, (sp_isrc, sp_tname, sp_aname, _, _) in sp_by_pos.items():
            if pos in ins_by_pos:
                ins_isrc, ins_tname, ins_aname = ins_by_pos[pos]
                if sp_isrc and sp_isrc == ins_isrc:
                    pos_result[pos] = ("MATCH", ins_isrc, ins_tname, ins_aname)
                elif sp_isrc and ins_isrc and (
                    normalize_name(sp_tname) + "|" + normalize_name(sp_aname) ==
                    normalize_name(ins_tname) + "|" + normalize_name(ins_aname)
                ):
                    diff_isrc_same_name.append((ins_isrc, sp_isrc))
                    pos_result[pos] = ("DIFFERENT_ISRC", ins_isrc, ins_tname, ins_aname)
                else:
                    missing_n += 1
                    pos_result[pos] = ("WRONG_TRACK", ins_isrc, ins_tname, ins_aname)
            else:
                missing_n += 1
                pos_result[pos] = ("MISSING", None, None, None)
    else:
        # Algorithmic playlist: ISRC set comparison
        sp_isrc_set = {v[0] for v in sp_by_pos.values() if v[0]}
        ins_by_name: dict[str, tuple[str, str | None, str | None]] = {}
        for _, isrc, tname, aname in insights_tracks:
            key = normalize_name(tname) + "|" + normalize_name(aname)
            if key and key != "|":
                ins_by_name[key] = (isrc, tname, aname)
        sub_isrc_map: dict[str, tuple[str, str | None, str | None]] = {}
        for sp_isrc in sp_isrc_set - ins_isrc_set:
            sp_tname, sp_aname = next(
                ((v[1], v[2]) for v in sp_by_pos.values() if v[0] == sp_isrc), (None, None)
            )
            key = normalize_name(sp_tname) + "|" + normalize_name(sp_aname)
            if key in ins_by_name:
                ins_isrc, ins_tname, ins_aname = ins_by_name[key]
                diff_isrc_same_name.append((ins_isrc, sp_isrc))
                sub_isrc_map[sp_isrc] = (ins_isrc, ins_tname, ins_aname)
            else:
                missing_n += 1
        for pos, (sp_isrc, _, _, _, _) in sp_by_pos.items():
            if sp_isrc in sub_isrc_map:
                ins_isrc, ins_tname, ins_aname = sub_isrc_map[sp_isrc]
                pos_result[pos] = ("DIFFERENT_ISRC", ins_isrc, ins_tname, ins_aname)
            elif sp_isrc and sp_isrc in ins_isrc_set:
                pos_result[pos] = ("MATCH", sp_isrc, None, None)
            else:
                pos_result[pos] = ("MISSING", None, None, None)

    sp_isrc_set_unique = {v[0] for v in sp_by_pos.values() if v[0]}
    truly_missing_count = max(0, len(sp_isrc_set_unique - ins_isrc_set) - len(diff_isrc_same_name))

    # Build detail rows
    detail_rows = []
    for pos, isrc, track_id, track_name, artist_name, added_at in track_rows:
        status, ins_isrc, ins_tname, ins_aname = pos_result.get(pos, ("MISSING", None, None, None))
        detail_rows.append((
            source, pid, pos, status,
            isrc, track_id, track_name, artist_name, added_at,
            ins_isrc, ins_tname, ins_aname,
            insights_checked_at,
        ))

    return {
        "source": source,
        "pid": pid,
        "spotify_total": spotify_total,
        "snapshot_id": snapshot_id,
        "snapshot_matches": missing_n == 0 and len(diff_isrc_same_name) == 0,
        "truly_missing_count": truly_missing_count,
        "positions_not_matching": missing_n,
        "diff_isrc_count": len(diff_isrc_same_name),
        "diff_isrc_same_name": diff_isrc_same_name,
        "detail_rows": detail_rows,
        "sp_by_pos": sp_by_pos,
    }


def process_one_playlist(
    token_mgr: TokenManager,
    source: str,
    pid: str,
    insights_tracks: list[tuple[int | None, str, str | None, str | None]],
    insights_checked_at: datetime,
    old_snapshot_id: str | None,
    force: bool,
) -> dict:
    """
    Fetch from Spotify + compare. Returns a result dict with a 'status' key:
      'not_found', 'skipped', 'snapshot_match', or 'compared'.
    """
    snapshot_id, spotify_total, track_rows, not_found_flag = get_spotify_tracks(token_mgr, pid)

    if not_found_flag:
        return {"status": "not_found", "source": source, "pid": pid}

    if spotify_total is None:
        return {"status": "skipped", "source": source, "pid": pid}

    if not force and old_snapshot_id == snapshot_id:
        return {"status": "snapshot_match", "source": source, "pid": pid}

    result = compare_playlist(
        source, pid, track_rows, spotify_total, snapshot_id,
        insights_tracks, insights_checked_at,
    )
    result["status"] = "compared"
    return result


# ── Main ───────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="Compare insights tracklist ISRCs against current Spotify snapshot"
    )
    parser.add_argument("--stale-hours", type=int, default=24,
                        help="Re-check rows last checked more than N hours ago (default 24)")
    parser.add_argument("--limit", type=int, default=0,
                        help="Max rows to process per run, 0 = no limit (default 0)")
    parser.add_argument("--source", choices=["hourly", "priority"],
                        help="Restrict to a single source (default: both)")
    parser.add_argument("--min-delta", type=int, default=1,
                        help="Minimum absolute ISRC set-difference to flag in summary (default 1)")
    parser.add_argument("--refresh", action="store_true",
                        help="Refresh insights counts from Snowflake before checking (task keeps these current)")
    parser.add_argument("--playlist-id",
                        help="Check a specific playlist ID only (bypasses stale-hours check)")
    parser.add_argument("--force", action="store_true",
                        help="Re-fetch and rewrite detail rows even if Spotify snapshot_id hasn't changed")
    parser.add_argument("--workers", type=int, default=5,
                        help="Number of parallel Spotify API workers (default 5)")
    args = parser.parse_args()

    source_filter      = f"AND s.source = '{args.source}'" if args.source else ""
    playlist_id_filter = f"AND s.store_playlist_id = '{args.playlist_id}'" if args.playlist_id else ""
    limit_clause       = f"LIMIT {args.limit}" if args.limit else ""

    if args.playlist_id:
        # Bypass stale-hours check when a specific playlist is requested
        query = f"""
            SELECT source, store_playlist_id, insights_track_count
            FROM   {SNOWFLAKE_TABLE}
            WHERE  store_playlist_id = '{args.playlist_id}'
            {source_filter}
        """
    else:
        query = SELECT_PENDING.format(
            source_filter=source_filter,
            playlist_id_filter=playlist_id_filter,
            limit_clause=limit_clause,
        )

    print("Connecting to Snowflake...")
    conn   = get_snowflake_connection()
    cursor = conn.cursor()

    if args.refresh:
        print("Refreshing Insights track counts from Snowflake...", flush=True)
        cursor.execute(REFRESH_INSIGHTS_COUNTS)
        print("Done.")

    if args.playlist_id:
        cursor.execute(query)
    else:
        cursor.execute(query, (args.stale_hours,))
    rows = cursor.fetchall()  # [(source, store_playlist_id, insights_track_count), ...]

    if not rows:
        print(f"\nNo playlists need checking (stale threshold: {args.stale_hours}h).")
        cursor.close()
        conn.close()
        return

    print(f"\nGetting the Insights tracklists for {len(rows)} playlists from Snowflake...", flush=True)
    insights_track_map = load_insights_tracks(cursor, rows)

    # Pre-load all existing snapshot IDs in one scan — filter in Python
    snapshot_ids: dict[tuple[str, str], str | None] = {}
    if not args.force:
        print("Loading existing snapshot IDs...", flush=True)
        cursor.execute(
            f"SELECT source, store_playlist_id, spotify_snapshot_id FROM {SNOWFLAKE_TABLE}"
        )
        for src, pid, snap_id in cursor.fetchall():
            snapshot_ids[(src, pid)] = snap_id

    print(f"Done. Starting {args.workers} workers...\n")
    token_mgr = TokenManager()

    matched        = 0
    has_gaps       = 0
    diff_isrc_only = 0
    not_found      = 0
    skipped        = 0
    done           = 0
    start          = time.time()

    discrepancies: list[tuple[str, str, int, int, list[tuple[str, str]], dict]] = []

    insights_checked_at = datetime.now(timezone.utc)

    # Submit all playlists to the thread pool
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {}
        for source, pid, insights_count in rows:
            fut = pool.submit(
                process_one_playlist,
                token_mgr,
                source,
                pid,
                insights_track_map.get((source, pid), []),
                insights_checked_at,
                snapshot_ids.get((source, pid)),
                args.force,
            )
            futures[fut] = (source, pid)

        for fut in as_completed(futures):
            done += 1
            elapsed = time.time() - start
            rate    = done / elapsed if elapsed > 0 else 0
            eta     = (len(rows) - done) / rate if rate > 0 else 0
            source, pid = futures[fut]
            print(f"  [{done:>5}/{len(rows)}] {pid}  ETA {fmt_eta(eta)}   ", end="\r", flush=True)

            result = fut.result()
            status = result["status"]

            if status == "not_found":
                cursor.execute(UPDATE_ROW, (None, None, None, None, None, True, None, insights_checked_at, result["source"], result["pid"]))
                not_found += 1
                continue

            if status == "skipped":
                skipped += 1
                continue

            if status == "snapshot_match":
                matched += 1
                continue

            # status == "compared"
            truly_missing = result["truly_missing_count"]
            positions_nm  = result["positions_not_matching"]
            diff_count    = result["diff_isrc_count"]

            if positions_nm == 0 and diff_count == 0:
                matched += 1
            elif truly_missing > 0:
                has_gaps += 1
            else:
                diff_isrc_only += 1

            if truly_missing >= args.min_delta:
                discrepancies.append((
                    result["source"], result["pid"], truly_missing, positions_nm,
                    result["diff_isrc_same_name"], result["sp_by_pos"],
                ))

            # Write summary row
            cursor.execute(UPDATE_ROW, (
                result["spotify_total"],
                result["snapshot_matches"],
                truly_missing,
                positions_nm,
                diff_count,
                False,
                result["snapshot_id"],
                insights_checked_at,
                result["source"],
                result["pid"],
            ))

            # Replace detail rows
            cursor.execute(DELETE_DETAIL_ROWS, (result["source"], result["pid"]))
            if result["detail_rows"]:
                cursor.executemany(INSERT_DETAIL_ROW, result["detail_rows"])

    print()  # clear progress line

    elapsed = time.time() - start
    print(f"\n{'='*60}")
    print(f"  Checked            : {len(rows)}")
    print(f"  Snapshot matches   : {matched}")
    print(f"  Has gaps           : {has_gaps}")
    print(f"  Different ISRC only: {diff_isrc_only}")
    print(f"  Not found (404)    : {not_found}")
    print(f"  Skipped (network)  : {skipped}")
    print(f"  Elapsed            : {fmt_eta(elapsed)}")
    print(f"  Workers            : {args.workers}")
    print(f"{'='*60}")

    if discrepancies:
        discrepancies.sort(key=lambda r: r[2], reverse=True)
        print(f"\n{len(discrepancies)} playlists missing {args.min_delta}+ tracks from insights:")
        for src, pid, truly_missing, positions_not_matching, diff_isrc, _ in discrepancies:
            sub_note = f"  [{len(diff_isrc)} different ISRC, same name and artist]" if diff_isrc else ""
            pos_note = f"  ({positions_not_matching} positions not matching)" if positions_not_matching != truly_missing else ""
            print(f"\n  {pid} ({src}): {truly_missing} tracks missing{sub_note}{pos_note}")

    cursor.close()
    conn.close()


if __name__ == "__main__":
    main()