Chapter 16: Building the Spotify Music Time Machine

pip install spotipy python-dotenv --break-system-packages

SPOTIPY_CLIENT_ID=your_client_id_here
SPOTIPY_CLIENT_SECRET=your_client_secret_here
SPOTIPY_REDIRECT_URI=http://localhost:8888/callback

.env
__pycache__/
*.pyc
.cache
*.db

"""
Quick Start: Create your first Spotify playlist
This script demonstrates OAuth authentication and basic API usage
"""
import os
from dotenv import load_dotenv
import spotipy
from spotipy.oauth2 import SpotifyOAuth

# Load credentials from .env file
load_dotenv()

# Define the permissions we need
scope = "user-top-read playlist-modify-public playlist-modify-private"

# Create Spotify client with OAuth
sp = spotipy.Spotify(auth_manager=SpotifyOAuth(scope=scope))

# Get current user info
user = sp.current_user()
print(f"Authenticated as: {user['display_name']}")

# Fetch top tracks from last 4 weeks
print("\nFetching your top tracks...")
top_tracks = sp.current_user_top_tracks(limit=20, time_range='short_term')

# Extract track URIs (Spotify's unique identifiers)
track_uris = [track['uri'] for track in top_tracks['items']]
track_names = [track['name'] for track in top_tracks['items']]

# Create a new playlist
print("\nCreating playlist...")
playlist = sp.user_playlist_create(
    user=user['id'],
    name="My Top Tracks - Quick Start",
    public=False,
    description="Created by Music Time Machine - My current favorites"
)

# Add tracks to the playlist
sp.playlist_add_items(playlist['id'], track_uris)

print(f"\n✓ Success! Created playlist with {len(track_uris)} tracks")
print(f"✓ Playlist URL: {playlist['external_urls']['spotify']}")
print("\nYour top tracks right now:")
for i, name in enumerate(track_names, 1):
    print(f"  {i}. {name}")

python quick_start.py

Authenticated as: John Smith

Fetching your top tracks...

Creating playlist...

✓ Success! Created playlist with 20 tracks
✓ Playlist URL: https://open.spotify.com/playlist/5a8d2f3...

Your top tracks right now:
  1. Karma Police
  2. Everlong
  3. Mr. Brightside
  4. Dreams
  5. Wonderwall
  6. Smells Like Teen Spirit
  7. Don't Look Back In Anger
  8. Champagne Supernova
  9. Under the Bridge
  10. Black Hole Sun
  11. Jeremy
  12. Interstate Love Song
  13. Glycerine
  14. No Rain
  15. Today
  16. Lightning Crashes
  17. What I Got
  18. Semi-Charmed Life
  19. Closing Time
  20. The Impression That I Get

scope = "user-top-read playlist-modify-public playlist-modify-private"

scope = (
    "user-top-read "                   # Read top tracks and artists
    "user-read-recently-played "       # Read listening history
    "playlist-modify-public "          # Create/modify public playlists
    "playlist-modify-private "         # Create/modify private playlists
    "user-library-read"                # Read saved tracks for mood analysis
)

# Example: Working with Spotify identifiers
track = sp.current_user_top_tracks(limit=1)['items'][0]

# Both identifiers point to the same track
track_id = track['id']          # "6rqhFgbbKwnb9MLmUQDhG6"
track_uri = track['uri']        # "spotify:track:6rqhFgbbKwnb9MLmUQDhG6"

# Convert between them
uri_from_id = f"spotify:track:{track_id}"        # Build URI from ID
id_from_uri = track_uri.split(':')[-1]           # Extract ID from URI

# API methods use both formats
audio_features = sp.audio_features([track_id])   # Accepts IDs
sp.playlist_add_items(playlist_id, [track_uri])  # Requires URIs

# Fetch top tracks for each time range
short_term = sp.current_user_top_tracks(limit=50, time_range='short_term')
medium_term = sp.current_user_top_tracks(limit=50, time_range='medium_term')
long_term = sp.current_user_top_tracks(limit=50, time_range='long_term')

# Extract track IDs from each range
short_ids = {track['id'] for track in short_term['items']}
medium_ids = {track['id'] for track in medium_term['items']}
long_ids = {track['id'] for track in long_term['items']}

# Find forgotten gems: in long-term but not in short-term
forgotten = long_ids - short_ids
print(f"You loved {len(forgotten)} tracks long-term but haven't heard them recently")

# Find new discoveries: in short-term but not in medium-term
discoveries = short_ids - medium_ids
print(f"You discovered {len(discoveries)} new tracks in the last month")

# Fetch audio features for a track
track_id = "6rqhFgbbKwnb9MLmUQDhG6"  # "Karma Police" by Radiohead
features = sp.audio_features([track_id])[0]

print(f"Energy: {features['energy']:.2f}")
print(f"Valence: {features['valence']:.2f}")
print(f"Danceability: {features['danceability']:.2f}")
print(f"Tempo: {features['tempo']:.0f} BPM")
print(f"Acousticness: {features['acousticness']:.2f}")
print(f"Instrumentalness: {features['instrumentalness']:.2f}")
print(f"Speechiness: {features['speechiness']:.2f}")

Energy: 0.55
Valence: 0.31
Danceability: 0.49
Tempo: 126 BPM
Acousticness: 0.18
Instrumentalness: 0.00
Speechiness: 0.03

# Example: Score a track against a workout profile
def matches_workout_profile(features):
    """Returns True if track matches workout criteria"""
    return (
        features['energy'] > 0.75 and
        features['valence'] > 0.50 and
        features['tempo'] > 140 and
        features['danceability'] > 0.6
    )

# Fetch your top 50 tracks and their audio features
tracks = sp.current_user_top_tracks(limit=50)['items']
track_ids = [track['id'] for track in tracks]
features_list = sp.audio_features(track_ids)

# Filter for workout tracks
workout_tracks = [
    (tracks[i], features_list[i])
    for i in range(len(tracks))
    if features_list[i] and matches_workout_profile(features_list[i])
]

print(f"Found {len(workout_tracks)} workout-suitable tracks from your top 50")
for track, features in workout_tracks[:5]:
    print(f"  {track['name']} - {track['artists'][0]['name']}")
    print(f"    Energy: {features['energy']:.2f}, Tempo: {features['tempo']:.0f} BPM")

{
  "id": "6rqhFgbbKwnb9MLmUQDhG6",
  "uri": "spotify:track:6rqhFgbbKwnb9MLmUQDhG6",
  "name": "Karma Police",
  "artists": [
    {
      "id": "4Z8W4fKeB5YxbusRsdQVPb",
      "name": "Radiohead"
    }
  ],
  "album": {
    "id": "6400dnyeDyD2mIFHfkwHXN",
    "name": "OK Computer",
    "release_date": "1997-06-16",
    "images": [
      {
        "url": "https://i.scdn.co/image/ab67616d0000b273c8b444df094279e70d0ed856",
        "height": 640,
        "width": 640
      }
    ]
  },
  "duration_ms": 263066,
  "popularity": 77,
  "explicit": false,
  "external_urls": {
    "spotify": "https://open.spotify.com/track/6rqhFgbbKwnb9MLmUQDhG6"
  }
}

CREATE TABLE everything (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    track_id TEXT,
    track_name TEXT,
    artist_name TEXT,
    album_name TEXT,
    duration_ms INTEGER,
    snapshot_date TEXT,
    time_range TEXT,
    energy REAL,
    valence REAL,
    tempo REAL,
    danceability REAL
    -- ... every field for every track for every snapshot
);

CREATE TABLE tracks (
    track_id TEXT PRIMARY KEY,      -- Spotify's track ID (unique identifier)
    name TEXT NOT NULL,              -- Track title
    artist_name TEXT NOT NULL,       -- Primary artist (denormalized for simplicity)
    album_name TEXT NOT NULL,        -- Album title (denormalized)
    duration_ms INTEGER NOT NULL,    -- Track length in milliseconds
    popularity INTEGER,              -- Spotify popularity score (0-100)
    album_image_url TEXT,            -- Album cover URL (for display)
    spotify_url TEXT,                -- Direct link to track in Spotify
    added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP  -- When we first encountered this track
);

import sqlite3

def save_track(conn, track):
    """Save a track to the database (insert or ignore if exists)"""
    conn.execute("""
        INSERT OR IGNORE INTO tracks (
            track_id, name, artist_name, album_name, 
            duration_ms, popularity, album_image_url, spotify_url
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """, (
        track['id'],
        track['name'],
        track['artists'][0]['name'],      # First artist only
        track['album']['name'],
        track['duration_ms'],
        track['popularity'],
        track['album']['images'][0]['url'] if track['album']['images'] else None,
        track['external_urls']['spotify']
    ))

# Usage: fetch tracks from Spotify and save them
top_tracks = sp.current_user_top_tracks(limit=50)['items']

with sqlite3.connect('music_time_machine.db') as conn:
    for track in top_tracks:
        save_track(conn, track)
    conn.commit()

print(f"Saved {len(top_tracks)} tracks to database")

CREATE TABLE audio_features (
    track_id TEXT PRIMARY KEY,       -- References tracks(track_id)
    energy REAL NOT NULL,             -- 0.0 to 1.0
    valence REAL NOT NULL,            -- 0.0 to 1.0 (happiness)
    danceability REAL NOT NULL,       -- 0.0 to 1.0
    tempo REAL NOT NULL,              -- BPM (typically 50-200)
    acousticness REAL NOT NULL,       -- 0.0 to 1.0
    instrumentalness REAL NOT NULL,   -- 0.0 to 1.0
    speechiness REAL NOT NULL,        -- 0.0 to 1.0
    loudness REAL,                    -- dB (typically -60 to 0)
    key INTEGER,                      -- Musical key (0-11, optional)
    mode INTEGER,                     -- Major (1) or Minor (0)
    time_signature INTEGER,           -- Beats per bar (3, 4, 5, 7)
    fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (track_id) REFERENCES tracks(track_id) ON DELETE CASCADE
);

