# Databricks notebook source
"""
Bronze -> Silver for TWO users' Spotify Extended Streaming History.

ASSUMED BRONZE: {catalog}.bronze.spotify_streaming_history
  raw export columns as landed (ts, ms_played, platform, conn_country,
  master_metadata_track_name, master_metadata_album_artist_name,
  master_metadata_album_album_name, spotify_track_uri, episode_name,
  episode_show_name, spotify_episode_uri, reason_start, reason_end,
  shuffle, skipped, offline, incognito_mode, ip_addr_decrypted, user_agent_decrypted)
  + ingestion columns: _user_label ('user_a'/'user_b', taken from the landing folder,
    NOT from the raw username field), _source_file, _ingested_at

Key facts baked in:
  * `ts` is when the play ENDED (UTC), so start = ts - ms_played.
  * Exports can overlap (re-requested downloads) -> dedupe.
  * Skips are signal, so flag them, don't delete them.
  * IP address and user agent are dropped in silver (privacy minimization).
"""
dbutils.widgets.text("catalog", "dev")
CAT = dbutils.widgets.get("catalog")

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {CAT}.silver")
spark.table(f"{CAT}.bronze.spotify_streaming_history").createOrReplaceTempView("bronze_v")

# Per-user home time zone (replace with your real config table)
spark.sql("""
    CREATE OR REPLACE TEMP VIEW user_config AS
    SELECT * FROM VALUES ('user_a', 'Europe/London'), ('user_b', 'America/New_York')
    AS t(user_id, home_tz)
""")

# COMMAND ----------
# MAGIC %md ## Step 1: type, rename, normalize (no rows lost yet)

# COMMAND ----------

spark.sql("""
CREATE OR REPLACE TEMP VIEW typed AS
SELECT
  _user_label                                            AS user_id,
  to_timestamp(ts)                                       AS ended_at_utc,
  CAST(ms_played AS BIGINT)                              AS ms_played,
  CASE WHEN lower(platform) LIKE '%ios%'     THEN 'ios'
       WHEN lower(platform) LIKE '%android%' THEN 'android'
       WHEN lower(platform) RLIKE 'windows|osx|mac|linux' THEN 'desktop'
       WHEN lower(platform) LIKE '%web%'     THEN 'web'
       ELSE 'other' END                                  AS platform,
  conn_country                                           AS country,
  CASE WHEN spotify_track_uri   IS NOT NULL THEN 'track'
       WHEN spotify_episode_uri IS NOT NULL THEN 'episode'
       ELSE 'other' END                                  AS content_type,
  NULLIF(regexp_extract(spotify_track_uri,   'spotify:track:([A-Za-z0-9]+)',   1), '') AS track_id,
  NULLIF(regexp_extract(spotify_episode_uri, 'spotify:episode:([A-Za-z0-9]+)', 1), '') AS episode_id,
  trim(master_metadata_track_name)                       AS track_name,
  trim(master_metadata_album_artist_name)                AS artist_name,
  trim(master_metadata_album_album_name)                 AS album_name,
  episode_show_name                                      AS show_name,
  reason_start, reason_end,
  COALESCE(shuffle, false)                               AS shuffle,
  skipped, offline,
  COALESCE(incognito_mode, false)                        AS incognito_mode,
  _source_file
FROM bronze_v
WHERE ts IS NOT NULL
  AND ms_played IS NOT NULL AND ms_played >= 0       -- impossible values out
""")

# COMMAND ----------
# MAGIC %md ## Step 2: dedupe overlapping exports + derive start time and flags

# COMMAND ----------

spark.sql("""
CREATE OR REPLACE TEMP VIEW cleaned AS
SELECT
  d.*,
  timestamp_millis(unix_millis(ended_at_utc) - ms_played)   AS started_at_utc,
  ms_played >= 30000                                         AS is_stream,      -- Spotify's 30s rule
  COALESCE(skipped, reason_end = 'fwdbtn' AND ms_played < 30000) AS is_skip,
  reason_end = 'trackdone'                                   AS is_complete,
  (content_type = 'track' AND ms_played > 3600000)           AS is_outlier,     -- >1h on one track
  from_utc_timestamp(timestamp_millis(unix_millis(ended_at_utc) - ms_played), c.home_tz)
                                                             AS started_at_local
FROM (
  SELECT * EXCEPT (rn) FROM (
    SELECT *, row_number() OVER (
             PARTITION BY user_id, ended_at_utc, track_id, episode_id
             ORDER BY ms_played DESC) AS rn
    FROM typed)
  WHERE rn = 1
) d
LEFT JOIN user_config c USING (user_id)
""")

# COMMAND ----------
# MAGIC %md ## Step 3: sessionize (30 min gap, per user) + sequence features

# COMMAND ----------