def save_audio_features(conn, track_id, features):
    """Save audio features for a track"""
    conn.execute("""
        INSERT OR REPLACE INTO audio_features (
            track_id, energy, valence, danceability, tempo,
            acousticness, instrumentalness, speechiness,
            loudness, key, mode, time_signature
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (
        track_id,
        features['energy'],
        features['valence'],
        features['danceability'],
        features['tempo'],
        features['acousticness'],
        features['instrumentalness'],
        features['speechiness'],
        features['loudness'],
        features['key'],
        features['mode'],
        features['time_signature']
    ))

# Fetch audio features for all tracks that don't have them yet
with sqlite3.connect('music_time_machine.db') as conn:
    # Find tracks without audio features
    cursor = conn.execute("""
        SELECT track_id FROM tracks 
        WHERE track_id NOT IN (SELECT track_id FROM audio_features)
    """)
    track_ids = [row[0] for row in cursor.fetchall()]
    
    if track_ids:
        print(f"Fetching audio features for {len(track_ids)} tracks...")
        
        # Spotify allows 100 track IDs per request
        for i in range(0, len(track_ids), 100):
            batch = track_ids[i:i+100]
            features_list = sp.audio_features(batch)
            
            for features in features_list:
                if features:  # Some tracks might not have features
                    save_audio_features(conn, features['id'], features)
        
        conn.commit()
        print("Audio features saved")

CREATE TABLE snapshots (
    track_id TEXT NOT NULL,          -- References tracks(track_id)
    snapshot_date DATE NOT NULL,     -- When this snapshot was taken
    time_range TEXT NOT NULL,        -- 'short_term', 'medium_term', or 'long_term'
    rank INTEGER NOT NULL,           -- Position in top tracks (1-50)
    PRIMARY KEY (track_id, snapshot_date, time_range),
    FOREIGN KEY (track_id) REFERENCES tracks(track_id) ON DELETE CASCADE
);

-- Index for time-based queries (critical for performance)
CREATE INDEX idx_snapshots_date ON snapshots(snapshot_date);
CREATE INDEX idx_snapshots_track_date ON snapshots(track_id, snapshot_date);

from datetime import date

def create_snapshot(conn, sp, time_range='short_term'):
    """
    Fetch current top tracks and save as a snapshot
    Returns number of tracks saved
    """
    # Fetch top tracks from Spotify
    top_tracks = sp.current_user_top_tracks(limit=50, time_range=time_range)['items']
    today = date.today().isoformat()
    
    # Save tracks to tracks table first
    for track in top_tracks:
        save_track(conn, track)  # INSERT OR IGNORE
    
    # Save snapshot records
    for rank, track in enumerate(top_tracks, start=1):
        conn.execute("""
            INSERT OR IGNORE INTO snapshots (track_id, snapshot_date, time_range, rank)
            VALUES (?, ?, ?, ?)
        """, (track['id'], today, time_range, rank))
    
    conn.commit()
    return len(top_tracks)

# Usage: create monthly snapshot
with sqlite3.connect('music_time_machine.db') as conn:
    count = create_snapshot(conn, sp, time_range='short_term')
    print(f"Created snapshot with {count} tracks")

-- Tracks that appeared 90-365 days ago but not in last 30 days
SELECT DISTINCT t.track_id, t.name, t.artist_name
FROM tracks t
JOIN snapshots s ON t.track_id = s.track_id
WHERE s.snapshot_date BETWEEN date('now', '-365 days') AND date('now', '-90 days')
  AND t.track_id NOT IN (
      SELECT track_id FROM snapshots 
      WHERE snapshot_date >= date('now', '-30 days')
  )
ORDER BY s.snapshot_date DESC
LIMIT 25;

-- High energy, high tempo tracks from your library
SELECT t.track_id, t.name, t.artist_name, 
       af.energy, af.tempo, af.valence
FROM tracks t
JOIN audio_features af ON t.track_id = af.track_id
WHERE af.energy > 0.75
  AND af.tempo > 140
  AND af.valence > 0.5
  AND af.danceability > 0.6
ORDER BY af.energy DESC, af.tempo DESC
LIMIT 25;

-- Average audio features by month
SELECT 
    strftime('%Y-%m', s.snapshot_date) AS month,
    AVG(af.energy) AS avg_energy,
    AVG(af.valence) AS avg_valence,
    AVG(af.tempo) AS avg_tempo,
    COUNT(DISTINCT t.artist_name) AS unique_artists
FROM snapshots s
JOIN tracks t ON s.track_id = t.track_id
JOIN audio_features af ON t.track_id = af.track_id
WHERE s.time_range = 'short_term'
GROUP BY month
ORDER BY month;

-- Music Time Machine Database Schema
-- SQLite 3 compatible

-- Enable foreign key constraints (disabled by default in SQLite)
PRAGMA foreign_keys = ON;

-- Track metadata
CREATE TABLE IF NOT EXISTS tracks (
    track_id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    artist_name TEXT NOT NULL,
    album_name TEXT NOT NULL,
    duration_ms INTEGER NOT NULL,
    popularity INTEGER,
    album_image_url TEXT,
    spotify_url TEXT,
    added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Audio features for algorithmic curation
CREATE TABLE IF NOT EXISTS audio_features (
    track_id TEXT PRIMARY KEY,
    energy REAL NOT NULL,
    valence REAL NOT NULL,
    danceability REAL NOT NULL,
    tempo REAL NOT NULL,
    acousticness REAL NOT NULL,
    instrumentalness REAL NOT NULL,
    speechiness REAL NOT NULL,
    loudness REAL,
    key INTEGER,
    mode INTEGER,
    time_signature INTEGER,
    fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (track_id) REFERENCES tracks(track_id) ON DELETE CASCADE
);

-- Historical snapshots of top tracks
CREATE TABLE IF NOT EXISTS snapshots (
    track_id TEXT NOT NULL,
    snapshot_date DATE NOT NULL,
    time_range TEXT NOT NULL CHECK(time_range IN ('short_term', 'medium_term', 'long_term')),
    rank INTEGER NOT NULL CHECK(rank >= 1 AND rank <= 50),
    PRIMARY KEY (track_id, snapshot_date, time_range),
    FOREIGN KEY (track_id) REFERENCES tracks(track_id) ON DELETE CASCADE
);

-- Indexes for performance
CREATE INDEX IF NOT EXISTS idx_snapshots_date ON snapshots(snapshot_date);
CREATE INDEX IF NOT EXISTS idx_snapshots_track_date ON snapshots(track_id, snapshot_date);
CREATE INDEX IF NOT EXISTS idx_tracks_artist ON tracks(artist_name);

-- Schema version tracking (for future migrations)
CREATE TABLE IF NOT EXISTS schema_version (
    version INTEGER PRIMARY KEY,
    applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO schema_version (version) VALUES (1);

import sqlite3

def initialize_database(db_path='music_time_machine.db'):
    """Create database and tables from schema"""
    conn = sqlite3.connect(db_path)
    
    # Read and execute schema
    with open('schema.sql', 'r') as f:
        schema_sql = f.read()
        conn.executescript(schema_sql)
    
    conn.commit()
    conn.close()
    print(f"Database initialized at {db_path}")

# Initialize on first run
if __name__ == '__main__':
    initialize_database()

-- Migration 002: Add genre tracking
-- Check current version first
SELECT MAX(version) FROM schema_version;

-- If version is 1, run this migration
ALTER TABLE tracks ADD COLUMN primary_genre TEXT;

-- Record that migration 2 was applied
INSERT INTO schema_version (version) VALUES (2);

import sqlite3
from datetime import date

def find_forgotten_gems(conn, limit=25):
    """
    Find tracks you loved months ago but haven't heard recently
    
    Args:
        conn: SQLite database connection
        limit: Maximum number of tracks to return
    
    Returns:
        List of (track_id, name, artist_name) tuples
    """
    cursor = conn.execute("""
        SELECT DISTINCT t.track_id, t.name, t.artist_name, t.album_name
        FROM tracks t
        JOIN snapshots s ON t.track_id = s.track_id
        WHERE s.snapshot_date BETWEEN date('now', '-365 days') AND date('now', '-90 days')
          AND t.track_id NOT IN (
              SELECT track_id FROM snapshots 
              WHERE snapshot_date >= date('now', '-30 days')
          )
        GROUP BY t.track_id
        ORDER BY MAX(s.snapshot_date) DESC
        LIMIT ?
    """, (limit,))
    
    return cursor.fetchall()

def create_forgotten_gems_playlist(sp, conn):
    """
    Create a playlist of forgotten gems in user's Spotify account
    
    Args:
        sp: Authenticated Spotipy client
        conn: SQLite database connection
    
    Returns:
        Playlist URL if successful, None otherwise
    """
    # Find forgotten tracks
    forgotten = find_forgotten_gems(conn, limit=25)
    
    if not forgotten:
        print("No forgotten gems found. This feature needs at least 3 months of history.")
        print("Keep taking monthly snapshots and check back later!")
        return None
    
    # Get user info
    user = sp.current_user()
    
    # Create playlist
    playlist_name = f"Forgotten Gems - {date.today().strftime('%B %Y')}"
    playlist = sp.user_playlist_create(
        user=user['id'],
        name=playlist_name,
        public=False,
        description="Tracks I loved months ago but haven't heard recently - rediscovered by Music Time Machine"
    )
    
    # Build track URIs
    track_uris = [f"spotify:track:{track[0]}" for track in forgotten]
    
    # Add tracks to playlist
    sp.playlist_add_items(playlist['id'], track_uris)
    
    # Display results
    print(f"\n✓ Created playlist: {playlist_name}")
    print(f"✓ Added {len(forgotten)} forgotten gems")
    print(f"✓ Playlist URL: {playlist['external_urls']['spotify']}\n")
    
    print("Your forgotten gems:")
    for i, (track_id, name, artist, album) in enumerate(forgotten, 1):
        print(f"  {i}. {name} - {artist}")
    
    return playlist['external_urls']['spotify']

# Usage
with sqlite3.connect('music_time_machine.db') as conn:
    url = create_forgotten_gems_playlist(sp, conn)
    if url:
        print(f"\nOpen in Spotify: {url}")

✓ Created playlist: Forgotten Gems - December 2024
✓ Added 25 forgotten gems
✓ Playlist URL: https://open.spotify.com/playlist/3cEYpjA8bZ0Iex...

Your forgotten gems:
  1. Fake Plastic Trees - Radiohead
  2. Dissolve - Absofacto
  3. The Less I Know The Better - Tame Impala
  4. Midnight City - M83
  5. Electric Feel - MGMT
  6. Such Great Heights - The Postal Service
  7. Little Talks - Of Monsters and Men
  ...
  25. Float On - Modest Mouse

Open in Spotify: https://open.spotify.com/playlist/3cEYpjA8bZ0Iex...

from datetime import date

def save_track(conn, track):
    """Save a track to the database (insert or ignore if exists)"""
    conn.execute("""
        INSERT OR REPLACE INTO tracks (
            track_id, name, artist_name, album_name, 
            duration_ms, popularity, album_image_url, spotify_url
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """, (
        track['id'],
        track['name'],
        track['artists'][0]['name'],
        track['album']['name'],
        track['duration_ms'],
        track.get('popularity', 0),
        track['album']['images'][0]['url'] if track['album']['images'] else None,
        track['external_urls']['spotify']
    ))

def create_monthly_snapshot(sp, conn, time_range='short_term'):
    """
    Create a monthly snapshot of current top tracks
    
    Args:
        sp: Authenticated Spotipy client
        conn: SQLite database connection
        time_range: 'short_term' (last 4 weeks), 'medium_term' (6 months), or 'long_term' (several years)
    
    Returns:
        Tuple of (playlist_url, track_count)
    """
    # Check if we already took a snapshot today
    today = date.today().isoformat()
    cursor = conn.execute("""
        SELECT COUNT(*) FROM snapshots 
        WHERE snapshot_date = ? AND time_range = ?
    """, (today, time_range))
    
    if cursor.fetchone()[0] > 0:
        print(f"Already took a {time_range} snapshot today.")
        print("Monthly snapshots are designed to run once per month.")
        return None, 0
    
    # Fetch top tracks from Spotify
    print(f"Fetching your top {time_range} tracks...")
    top_tracks = sp.current_user_top_tracks(limit=50, time_range=time_range)['items']
    
    # Save tracks to database
    for track in top_tracks:
        save_track(conn, track)
    
    # Save snapshot records
    for rank, track in enumerate(top_tracks, start=1):
        conn.execute("""
            INSERT OR IGNORE INTO snapshots (track_id, snapshot_date, time_range, rank)
            VALUES (?, ?, ?, ?)
        """, (track['id'], today, time_range, rank))
    
    conn.commit()
    
    # Create dated playlist in Spotify
    user = sp.current_user()
    month_year = date.today().strftime('%B %Y')
    playlist_name = f"Currently Obsessed - {month_year}"
    
    playlist = sp.user_playlist_create(
        user=user['id'],
        name=playlist_name,
        public=False,
        description=f"My top tracks from {month_year} - captured by Music Time Machine"
    )
    
    # Add tracks to playlist
    track_uris = [f"spotify:track:{track['id']}" for track in top_tracks]
    sp.playlist_add_items(playlist['id'], track_uris)
    
    # Display results
    print(f"\n✓ Snapshot saved: {len(top_tracks)} tracks from {today}")
    print(f"✓ Created playlist: {playlist_name}")
    print(f"✓ Playlist URL: {playlist['external_urls']['spotify']}\n")
    
    print("Your current top tracks:")
    for i, track in enumerate(top_tracks[:10], 1):
        print(f"  {i}. {track['name']} - {track['artists'][0]['name']}")
    if len(top_tracks) > 10:
        print(f"  ... and {len(top_tracks) - 10} more")
    
    return playlist['external_urls']['spotify'], len(top_tracks)

# Usage: Run this once per month
with sqlite3.connect('music_time_machine.db') as conn:
    url, count = create_monthly_snapshot(sp, conn, time_range='short_term')
    if url:
        print(f"\nCome back next month for another snapshot!")
        print(f"After 3+ months, you'll have enough history for Forgotten Gems.")

Fetching your top short_term tracks...

✓ Snapshot saved: 50 tracks from 2024-12-03
✓ Created playlist: Currently Obsessed - December 2024
✓ Playlist URL: https://open.spotify.com/playlist/7dGJo6EpROf...

Your current top tracks:
  1. Creep - Radiohead
  2. Karma Police - Radiohead
  3. No Surprises - Radiohead
  4. High and Dry - Radiohead
  5. Street Spirit (Fade Out) - Radiohead
  6. The Less I Know The Better - Tame Impala
  7. Elephant - Tame Impala
  8. Borderline - Tame Impala
  9. Let It Happen - Tame Impala
  10. New Person, Same Old Mistakes - Tame Impala
  ... and 40 more

Come back next month for another snapshot!
After 3+ months, you'll have enough history for Forgotten Gems.

# Mood profile definitions
MOOD_PROFILES = {
    'workout': {
        'description': 'High energy tracks for running or gym',
        'criteria': {
            'energy_min': 0.75,
            'tempo_min': 140,
            'valence_min': 0.50,
            'danceability_min': 0.60
        }
    },
    'focus': {
        'description': 'Low energy instrumental tracks for concentration',
        'criteria': {
            'energy_max': 0.40,
            'tempo_min': 90,
            'tempo_max': 110,
            'instrumentalness_min': 0.50,
            'speechiness_max': 0.10
        }
    },
    'chill': {
        'description': 'Relaxed, mellow tracks',
        'criteria': {
            'energy_max': 0.50,
            'valence_min': 0.40,
            'valence_max': 0.70,
            'tempo_min': 80,
            'tempo_max': 110,
            'acousticness_min': 0.30
        }
    },
    'party': {
        'description': 'High energy, upbeat dance tracks',
        'criteria': {
            'energy_min': 0.70,
            'valence_min': 0.70,
            'danceability_min': 0.70,
            'tempo_min': 118,
            'tempo_max': 135
        }
    },
    'melancholic': {
        'description': 'Sad, introspective tracks',
        'criteria': {
            'energy_max': 0.50,
            'valence_max': 0.30,
            'acousticness_min': 0.40
        }
    }
}

def display_available_moods():
    """Show available mood profiles"""
    print("Available mood profiles:")
    for mood_name, profile in MOOD_PROFILES.items():
        print(f"  • {mood_name}: {profile['description']}")
    print()

def build_mood_query(criteria):
    """
    Build SQL WHERE clause from mood criteria
    
    Args:
        criteria: Dictionary of min/max constraints for audio features
    
    Returns:
        Tuple of (where_clause, params)
    """
    conditions = []
    params = []
    
    for key, value in criteria.items():
        if key.endswith('_min'):
            feature = key[:-4]  # Remove '_min' suffix
            conditions.append(f"af.{feature} >= ?")
            params.append(value)
        elif key.endswith('_max'):
            feature = key[:-4]  # Remove '_max' suffix
            conditions.append(f"af.{feature} <= ?")
            params.append(value)
    
    where_clause = " AND ".join(conditions)
    return where_clause, params

def find_tracks_by_mood(conn, mood_name, limit=25):
    """
    Find tracks matching a mood profile
    
    Args:
        conn: SQLite database connection
        mood_name: Name of mood profile from MOOD_PROFILES
        limit: Maximum number of tracks to return
    
    Returns:
        List of (track_id, name, artist_name, album_name) tuples
    """
    if mood_name not in MOOD_PROFILES:
        print(f"Unknown mood: {mood_name}")
        display_available_moods()
        return []
    
    profile = MOOD_PROFILES[mood_name]
    where_clause, params = build_mood_query(profile['criteria'])
    
    query = f"""
        SELECT t.track_id, t.name, t.artist_name, t.album_name,
               af.energy, af.valence, af.tempo
        FROM tracks t
        JOIN audio_features af ON t.track_id = af.track_id
        WHERE {where_clause}
        ORDER BY af.energy DESC, af.tempo DESC
        LIMIT ?
    """
    
    params.append(limit)
    cursor = conn.execute(query, params)
    return cursor.fetchall()

import time

def fetch_missing_audio_features(sp, conn):
    """
    Fetch audio features for tracks that don't have them yet.
    Uses batching and rate limiting to stay under Spotify's limits.
    
    Args:
        sp: Authenticated Spotipy client
        conn: SQLite database connection
    
    Returns:
        Number of tracks for which features were fetched
    """
    # Find tracks without audio features
    cursor = conn.execute("""
        SELECT track_id FROM tracks 
        WHERE track_id NOT IN (SELECT track_id FROM audio_features)
    """)
    track_ids = [row[0] for row in cursor.fetchall()]
    
    if not track_ids:
        return 0
    
    print(f"Fetching audio features for {len(track_ids)} tracks...")
    
    # Spotify allows 100 track IDs per request
    fetched_count = 0
    batch_size = 100
    
    for i in range(0, len(track_ids), batch_size):
        batch = track_ids[i:i+batch_size]
        
        try:
            features_list = sp.audio_features(batch)
            
            for features in features_list:
                if features:  # Some tracks might not have features
                    conn.execute("""
                        INSERT OR REPLACE INTO audio_features (
                            track_id, energy, valence, danceability, tempo,
                            acousticness, instrumentalness, speechiness,
                            loudness, key, mode, time_signature
                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """, (
                        features['id'],
                        features['energy'],
                        features['valence'],
                        features['danceability'],
                        features['tempo'],
                        features['acousticness'],
                        features['instrumentalness'],
                        features['speechiness'],
                        features['loudness'],
                        features['key'],
                        features['mode'],
                        features['time_signature']
                    ))
                    fetched_count += 1
            
            # Proactive rate limiting: sleep between batches
            # This keeps us comfortably under Spotify's rate limits
            if i + batch_size < len(track_ids):
                time.sleep(0.5)
                
        except Exception as e:
            print(f"Error fetching batch {i//batch_size + 1}: {e}")
            # Continue with next batch even if this one fails
            continue
    
    conn.commit()
    print(f"✓ Saved audio features for {fetched_count} tracks\n")
    return fetched_count

def create_mood_playlist(sp, conn, mood_name, limit=25):
    """
    Create a playlist based on mood profile
    
    Args:
        sp: Authenticated Spotipy client
        conn: SQLite database connection
        mood_name: Name of mood profile from MOOD_PROFILES
        limit: Number of tracks to include
    
    Returns:
        Playlist URL if successful, None otherwise
    """
    # Ensure we have audio features
    fetch_missing_audio_features(sp, conn)
    
    # Find matching tracks
    tracks = find_tracks_by_mood(conn, mood_name, limit)
    
    if not tracks:
        print(f"No tracks found matching '{mood_name}' mood profile.")
        print("This usually means:")
        print("  1. You don't have enough tracks in your database yet")
        print("  2. Your music taste doesn't match this mood profile")
        print("Try taking more monthly snapshots or try a different mood.")
        return None
    
    # Get user info
    user = sp.current_user()
    
    # Create playlist
    profile = MOOD_PROFILES[mood_name]
    playlist_name = f"{mood_name.title()} - {date.today().strftime('%B %Y')}"
    playlist = sp.user_playlist_create(
        user=user['id'],
        name=playlist_name,
        public=False,
        description=f"{profile['description']} - curated by Music Time Machine"
    )
    
    # Add tracks to playlist
    track_uris = [f"spotify:track:{track[0]}" for track in tracks]
    sp.playlist_add_items(playlist['id'], track_uris)
    
    # Display results
    print(f"\n✓ Created playlist: {playlist_name}")
    print(f"✓ Added {len(tracks)} tracks matching '{mood_name}' profile")
    print(f"✓ Playlist URL: {playlist['external_urls']['spotify']}\n")
    
    print(f"Tracks in your {mood_name} playlist:")
    for i, track in enumerate(tracks[:10], 1):
        track_id, name, artist, album, energy, valence, tempo = track
        print(f"  {i}. {name} - {artist}")
        print(f"     Energy: {energy:.2f}, Valence: {valence:.2f}, Tempo: {tempo:.0f} BPM")
    
    if len(tracks) > 10:
        print(f"  ... and {len(tracks) - 10} more")
    
    return playlist['external_urls']['spotify']

# Usage: Generate a workout playlist
with sqlite3.connect('music_time_machine.db') as conn:
    display_available_moods()
    url = create_mood_playlist(sp, conn, 'workout', limit=25)
    if url:
        print(f"\nOpen in Spotify: {url}")

Available mood profiles:
  • workout: High energy tracks for running or gym
  • focus: Low energy instrumental tracks for concentration
  • chill: Relaxed, mellow tracks
  • party: High energy, upbeat dance tracks
  • melancholic: Sad, introspective tracks

Fetching audio features for 327 tracks...
✓ Saved audio features for 327 tracks

✓ Created playlist: Workout - December 2024
✓ Added 25 tracks matching 'workout' profile
✓ Playlist URL: https://open.spotify.com/playlist/4kL9mN3pR...

Tracks in your workout playlist:
  1. Losing My Religion - R.E.M.
     Energy: 0.84, Valence: 0.52, Tempo: 163 BPM
  2. The Less I Know The Better - Tame Impala
     Energy: 0.83, Valence: 0.61, Tempo: 117 BPM
  3. Feels Like We Only Go Backwards - Tame Impala
     Energy: 0.79, Valence: 0.54, Tempo: 156 BPM
  ... and 22 more

Open in Spotify: https://open.spotify.com/playlist/4kL9mN3pR...

def analyze_musical_evolution(conn):
    """
    Generate analytics about musical taste evolution over time
    
    Args:
        conn: SQLite database connection
    
    Returns:
        Dictionary containing various analytics metrics
    """
    analytics = {}
    
    # Total unique tracks and artists
    cursor = conn.execute("SELECT COUNT(DISTINCT track_id) FROM tracks")
    analytics['total_tracks'] = cursor.fetchone()[0]
    
    cursor = conn.execute("SELECT COUNT(DISTINCT artist_name) FROM tracks")
    analytics['total_artists'] = cursor.fetchone()[0]
    
    # Number of snapshots taken
    cursor = conn.execute("""
        SELECT COUNT(DISTINCT snapshot_date) FROM snapshots 
        WHERE time_range = 'short_term'
    """)
    analytics['snapshot_count'] = cursor.fetchone()[0]
    
    # Date range of data
    cursor = conn.execute("""
        SELECT MIN(snapshot_date), MAX(snapshot_date) 
        FROM snapshots WHERE time_range = 'short_term'
    """)
    min_date, max_date = cursor.fetchone()
    analytics['date_range'] = (min_date, max_date)
    
    # Track turnover rate: how many tracks appear in one snapshot but not the next
    cursor = conn.execute("""
        WITH consecutive_snapshots AS (
            SELECT snapshot_date,
                   LAG(snapshot_date) OVER (ORDER BY snapshot_date) as prev_date
            FROM (SELECT DISTINCT snapshot_date FROM snapshots 
                  WHERE time_range = 'short_term')
        ),
        turnover_by_snapshot AS (
            SELECT cs.snapshot_date,
                   COUNT(DISTINCT s1.track_id) as current_tracks,
                   COUNT(DISTINCT s2.track_id) as previous_tracks,
                   COUNT(DISTINCT CASE WHEN s2.track_id IS NULL THEN s1.track_id END) as new_tracks
            FROM consecutive_snapshots cs
            JOIN snapshots s1 ON s1.snapshot_date = cs.snapshot_date 
                              AND s1.time_range = 'short_term'
            LEFT JOIN snapshots s2 ON s2.snapshot_date = cs.prev_date 
                                   AND s2.track_id = s1.track_id
                                   AND s2.time_range = 'short_term'
            WHERE cs.prev_date IS NOT NULL
            GROUP BY cs.snapshot_date
        )
        SELECT AVG(CAST(new_tracks AS FLOAT) / current_tracks * 100)
        FROM turnover_by_snapshot
    """)
    result = cursor.fetchone()[0]
    analytics['avg_turnover_pct'] = result if result else 0
    
    # Average audio features by month
    cursor = conn.execute("""
        SELECT 
            strftime('%Y-%m', s.snapshot_date) AS month,
            AVG(af.energy) AS avg_energy,
            AVG(af.valence) AS avg_valence,
            AVG(af.tempo) AS avg_tempo,
            COUNT(DISTINCT t.artist_name) AS unique_artists
        FROM snapshots s
        JOIN tracks t ON s.track_id = t.track_id
        LEFT JOIN audio_features af ON t.track_id = af.track_id
        WHERE s.time_range = 'short_term'
        GROUP BY month
        ORDER BY month
    """)
    analytics['monthly_trends'] = cursor.fetchall()
    
    # Top artists across all time
    cursor = conn.execute("""
        SELECT t.artist_name, COUNT(DISTINCT s.snapshot_date) as appearances
        FROM tracks t
        JOIN snapshots s ON t.track_id = s.track_id
        WHERE s.time_range = 'short_term'
        GROUP BY t.artist_name
        ORDER BY appearances DESC
        LIMIT 10
    """)
    analytics['top_artists'] = cursor.fetchall()
    
    return analytics

def display_analytics(analytics):
    """Pretty-print analytics results"""
    print("\n" + "="*60)
    print("MUSICAL EVOLUTION ANALYTICS")
    print("="*60 + "\n")
    
    # Overview
    print("📊 Overview")
    print(f"  Total unique tracks discovered: {analytics['total_tracks']}")
    print(f"  Total unique artists: {analytics['total_artists']}")
    print(f"  Snapshots taken: {analytics['snapshot_count']}")
    
    if analytics['date_range'][0]:
        print(f"  Data spans: {analytics['date_range'][0]} to {analytics['date_range'][1]}")
    print()
    
    # Turnover rate
    if analytics['avg_turnover_pct'] > 0:
        print("🔄 Track Turnover Rate")
        print(f"  Average: {analytics['avg_turnover_pct']:.1f}% of tracks change between snapshots")
        if analytics['avg_turnover_pct'] < 20:
            print("  → You stick with favorites for a long time")
        elif analytics['avg_turnover_pct'] > 40:
            print("  → You discover and rotate through music quickly")
        else:
            print("  → You balance discovering new music with replaying favorites")
        print()
    
    # Monthly trends
    if analytics['monthly_trends']:
        print("📈 Monthly Trends")
        print("  Month      Energy  Valence  Tempo   Artists")
        print("  " + "-"*52)
        for row in analytics['monthly_trends']:
            month, energy, valence, tempo, artists = row
            energy_str = f"{energy:.2f}" if energy else "N/A"
            valence_str = f"{valence:.2f}" if valence else "N/A"
            tempo_str = f"{tempo:.0f}" if tempo else "N/A"
            print(f"  {month}    {energy_str:>6}  {valence_str:>6}  {tempo_str:>6}  {artists:>4}")
        print()
    
    # Top artists
    if analytics['top_artists']:
        print("🎤 Most Consistent Artists")
        print("  (Artists who appeared in the most monthly snapshots)")
        for i, (artist, appearances) in enumerate(analytics['top_artists'], 1):
            print(f"  {i:2}. {artist:<40} {appearances} months")
        print()
    
    print("="*60 + "\n")

# Usage
with sqlite3.connect('music_time_machine.db') as conn:
    analytics = analyze_musical_evolution(conn)
    display_analytics(analytics)

============================================================
MUSICAL EVOLUTION ANALYTICS
============================================================

📊 Overview
  Total unique tracks discovered: 487
  Total unique artists: 142
  Snapshots taken: 8
  Data spans: 2024-05-01 to 2024-12-03

🔄 Track Turnover Rate
  Average: 32.4% of tracks change between snapshots
  → You balance discovering new music with replaying favorites

📈 Monthly Trends
  Month      Energy  Valence  Tempo   Artists
  ----------------------------------------------------
  2024-05    0.61    0.48     124      38
  2024-06    0.65    0.52     128      41
  2024-07    0.68    0.58     132      39
  2024-08    0.64    0.55     129      42
  2024-09    0.59    0.46     121      44
  2024-10    0.62    0.49     125      40
  2024-11    0.66    0.53     130      43
  2024-12    0.63    0.50     126      45

🎤 Most Consistent Artists
  (Artists who appeared in the most monthly snapshots)
   1. Radiohead                                      8 months
   2. Tame Impala                                    7 months
   3. Arctic Monkeys                                 6 months
   4. The Strokes                                    6 months
   5. MGMT                                           5 months
   6. Foster The People                              5 months
   7. Phoenix                                        5 months
   8. Two Door Cinema Club                           4 months
   9. Vampire Weekend                                4 months
  10. alt-J                                          4 months

============================================================

"""
Music Time Machine - Command Line Interface
Your personal music history tracker and playlist generator
"""
import os
import sqlite3
from dotenv import load_dotenv
import spotipy
from spotipy.oauth2 import SpotifyOAuth

# Load environment variables
load_dotenv()

# Initialize Spotify client
scope = "user-top-read playlist-modify-public playlist-modify-private"
sp = spotipy.Spotify(auth_manager=SpotifyOAuth(scope=scope))

# Database connection
DB_PATH = 'music_time_machine.db'

def display_menu():
    """Display main menu"""
    print("\n" + "="*60)
    print("MUSIC TIME MACHINE")
    print("="*60)
    print("\n1. Create Monthly Snapshot (Currently Obsessed)")
    print("2. Find Forgotten Gems")
    print("3. Generate Mood Playlist")
    print("4. View Analytics")
    print("5. Exit")
    print()

def main():
    """Main application loop"""
    # Ensure database exists
    if not os.path.exists(DB_PATH):
        print("Initializing database...")
        # Run schema.sql here (or call initialize_database())
    
    while True:
        display_menu()
        choice = input("Select an option (1-5): ").strip()
        
        with sqlite3.connect(DB_PATH) as conn:
            if choice == '1':
                create_monthly_snapshot(sp, conn)
            
            elif choice == '2':
                create_forgotten_gems_playlist(sp, conn)
            
            elif choice == '3':
                display_available_moods()
                mood = input("Enter mood name: ").strip().lower()
                create_mood_playlist(sp, conn, mood, limit=25)
            
            elif choice == '4':
                analytics = analyze_musical_evolution(conn)
                display_analytics(analytics)
            
            elif choice == '5':
                print("\nThanks for using Music Time Machine!")
                print("Keep taking monthly snapshots to build your musical history.\n")
                break
            
            else:
                print("Invalid choice. Please select 1-5.")
        
        if choice != '5':
            input("\nPress Enter to continue...")

if __name__ == '__main__':
    main()

import time
import random
from spotipy.exceptions import SpotifyException
import requests

class MusicTimeMachineError(Exception):
    """Base exception for Music Time Machine errors"""
    pass

class TransientError(MusicTimeMachineError):
    """Temporary error that might resolve with retry"""
    pass

class AuthorizationError(MusicTimeMachineError):
    """OAuth or permission error requiring user action"""
    pass

class RateLimitError(MusicTimeMachineError):
    """Hit Spotify's rate limit"""
    def __init__(self, message, retry_after=None):
        super().__init__(message)
        self.retry_after = retry_after

def retry_with_backoff(func, max_attempts=3, base_delay=1.0):
    """
    Retry a function with exponential backoff and jitter
    
    Args:
        func: Function to retry (should take no arguments, use lambda if needed)
        max_attempts: Maximum number of retry attempts
        base_delay: Initial delay in seconds (doubles each retry)
    
    Returns:
        Function result if successful
    
    Raises:
        Last exception if all retries fail
    """
    last_exception = None
    
    for attempt in range(max_attempts):
        try:
            return func()
        
        except (requests.Timeout, requests.ConnectionError) as e:
            last_exception = TransientError(f"Network error: {e}")
            error_type = "network timeout"
        
        except SpotifyException as e:
            if e.http_status == 429:
                # Rate limit - check for Retry-After header
                retry_after = e.headers.get('Retry-After')
                if retry_after:
                    wait_time = int(retry_after)
                    print(f"Rate limited. Waiting {wait_time} seconds as requested...")
                    time.sleep(wait_time)
                    continue
                else:
                    last_exception = RateLimitError("Hit rate limit without Retry-After header")
                    error_type = "rate limit"
            
            elif e.http_status in (500, 502, 503, 504):
                # Server errors are transient
                last_exception = TransientError(f"Spotify server error: {e.http_status}")
                error_type = "server error"
            
            elif e.http_status in (401, 403):
                # Authorization errors don't benefit from retry
                raise AuthorizationError(f"Authorization failed: {e.msg}")
            
            else:
                # Other errors are not transient
                raise
        
        except Exception as e:
            # Unknown error - don't retry
            raise
        
        # Calculate backoff with jitter
        if attempt < max_attempts - 1:
            delay = base_delay * (2 ** attempt)
            jitter = random.uniform(0, delay * 0.1)  # 0-10% jitter
            wait_time = delay + jitter
            
            print(f"Attempt {attempt + 1} failed ({error_type}). Retrying in {wait_time:.1f}s...")
            time.sleep(wait_time)
    
    # All retries exhausted
    raise last_exception

# Fragile: crashes on network hiccup or server error
def create_monthly_snapshot_fragile(sp, conn):
    top_tracks = sp.current_user_top_tracks(limit=50, time_range='short_term')['items']
    
    for track in top_tracks:
        save_track(conn, track)
    
    # If this fails, you lose all the work above
    playlist = sp.user_playlist_create(user['id'], name=playlist_name, public=False)
    sp.playlist_add_items(playlist['id'], track_uris)

def create_monthly_snapshot_robust(sp, conn):
    """Create monthly snapshot with error handling"""
    try:
        # Fetch tracks with retry
        top_tracks = retry_with_backoff(
            lambda: sp.current_user_top_tracks(limit=50, time_range='short_term')['items']
        )
        
        # Save to database
        for track in top_tracks:
            save_track(conn, track)
        conn.commit()
        
        # Create playlist with retry
        user = retry_with_backoff(lambda: sp.current_user())
        
        playlist = retry_with_backoff(
            lambda: sp.user_playlist_create(
                user=user['id'],
                name=playlist_name,
                public=False,
                description=f"My top tracks from {month_year}"
            )
        )
        
        # Add tracks with retry
        track_uris = [f"spotify:track:{track['id']}" for track in top_tracks]
        retry_with_backoff(
            lambda: sp.playlist_add_items(playlist['id'], track_uris)
        )
        
        return playlist['external_urls']['spotify'], len(top_tracks)
    
    except AuthorizationError as e:
        print(f"\n❌ Authorization Error: {e}")
        print("Please re-run the application to re-authorize with Spotify.")
        print("Delete the .cache file if you need to reset OAuth completely.\n")
        return None, 0
    
    except TransientError as e:
        print(f"\n❌ Temporary Error: {e}")
        print("Spotify's servers might be experiencing issues.")
        print("Please try again in a few minutes.\n")
        return None, 0
    
    except RateLimitError as e:
        print(f"\n❌ Rate Limit Error: {e}")
        print("You've made too many requests to Spotify's API.")
        print("Wait a few minutes before trying again.\n")
        return None, 0
    
    except Exception as e:
        print(f"\n❌ Unexpected Error: {e}")
        print("Something went wrong. Please check your internet connection")
        print("and make sure Spotify's API is accessible.\n")
        return None, 0

import sqlite3

def safe_database_operation(conn, operation_func, max_attempts=3):
    """
    Execute a database operation with retry on lock errors
    
    Args:
        conn: SQLite connection
        operation_func: Function that performs database operations
        max_attempts: Maximum retry attempts for lock errors
    
    Returns:
        Result of operation_func
    
    Raises:
        sqlite3.Error if operation fails after retries
    """
    for attempt in range(max_attempts):
        try:
            result = operation_func(conn)
            conn.commit()
            return result
        
        except sqlite3.OperationalError as e:
            if "database is locked" in str(e).lower():
                if attempt < max_attempts - 1:
                    wait_time = 0.5 * (2 ** attempt)  # 0.5s, 1s, 2s
                    print(f"Database locked. Retrying in {wait_time}s...")
                    time.sleep(wait_time)
                    continue
                else:
                    raise MusicTimeMachineError(
                        "Database is locked by another process. "
                        "Close other applications using the database and try again."
                    )
            else:
                # Other operational errors (disk full, corrupted database)
                raise MusicTimeMachineError(f"Database error: {e}")
        
        except sqlite3.IntegrityError as e:
            # Constraint violations (foreign key, unique, etc.)
            raise MusicTimeMachineError(f"Data integrity error: {e}")
        
        except sqlite3.DatabaseError as e:
            # Malformed database, schema errors
            raise MusicTimeMachineError(
                f"Database structure error: {e}. "
                "You might need to rebuild the database from schema.sql"
            )

def save_track_safe(conn, track):
    """Save track with error handling"""
    def operation(conn):
        conn.execute("""
            INSERT OR REPLACE INTO tracks (
                track_id, name, artist_name, album_name, 
                duration_ms, popularity, album_image_url, spotify_url
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            track['id'],
            track['name'],
            track['artists'][0]['name'],
            track['album']['name'],
            track['duration_ms'],
            track.get('popularity', 0),
            track['album']['images'][0]['url'] if track['album']['images'] else None,
            track['external_urls']['spotify']
        ))
    
    safe_database_operation(conn, operation)

def fetch_missing_audio_features_robust(sp, conn):
    """
    Fetch audio features with graceful degradation
    Continue even if some tracks fail
    
    Returns:
        Tuple of (successful_count, failed_count)
    """
    # Find tracks without features
    cursor = conn.execute("""
        SELECT track_id FROM tracks 
        WHERE track_id NOT IN (SELECT track_id FROM audio_features)
    """)
    track_ids = [row[0] for row in cursor.fetchall()]
    
    if not track_ids:
        return 0, 0
    
    print(f"Fetching audio features for {len(track_ids)} tracks...")
    
    successful = 0
    failed = 0
    failed_tracks = []
    
    # Process in batches of 100 (Spotify's limit)
    for i in range(0, len(track_ids), 100):
        batch = track_ids[i:i+100]
        
        try:
            # Fetch features with retry
            features_list = retry_with_backoff(
                lambda: sp.audio_features(batch),
                max_attempts=3
            )
            
            # Save successful features
            for features in features_list:
                if features is None:
                    # Spotify doesn't have features for this track
                    failed += 1
                    continue
                
                try:
                    conn.execute("""
                        INSERT OR REPLACE INTO audio_features (
                            track_id, energy, valence, danceability, tempo,
                            acousticness, instrumentalness, speechiness,
                            loudness, key, mode, time_signature
                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """, (
                        features['id'],
                        features['energy'],
                        features['valence'],
                        features['danceability'],
                        features['tempo'],
                        features['acousticness'],
                        features['instrumentalness'],
                        features['speechiness'],
                        features['loudness'],
                        features['key'],
                        features['mode'],
                        features['time_signature']
                    ))
                    successful += 1
                
                except (sqlite3.Error, KeyError) as e:
                    # Database error or malformed feature data
                    failed += 1
                    failed_tracks.append(features.get('id', 'unknown'))
            
            conn.commit()
        
        except (TransientError, RateLimitError) as e:
            # Entire batch failed - count all tracks as failed
            print(f"Batch failed: {e}")
            failed += len(batch)
            failed_tracks.extend(batch)
        
        except Exception as e:
            # Unexpected error - log and continue with next batch
            print(f"Unexpected error in batch: {e}")
            failed += len(batch)
            failed_tracks.extend(batch)
    
    # Report results
    print(f"✓ Saved audio features for {successful} tracks")
    if failed > 0:
        print(f"⚠ Could not fetch features for {failed} tracks")
        print(f"  This is normal - some tracks don't have audio analysis available.")
    
    return successful, failed

import logging
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('music_time_machine.log'),
        logging.StreamHandler()  # Also print to console
    ]
)

logger = logging.getLogger('MusicTimeMachine')

def create_monthly_snapshot_with_logging(sp, conn):
    """Create monthly snapshot with comprehensive logging"""
    logger.info("Starting monthly snapshot creation")
    
    try:
        # Fetch tracks
        logger.info("Fetching top tracks from Spotify")
        top_tracks = retry_with_backoff(
            lambda: sp.current_user_top_tracks(limit=50, time_range='short_term')['items']
        )
        logger.info(f"Fetched {len(top_tracks)} tracks successfully")
        
        # Save to database
        logger.info("Saving tracks to database")
        for i, track in enumerate(top_tracks):
            save_track_safe(conn, track)
            if (i + 1) % 10 == 0:
                logger.debug(f"Saved {i + 1}/{len(top_tracks)} tracks")
        
        conn.commit()
        logger.info("Database commit successful")
        
        # Create playlist
        logger.info("Creating Spotify playlist")
        user = retry_with_backoff(lambda: sp.current_user())
        logger.debug(f"Authenticated as user: {user['id']}")
        
        month_year = datetime.now().strftime('%B %Y')
        playlist_name = f"Currently Obsessed - {month_year}"
        
        playlist = retry_with_backoff(
            lambda: sp.user_playlist_create(
                user=user['id'],
                name=playlist_name,
                public=False
            )
        )
        logger.info(f"Created playlist: {playlist['id']}")
        
        # Add tracks
        track_uris = [f"spotify:track:{track['id']}" for track in top_tracks]
        retry_with_backoff(
            lambda: sp.playlist_add_items(playlist['id'], track_uris)
        )
        logger.info(f"Added {len(track_uris)} tracks to playlist")
        
        logger.info("Monthly snapshot completed successfully")
        return playlist['external_urls']['spotify'], len(top_tracks)
    
    except AuthorizationError as e:
        logger.error(f"Authorization failed: {e}")
        return None, 0
    
    except TransientError as e:
        logger.warning(f"Transient error (after retries): {e}")
        return None, 0
    
    except Exception as e:
        logger.exception(f"Unexpected error during snapshot: {e}")
        return None, 0

2024-12-03 14:32:15 - MusicTimeMachine - INFO - Starting monthly snapshot creation
2024-12-03 14:32:15 - MusicTimeMachine - INFO - Fetching top tracks from Spotify
2024-12-03 14:32:17 - MusicTimeMachine - INFO - Fetched 50 tracks successfully
2024-12-03 14:32:17 - MusicTimeMachine - INFO - Saving tracks to database
2024-12-03 14:32:17 - MusicTimeMachine - DEBUG - Saved 10/50 tracks
2024-12-03 14:32:18 - MusicTimeMachine - DEBUG - Saved 20/50 tracks
2024-12-03 14:32:18 - MusicTimeMachine - DEBUG - Saved 30/50 tracks
2024-12-03 14:32:19 - MusicTimeMachine - DEBUG - Saved 40/50 tracks
2024-12-03 14:32:19 - MusicTimeMachine - DEBUG - Saved 50/50 tracks
2024-12-03 14:32:19 - MusicTimeMachine - INFO - Database commit successful
2024-12-03 14:32:19 - MusicTimeMachine - INFO - Creating Spotify playlist
2024-12-03 14:32:19 - MusicTimeMachine - DEBUG - Authenticated as user: john_smith
2024-12-03 14:32:20 - MusicTimeMachine - INFO - Created playlist: 3cEYpjA8bZ0Iex
2024-12-03 14:32:21 - MusicTimeMachine - INFO - Added 50 tracks to playlist
2024-12-03 14:32:21 - MusicTimeMachine - INFO - Monthly snapshot completed successfully

def production_ready_feature(sp, conn, feature_name):
    """
    Template for production-ready feature implementation
    
    Demonstrates: retry logic, error categorization, logging,
    graceful degradation, user-friendly messages
    """
    logger.info(f"Starting {feature_name}")
    
    try:
        # Step 1: Fetch data with retry
        logger.info("Fetching data from Spotify")
        data = retry_with_backoff(
            lambda: sp.some_spotify_method(),
            max_attempts=3
        )
        logger.info(f"Fetched {len(data)} items")
        
        # Step 2: Process data with graceful degradation
        logger.info("Processing data")
        successful = 0
        failed = 0
        
        for item in data:
            try:
                # Process individual item
                process_item(conn, item)
                successful += 1
            except Exception as e:
                # Log failure but continue processing
                logger.warning(f"Failed to process item: {e}")
                failed += 1
        
        conn.commit()
        logger.info(f"Processed {successful} items successfully, {failed} failed")
        
        # Step 3: Create output with retry
        logger.info("Creating output")
        result = retry_with_backoff(
            lambda: create_output(sp, processed_data),
            max_attempts=3
        )
        logger.info(f"{feature_name} completed successfully")
        
        # Step 4: User-friendly success message
        print(f"\n✓ {feature_name} completed successfully")
        print(f"✓ Processed {successful} items")
        if failed > 0:
            print(f"⚠ {failed} items could not be processed (see log for details)")
        
        return result
    
    except AuthorizationError as e:
        logger.error(f"Authorization failed: {e}")
        print(f"\n❌ Authorization Error")
        print("Your Spotify authentication has expired or is invalid.")
        print("Please re-run the application to re-authorize.")
        return None
    
    except TransientError as e:
        logger.warning(f"Transient error: {e}")
        print(f"\n❌ Temporary Error")
        print("Spotify's servers are experiencing issues.")
        print("Please try again in a few minutes.")
        return None
    
    except RateLimitError as e:
        logger.warning(f"Rate limit hit: {e}")
        print(f"\n❌ Rate Limit Exceeded")
        print("You've made too many requests to Spotify.")
        print("Please wait a few minutes before trying again.")
        return None
    
    except MusicTimeMachineError as e:
        logger.error(f"Application error: {e}")
        print(f"\n❌ Error: {e}")
        return None
    
    except Exception as e:
        logger.exception(f"Unexpected error: {e}")
        print(f"\n❌ Unexpected Error")
        print(f"Something went wrong: {e}")
        print("Please check music_time_machine.log for details.")
        return None

"""
diagnose.py - System diagnostic tool for Music Time Machine

Run this script when something breaks. It tests each system component 
independently and reports exactly what's working and what's not.

Usage: python diagnose.py
"""

import sqlite3
import os
import sys
from pathlib import Path
import spotipy
from spotipy.oauth2 import SpotifyOAuth

def check_database():
    """Test database connectivity and schema."""
    print("\n" + "="*60)
    print("DATABASE DIAGNOSTICS")
    print("="*60)
    
    db_path = Path("music_time_machine.db")
    
    # Check if database file exists
    if not db_path.exists():
        print("❌ Database file not found")
        print(f"   Expected: {db_path.absolute()}")
        print("   Action: Run the database setup script first")
        return False
    
    print(f"✓ Database file exists: {db_path.absolute()}")
    
    # Test connection
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        print("✓ Database connection successful")
    except Exception as e:
        print(f"❌ Cannot connect to database: {e}")
        return False
    
    # Verify schema
    required_tables = ['listening_history', 'monthly_snapshots']
    
    cursor.execute("""
        SELECT name FROM sqlite_master 
        WHERE type='table'
    """)
    existing_tables = [row[0] for row in cursor.fetchall()]
    
    schema_valid = True
    for table in required_tables:
        if table in existing_tables:
            print(f"✓ Table exists: {table}")
            
            # Check row count
            cursor.execute(f"SELECT COUNT(*) FROM {table}")
            count = cursor.fetchone()[0]
            print(f"  → Contains {count} rows")
        else:
            print(f"❌ Missing table: {table}")
            schema_valid = False
    
    # Check for audio features columns
    cursor.execute("PRAGMA table_info(listening_history)")
    columns = [row[1] for row in cursor.fetchall()]
    
    required_columns = ['energy', 'valence', 'danceability', 'tempo']
    for col in required_columns:
        if col in columns:
            print(f"✓ Column exists: {col}")
        else:
            print(f"❌ Missing column: {col}")
            schema_valid = False
    
    conn.close()
    
    if schema_valid:
        print("\n✓ Database schema is valid")
    else:
        print("\n❌ Database schema has issues")
        print("   Action: Run database migration script")
    
    return schema_valid


def check_oauth():
    """Test OAuth configuration and token validity."""
    print("\n" + "="*60)
    print("OAUTH DIAGNOSTICS")
    print("="*60)
    
    # Check for credentials
    client_id = os.getenv('SPOTIPY_CLIENT_ID')
    client_secret = os.getenv('SPOTIPY_CLIENT_SECRET')
    redirect_uri = os.getenv('SPOTIPY_REDIRECT_URI')
    
    creds_valid = True
    
    if client_id:
        print(f"✓ Client ID found: {client_id[:8]}...")
    else:
        print("❌ SPOTIPY_CLIENT_ID not set")
        creds_valid = False
    
    if client_secret:
        print(f"✓ Client Secret found: {client_secret[:8]}...")
    else:
        print("❌ SPOTIPY_CLIENT_SECRET not set")
        creds_valid = False
    
    if redirect_uri:
        print(f"✓ Redirect URI: {redirect_uri}")
    else:
        print("❌ SPOTIPY_REDIRECT_URI not set")
        creds_valid = False
    
    if not creds_valid:
        print("\n❌ OAuth credentials incomplete")
        print("   Action: Set environment variables or create .env file")
        return False
    
    # Check for token cache
    cache_path = Path(".cache")
    if cache_path.exists():
        print(f"✓ Token cache exists: {cache_path}")
    else:
        print("⚠ No token cache found (will require login)")
    
    # Test token validity
    try:
        scope = "user-top-read user-read-recently-played"
        auth_manager = SpotifyOAuth(
            client_id=client_id,
            client_secret=client_secret,
            redirect_uri=redirect_uri,
            scope=scope
        )
        
        token_info = auth_manager.get_cached_token()
        
        if token_info:
            print("✓ Valid token found in cache")
            
            # Check if token is expired
            import time
            expires_at = token_info.get('expires_at', 0)
            now = int(time.time())
            
            if expires_at > now:
                remaining = expires_at - now
                print(f"  → Token valid for {remaining // 60} minutes")
            else:
                print("⚠ Token expired (will auto-refresh)")
        else:
            print("⚠ No cached token (will require login)")
            
    except Exception as e:
        print(f"❌ OAuth error: {e}")
        return False
    
    print("\n✓ OAuth configuration is valid")
    return True


def check_api_access():
    """Test actual API connectivity."""
    print("\n" + "="*60)
    print("API DIAGNOSTICS")
    print("="*60)
    
    try:
        scope = "user-top-read user-read-recently-played"
        auth_manager = SpotifyOAuth(scope=scope)
        sp = spotipy.Spotify(auth_manager=auth_manager)
        
        # Test with simple API call
        print("Testing API connection...")
        user = sp.current_user()
        
        print(f"✓ Successfully connected to Spotify API")
        print(f"  → Authenticated as: {user['display_name']}")
        print(f"  → User ID: {user['id']}")
        
        # Test top tracks endpoint
        print("\nTesting top tracks endpoint...")
        top_tracks = sp.current_user_top_tracks(limit=5, time_range='short_term')
        track_count = len(top_tracks['items'])
        print(f"✓ Retrieved {track_count} top tracks")
        
        # Test audio features endpoint
        if track_count > 0:
            print("\nTesting audio features endpoint...")
            track_id = top_tracks['items'][0]['id']
            features = sp.audio_features([track_id])
            
            if features and features[0]:
                print(f"✓ Audio features working")
                print(f"  → Sample: energy={features[0]['energy']:.2f}, "
                      f"valence={features[0]['valence']:.2f}")
            else:
                print("⚠ Audio features returned None")
        
        print("\n✓ All API endpoints accessible")
        return True
        
    except spotipy.exceptions.SpotifyException as e:
        print(f"❌ Spotify API error: {e}")
        
        if e.http_status == 401:
            print("   → Authentication failed")
            print("   Action: Delete .cache and re-authenticate")
        elif e.http_status == 429:
            print("   → Rate limit exceeded")
            print("   Action: Wait 60 seconds and try again")
        else:
            print(f"   → HTTP {e.http_status}")
        
        return False
        
    except Exception as e:
        print(f"❌ Unexpected error: {e}")
        print("   Action: Check network connection")
        return False


def run_full_diagnostic():
    """Run all diagnostic checks."""
    print("\n" + "="*60)
    print("MUSIC TIME MACHINE - SYSTEM DIAGNOSTICS")
    print("="*60)
    
    results = {
        'database': check_database(),
        'oauth': check_oauth(),
        'api': check_api_access()
    }
    
    print("\n" + "="*60)
    print("DIAGNOSTIC SUMMARY")
    print("="*60)
    
    for component, status in results.items():
        status_icon = "✓" if status else "❌"
        print(f"{status_icon} {component.upper()}: {'PASS' if status else 'FAIL'}")
    
    all_pass = all(results.values())
    
    if all_pass:
        print("\n✓ All systems operational")
        print("  Your Music Time Machine should work correctly")
    else:
        print("\n❌ System has issues")
        print("  Review the diagnostics above for specific problems")
    
    print("="*60 + "\n")
    
    return all_pass


if __name__ == "__main__":
    success = run_full_diagnostic()
    sys.exit(0 if success else 1)

music_time_machine/
├── music_time_machine.py      # Main application
├── features.py                # Feature implementations
├── database.py                # Database operations
├── schema.sql                 # Database schema
├── .env                       # Credentials (gitignored)
├── tests/
│   ├── __init__.py
│   ├── test_forgotten_gems.py
│   ├── test_monthly_snapshots.py
│   ├── test_mood_playlists.py
│   └── test_database.py
└── requirements.txt

import unittest
import sqlite3
from datetime import date, timedelta
from unittest.mock import Mock, patch
import sys
import os

# Add parent directory to path so we can import our modules
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from features import find_forgotten_gems, create_forgotten_gems_playlist

class TestForgottenGems(unittest.TestCase):
    """Test suite for Forgotten Gems feature"""
    
    def setUp(self):
        """Run before each test - create in-memory database"""
        # Use in-memory database for test isolation
        self.conn = sqlite3.connect(':memory:')
        
        # Create schema
        self.conn.executescript("""
            CREATE TABLE tracks (
                track_id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                artist_name TEXT NOT NULL,
                album_name TEXT NOT NULL,
                duration_ms INTEGER NOT NULL
            );
            
            CREATE TABLE snapshots (
                track_id TEXT NOT NULL,
                snapshot_date DATE NOT NULL,
                time_range TEXT NOT NULL,
                rank INTEGER NOT NULL,
                PRIMARY KEY (track_id, snapshot_date, time_range)
            );
        """)
    
    def tearDown(self):
        """Run after each test - cleanup"""
        self.conn.close()
    
    def test_finds_tracks_from_old_snapshots_not_in_recent(self):
        """Should return tracks from 90-365 days ago that aren't in last 30 days"""
        # Arrange: Insert test data
        old_date = (date.today() - timedelta(days=180)).isoformat()
        recent_date = (date.today() - timedelta(days=10)).isoformat()
        
        # Track that appears in old snapshot only
        self.conn.execute(
            "INSERT INTO tracks VALUES (?, ?, ?, ?, ?)",
            ('track1', 'Old Favorite', 'Artist A', 'Album A', 240000)
        )
        self.conn.execute(
            "INSERT INTO snapshots VALUES (?, ?, ?, ?)",
            ('track1', old_date, 'short_term', 1)
        )
        
        # Track that appears in both (should be excluded)
        self.conn.execute(
            "INSERT INTO tracks VALUES (?, ?, ?, ?, ?)",
            ('track2', 'Still Playing', 'Artist B', 'Album B', 200000)
        )
        self.conn.execute(
            "INSERT INTO snapshots VALUES (?, ?, ?, ?)",
            ('track2', old_date, 'short_term', 2)
        )
        self.conn.execute(
            "INSERT INTO snapshots VALUES (?, ?, ?, ?)",
            ('track2', recent_date, 'short_term', 1)
        )
        
        self.conn.commit()
        
        # Act: Call function
        forgotten = find_forgotten_gems(self.conn, limit=10)
        
        # Assert: Check results
        self.assertEqual(len(forgotten), 1)
        self.assertEqual(forgotten[0][1], 'Old Favorite')  # Check track name
        self.assertEqual(forgotten[0][2], 'Artist A')      # Check artist
    
    def test_returns_empty_list_when_no_old_snapshots(self):
        """Should return empty list when no snapshots exist from 90-365 days ago"""
        # Arrange: Only recent snapshot
        recent_date = (date.today() - timedelta(days=5)).isoformat()
        
        self.conn.execute(
            "INSERT INTO tracks VALUES (?, ?, ?, ?, ?)",
            ('track1', 'Recent Track', 'Artist A', 'Album A', 240000)
        )
        self.conn.execute(
            "INSERT INTO snapshots VALUES (?, ?, ?, ?)",
            ('track1', recent_date, 'short_term', 1)
        )
        self.conn.commit()
        
        # Act
        forgotten = find_forgotten_gems(self.conn, limit=10)
        
        # Assert
        self.assertEqual(len(forgotten), 0)
    
    def test_respects_limit_parameter(self):
        """Should return at most 'limit' tracks even if more qualify"""
        # Arrange: Insert 10 forgotten tracks
        old_date = (date.today() - timedelta(days=180)).isoformat()
        
        for i in range(10):
            track_id = f'track{i}'
            self.conn.execute(
                "INSERT INTO tracks VALUES (?, ?, ?, ?, ?)",
                (track_id, f'Track {i}', 'Artist', 'Album', 240000)
            )
            self.conn.execute(
                "INSERT INTO snapshots VALUES (?, ?, ?, ?)",
                (track_id, old_date, 'short_term', i+1)
            )
        
        self.conn.commit()
        
        # Act: Request only 5
        forgotten = find_forgotten_gems(self.conn, limit=5)
        
        # Assert
        self.assertEqual(len(forgotten), 5)

if __name__ == '__main__':
    unittest.main()

# Run all tests
python -m unittest discover tests/

# Run specific test file
python -m unittest tests.test_forgotten_gems

# Run specific test
python -m unittest tests.test_forgotten_gems.TestForgottenGems.test_respects_limit_parameter

# Run with verbose output
python -m unittest discover tests/ -v

...
----------------------------------------------------------------------
Ran 3 tests in 0.012s

OK

import unittest
from unittest.mock import Mock, patch
import sqlite3
from datetime import date

from features import create_monthly_snapshot

class TestMonthlySnapshots(unittest.TestCase):
    
    def setUp(self):
        """Create in-memory database for tests"""
        self.conn = sqlite3.connect(':memory:')
        self.conn.executescript("""
            CREATE TABLE tracks (
                track_id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                artist_name TEXT NOT NULL,
                album_name TEXT NOT NULL,
                duration_ms INTEGER NOT NULL,
                popularity INTEGER,
                album_image_url TEXT,
                spotify_url TEXT
            );
            
            CREATE TABLE snapshots (
                track_id TEXT NOT NULL,
                snapshot_date DATE NOT NULL,
                time_range TEXT NOT NULL,
                rank INTEGER NOT NULL,
                PRIMARY KEY (track_id, snapshot_date, time_range)
            );
        """)
    
    def tearDown(self):
        self.conn.close()
    
    def test_creates_snapshot_with_mocked_spotify(self):
        """Should fetch tracks from Spotify, save to database, and create playlist"""
        # Arrange: Create mock Spotify client
        mock_sp = Mock()
        
        # Mock the API responses
        mock_sp.current_user_top_tracks.return_value = {
            'items': [
                {
                    'id': 'track1',
                    'name': 'Test Track 1',
                    'artists': [{'name': 'Artist A'}],
                    'album': {
                        'name': 'Album A',
                        'images': [{'url': 'http://example.com/image1.jpg'}]
                    },
                    'duration_ms': 240000,
                    'popularity': 85,
                    'external_urls': {'spotify': 'https://open.spotify.com/track/track1'}
                },
                {
                    'id': 'track2',
                    'name': 'Test Track 2',
                    'artists': [{'name': 'Artist B'}],
                    'album': {
                        'name': 'Album B',
                        'images': [{'url': 'http://example.com/image2.jpg'}]
                    },
                    'duration_ms': 200000,
                    'popularity': 90,
                    'external_urls': {'spotify': 'https://open.spotify.com/track/track2'}
                }
            ]
        }
        
        mock_sp.current_user.return_value = {
            'id': 'test_user',
            'display_name': 'Test User'
        }
        
        mock_sp.user_playlist_create.return_value = {
            'id': 'playlist123',
            'external_urls': {'spotify': 'https://open.spotify.com/playlist/playlist123'}
        }
        
        # Act: Call function with mock
        playlist_url, track_count = create_monthly_snapshot(
            mock_sp, 
            self.conn, 
            time_range='short_term'
        )
        
        # Assert: Verify tracks were saved to database
        cursor = self.conn.execute("SELECT COUNT(*) FROM tracks")
        self.assertEqual(cursor.fetchone()[0], 2)
        
        cursor = self.conn.execute("SELECT COUNT(*) FROM snapshots")
        self.assertEqual(cursor.fetchone()[0], 2)
        
        # Assert: Verify Spotify methods were called correctly
        mock_sp.current_user_top_tracks.assert_called_once_with(
            limit=50, 
            time_range='short_term'
        )
        mock_sp.user_playlist_create.assert_called_once()
        mock_sp.playlist_add_items.assert_called_once()
        
        # Assert: Verify return values
        self.assertIsNotNone(playlist_url)
        self.assertEqual(track_count, 2)
    
    def test_prevents_duplicate_snapshots_same_day(self):
        """Should refuse to create second snapshot on same day"""
        # Arrange: Create existing snapshot
        today = date.today().isoformat()
        self.conn.execute(
            "INSERT INTO tracks VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            ('track1', 'Track', 'Artist', 'Album', 240000, 80, None, 'url')
        )
        self.conn.execute(
            "INSERT INTO snapshots VALUES (?, ?, ?, ?)",
            ('track1', today, 'short_term', 1)
        )
        self.conn.commit()
        
        mock_sp = Mock()
        
        # Act: Try to create another snapshot
        playlist_url, track_count = create_monthly_snapshot(
            mock_sp,
            self.conn,
            time_range='short_term'
        )
        
        # Assert: Should return None (refused to create duplicate)
        self.assertIsNone(playlist_url)
        self.assertEqual(track_count, 0)
        
        # Assert: Spotify API should not have been called
        mock_sp.current_user_top_tracks.assert_not_called()
    
    def test_handles_spotify_api_error(self):
        """Should handle gracefully when Spotify API fails"""
        # Arrange: Mock that raises exception
        mock_sp = Mock()
        mock_sp.current_user_top_tracks.side_effect = Exception("Spotify API error")
        
        # Act & Assert: Should not crash
        with self.assertRaises(Exception):
            create_monthly_snapshot(mock_sp, self.conn)

if __name__ == '__main__':
    unittest.main()

import unittest
import sqlite3
from unittest.mock import Mock

from features import find_tracks_by_mood, create_mood_playlist

class TestMoodPlaylists(unittest.TestCase):
    
    def setUp(self):
        """Create database with tracks and audio features"""
        self.conn = sqlite3.connect(':memory:')
        self.conn.executescript("""
            CREATE TABLE tracks (
                track_id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                artist_name TEXT NOT NULL,
                album_name TEXT NOT NULL,
                duration_ms INTEGER NOT NULL
            );
            
            CREATE TABLE audio_features (
                track_id TEXT PRIMARY KEY,
                energy REAL NOT NULL,
                valence REAL NOT NULL,
                danceability REAL NOT NULL,
                tempo REAL NOT NULL,
                acousticness REAL NOT NULL,
                instrumentalness REAL NOT NULL,
                speechiness REAL NOT NULL
            );
        """)
        
        # Insert test tracks with various audio features
        test_data = [
            # High energy workout tracks
            ('track1', 'Workout Track 1', 'Artist A', 'Album A', 0.85, 0.70, 0.75, 150),
            ('track2', 'Workout Track 2', 'Artist B', 'Album B', 0.90, 0.65, 0.80, 160),
            
            # Low energy focus tracks
            ('track3', 'Focus Track 1', 'Artist C', 'Album C', 0.25, 0.40, 0.30, 95),
            ('track4', 'Focus Track 2', 'Artist D', 'Album D', 0.30, 0.35, 0.25, 100),
            
            # Medium energy chill tracks
            ('track5', 'Chill Track 1', 'Artist E', 'Album E', 0.40, 0.55, 0.40, 90),
        ]
        
        for track_id, name, artist, album, energy, valence, dance, tempo in test_data:
            self.conn.execute(
                "INSERT INTO tracks VALUES (?, ?, ?, ?, ?)",
                (track_id, name, artist, album, 240000)
            )
            self.conn.execute(
                "INSERT INTO audio_features VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                (track_id, energy, valence, dance, tempo, 0.5, 0.5, 0.05)
            )
        
        self.conn.commit()
    
    def tearDown(self):
        self.conn.close()
    
    def test_workout_mood_filters_correctly(self):
        """Should return only high energy, high tempo tracks for workout mood"""
        # Act: Find workout tracks
        tracks = find_tracks_by_mood(self.conn, 'workout', limit=10)
        
        # Assert: Should only return the 2 workout tracks
        self.assertEqual(len(tracks), 2)
        
        # Verify both returned tracks meet workout criteria
        for track in tracks:
            track_id, name, artist, album, energy, valence, tempo = track
            self.assertGreater(energy, 0.75, f"Track {name} energy too low")
            self.assertGreater(tempo, 140, f"Track {name} tempo too low")
    
    def test_focus_mood_filters_correctly(self):
        """Should return only low energy tracks for focus mood"""
        # Act
        tracks = find_tracks_by_mood(self.conn, 'focus', limit=10)
        
        # Assert
        self.assertEqual(len(tracks), 2)
        
        for track in tracks:
            energy = track[4]
            self.assertLess(energy, 0.40, "Focus track energy too high")
    
    def test_chill_mood_filters_correctly(self):
        """Should return medium energy, moderate tempo tracks for chill mood"""
        # Act
        tracks = find_tracks_by_mood(self.conn, 'chill', limit=10)
        
        # Assert: Should return the 1 chill track
        self.assertEqual(len(tracks), 1)
        self.assertEqual(tracks[0][1], 'Chill Track 1')
    
    def test_returns_empty_when_no_tracks_match_mood(self):
        """Should return empty list when no tracks match mood criteria"""
        # Arrange: Clear database and insert only low-energy tracks
        self.conn.execute("DELETE FROM audio_features")
        self.conn.execute(
            "INSERT INTO audio_features VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            ('track1', 0.20, 0.30, 0.25, 80, 0.5, 0.5, 0.05)
        )
        self.conn.commit()
        
        # Act: Try to find workout tracks (should find none)
        tracks = find_tracks_by_mood(self.conn, 'workout', limit=10)
        
        # Assert
        self.assertEqual(len(tracks), 0)
    
    def test_respects_limit_parameter(self):
        """Should return at most 'limit' tracks"""
        # Act: Request only 1 workout track (even though 2 exist)
        tracks = find_tracks_by_mood(self.conn, 'workout', limit=1)
        
        # Assert
        self.assertEqual(len(tracks), 1)
    
    def test_create_mood_playlist_with_mocked_spotify(self):
        """Should create playlist with tracks matching mood"""
        # Arrange: Mock Spotify client
        mock_sp = Mock()
        mock_sp.current_user.return_value = {'id': 'test_user'}
        mock_sp.user_playlist_create.return_value = {
            'id': 'playlist123',
            'external_urls': {'spotify': 'https://open.spotify.com/playlist/123'}
        }
        
        # Act: Create workout playlist
        playlist_url = create_mood_playlist(mock_sp, self.conn, 'workout', limit=25)
        
        # Assert: Verify playlist was created with correct tracks
        self.assertIsNotNone(playlist_url)
        mock_sp.user_playlist_create.assert_called_once()
        
        # Verify playlist_add_items was called with workout track URIs
        call_args = mock_sp.playlist_add_items.call_args
        added_uris = call_args[0][1]  # Second argument to playlist_add_items
        
        # Should have 2 workout tracks
        self.assertEqual(len(added_uris), 2)
        self.assertTrue(all(uri.startswith('spotify:track:') for uri in added_uris))

if __name__ == '__main__':
    unittest.main()

from datetime import date, timedelta
import unittest

class TestTimeDependentLogic(unittest.TestCase):
    
    def test_forgotten_gems_date_ranges(self):
        """Test that date calculations are correct"""
        # Calculate expected date boundaries
        today = date.today()
        days_90_ago = today - timedelta(days=90)
        days_365_ago = today - timedelta(days=365)
        days_30_ago = today - timedelta(days=30)
        
        # Verify dates fall in expected windows
        self.assertLess(days_365_ago, days_90_ago)
        self.assertLess(days_30_ago, today)
        
        # These date calculations should match your SQL queries
        # SELECT ... WHERE snapshot_date BETWEEN date('now', '-365 days') AND date('now', '-90 days')
    
    def test_boundary_conditions_for_dates(self):
        """Test edge cases at date boundaries"""
        conn = sqlite3.connect(':memory:')
        conn.executescript("""
            CREATE TABLE tracks (track_id TEXT PRIMARY KEY, name TEXT, 
                               artist_name TEXT, album_name TEXT, duration_ms INTEGER);
            CREATE TABLE snapshots (track_id TEXT, snapshot_date DATE, 
                                  time_range TEXT, rank INTEGER);
        """)
        
        # Test track exactly 90 days ago (should be included)
        exactly_90_days = (date.today() - timedelta(days=90)).isoformat()
        conn.execute("INSERT INTO tracks VALUES ('track1', 'Name', 'Artist', 'Album', 240000)")
        conn.execute("INSERT INTO snapshots VALUES ('track1', ?, 'short_term', 1)", (exactly_90_days,))
        
        # Test track exactly 365 days ago (should be included)
        exactly_365_days = (date.today() - timedelta(days=365)).isoformat()
        conn.execute("INSERT INTO tracks VALUES ('track2', 'Name2', 'Artist', 'Album', 240000)")
        conn.execute("INSERT INTO snapshots VALUES ('track2', ?, 'short_term', 1)", (exactly_365_days,))
        
        # Test track 89 days ago (too recent, should be excluded)
        days_89_ago = (date.today() - timedelta(days=89)).isoformat()
        conn.execute("INSERT INTO tracks VALUES ('track3', 'Name3', 'Artist', 'Album', 240000)")
        conn.execute("INSERT INTO snapshots VALUES ('track3', ?, 'short_term', 1)", (days_89_ago,))
        
        # Test track 366 days ago (too old, should be excluded)
        days_366_ago = (date.today() - timedelta(days=366)).isoformat()
        conn.execute("INSERT INTO tracks VALUES ('track4', 'Name4', 'Artist', 'Album', 240000)")
        conn.execute("INSERT INTO snapshots VALUES ('track4', ?, 'short_term', 1)", (days_366_ago,))
        
        conn.commit()
        
        # Act: Find forgotten gems
        from features import find_forgotten_gems
        forgotten = find_forgotten_gems(conn, limit=10)
        
        # Assert: Should include tracks at exactly 90 and 365 days, exclude 89 and 366
        self.assertEqual(len(forgotten), 2)
        track_names = [track[1] for track in forgotten]
        self.assertIn('Name', track_names)   # 90 days
        self.assertIn('Name2', track_names)  # 365 days
        self.assertNotIn('Name3', track_names)  # 89 days (too recent)
        self.assertNotIn('Name4', track_names)  # 366 days (too old)
        
        conn.close()

# Install coverage tool
pip install coverage

# Run tests with coverage tracking
coverage run -m unittest discover tests/

# Generate coverage report
coverage report

# Generate detailed HTML report
coverage html
# Open htmlcov/index.html in browser to see line-by-line coverage

Name                    Stmts   Miss  Cover
-------------------------------------------
features.py               145     12    92%
database.py                67      8    88%
music_time_machine.py      89     25    72%
-------------------------------------------
TOTAL                     301     45    85%

name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'
    
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
        pip install coverage
    
    - name: Run tests
      run: |
        coverage run -m unittest discover tests/
        coverage report
        coverage xml
    
    - name: Upload coverage
      uses: codecov/codecov-action@v2
      with:
        files: ./coverage.xml