spark.sql("""
CREATE OR REPLACE TEMP VIEW sessioned AS
SELECT *,
  concat(user_id, '-', session_seq)                                           AS session_id,
  row_number() OVER (PARTITION BY user_id, session_seq ORDER BY started_at_utc) AS pos_in_session,
  lag(track_id) OVER (PARTITION BY user_id ORDER BY started_at_utc)           AS prev_track_id
FROM (
  SELECT *,
    SUM(new_session) OVER (PARTITION BY user_id ORDER BY started_at_utc
                           ROWS UNBOUNDED PRECEDING) AS session_seq
  FROM (
    SELECT *,
      CASE WHEN lag(ended_at_utc) OVER (PARTITION BY user_id ORDER BY started_at_utc) IS NULL
             OR started_at_utc > lag(ended_at_utc) OVER (PARTITION BY user_id ORDER BY started_at_utc)
                                 + INTERVAL 30 MINUTES
           THEN 1 ELSE 0 END AS new_session
    FROM cleaned
  )
)
""")

# COMMAND ----------
# MAGIC %md ## Step 4: discovery + cross-user features (what makes two users special)

# COMMAND ----------

spark.sql(f"""
CREATE OR REPLACE TABLE {CAT}.silver.spotify_plays AS
WITH first_play AS (                       -- first real listen per user and track
  SELECT user_id, track_id, MIN(started_at_utc) AS first_play_utc
  FROM sessioned WHERE is_stream AND track_id IS NOT NULL
  GROUP BY user_id, track_id
)
SELECT s.*,
  s.started_at_utc = fp.first_play_utc                          AS is_first_listen,
  other.first_play_utc                                          AS other_user_first_play_utc,
  other.first_play_utc IS NOT NULL
    AND other.first_play_utc < s.started_at_utc                 AS other_user_heard_first,
  hour(s.started_at_local)                                      AS hour_local,
  dayofweek(s.started_at_local) IN (1, 7)                       AS is_weekend,
  CAST(s.started_at_local AS DATE)                              AS play_date_local
FROM sessioned s
LEFT JOIN first_play fp    ON fp.user_id = s.user_id AND fp.track_id = s.track_id
LEFT JOIN first_play other ON other.track_id = s.track_id AND other.user_id <> s.user_id
""")

# COMMAND ----------
# MAGIC %md ## Step 5: data-quality checks (fail loudly)

# COMMAND ----------

dq = spark.sql(f"""
SELECT user_id,
       COUNT(*)                                              AS rows,
       MIN(started_at_utc) AS first_play, MAX(started_at_utc) AS last_play,
       ROUND(AVG(CAST(track_id IS NULL AND content_type = 'track' AS INT)), 4) AS null_track_share,
       SUM(CAST(is_outlier AS INT))                          AS outliers,
       SUM(CAST(started_at_utc > current_timestamp() AS INT)) AS future_rows
FROM {CAT}.silver.spotify_plays GROUP BY user_id
""")
dq.show(truncate=False)
assert dq.filter("future_rows > 0").count() == 0, "future timestamps found"
assert dq.count() == 2, "expected exactly two users"

# COMMAND ----------
# MAGIC %md ## Two-user analytics: taste overlap and cross-recommendation candidates

# COMMAND ----------

# Artist-level Jaccard and cosine similarity between the two users
spark.sql(f"""
WITH ua AS (
  SELECT user_id, artist_name, COUNT(*) AS plays
  FROM {CAT}.silver.spotify_plays
  WHERE is_stream AND content_type = 'track' AND artist_name IS NOT NULL
  GROUP BY 1, 2
),
shared AS (
  SELECT a.artist_name, a.plays AS pa, b.plays AS pb
  FROM ua a JOIN ua b ON a.artist_name = b.artist_name
  WHERE a.user_id = 'user_a' AND b.user_id = 'user_b'
),
sizes AS (
  SELECT SUM(CAST(user_id = 'user_a' AS INT)) AS n_a,
         SUM(CAST(user_id = 'user_b' AS INT)) AS n_b,
         SQRT(SUM(CASE WHEN user_id = 'user_a' THEN plays * plays END)) AS norm_a,
         SQRT(SUM(CASE WHEN user_id = 'user_b' THEN plays * plays END)) AS norm_b
  FROM ua
)
SELECT COUNT(*)                                            AS shared_artists,
       COUNT(*) / (MAX(n_a) + MAX(n_b) - COUNT(*))         AS jaccard,
       SUM(pa * pb) / (MAX(norm_a) * MAX(norm_b))          AS cosine_plays
FROM shared CROSS JOIN sizes
""").show()

# Candidates for user A: tracks B plays a lot that A has never played,
# preferring artists A already listens to
spark.sql(f"""
WITH aff AS (
  SELECT user_id, track_id, artist_name, MAX(track_name) AS track_name, COUNT(*) AS plays
  FROM {CAT}.silver.spotify_plays
  WHERE is_stream AND content_type = 'track' AND track_id IS NOT NULL
  GROUP BY 1, 2, 3
),
a_artists AS (SELECT artist_name, SUM(plays) AS plays FROM aff WHERE user_id = 'user_a' GROUP BY 1)
SELECT b.track_name, b.artist_name, b.plays AS b_plays,
       COALESCE(aa.plays, 0) AS a_plays_for_artist
FROM aff b
LEFT JOIN aff a_trk ON a_trk.user_id = 'user_a' AND a_trk.track_id = b.track_id
LEFT JOIN a_artists aa ON aa.artist_name = b.artist_name
WHERE b.user_id = 'user_b' AND a_trk.track_id IS NULL
ORDER BY (COALESCE(aa.plays, 0) > 0) DESC, b.plays DESC
LIMIT 20
""").show(truncate=False)