Scrobbling and Data Tracking

# Scrobbling methods available on LastFMNetwork and LibreFMNetwork

def scrobble(self, artist: str, title: str, timestamp: int, album=None, track_number=None, mbid=None, duration=None) -> None:
    """
    Scrobble a single track.
    
    Args:
        artist (str): Artist name
        title (str): Track title
        timestamp (int): Unix timestamp when track was played
        album (str, optional): Album name
        track_number (int, optional): Track number on album
        mbid (str, optional): MusicBrainz track ID
        duration (int, optional): Track duration in seconds
    
    Raises:
        WSError: If scrobbling fails due to API error
        NetworkError: If network connection fails
    """

def update_now_playing(self, artist: str, title: str, album=None, track_number=None, mbid=None, duration=None) -> None:
    """
    Update now playing status without scrobbling.
    
    Args:
        artist (str): Artist name
        title (str): Track title
        album (str, optional): Album name
        track_number (int, optional): Track number on album
        mbid (str, optional): MusicBrainz track ID
        duration (int, optional): Track duration in seconds
    """

def scrobble_many(self, tracks: list[dict]) -> None:
    """
    Scrobble multiple tracks at once.
    
    Args:
        tracks (list[dict]): List of track dictionaries, each containing:
            - artist (str): Artist name
            - title (str): Track title
            - timestamp (int): Unix timestamp when played
            - album (str, optional): Album name
            - track_number (int, optional): Track number
            - mbid (str, optional): MusicBrainz track ID
            - duration (int, optional): Duration in seconds
    
    Example:
        tracks = [
            {
                'artist': 'The Beatles',
                'title': 'Come Together',
                'timestamp': 1234567890,
                'album': 'Abbey Road',
                'track_number': 1
            },
            {
                'artist': 'Pink Floyd',
                'title': 'Time',
                'timestamp': 1234567950,
                'album': 'The Dark Side of the Moon',
                'duration': 421
            }
        ]
        network.scrobble_many(tracks)
    """

# Scrobble source constants
SCROBBLE_SOURCE_USER = "P"                          # User chose the music
SCROBBLE_SOURCE_NON_PERSONALIZED_BROADCAST = "R"   # Non-personalized broadcast (e.g., radio)
SCROBBLE_SOURCE_PERSONALIZED_BROADCAST = "E"       # Personalized broadcast (e.g., personalized radio)
SCROBBLE_SOURCE_LASTFM = "L"                        # Last.fm recommendation
SCROBBLE_SOURCE_UNKNOWN = "U"                       # Unknown source

# Scrobble mode constants
SCROBBLE_MODE_PLAYED = ""                           # Track was played normally
SCROBBLE_MODE_LOVED = "L"                           # Track was loved
SCROBBLE_MODE_BANNED = "B"                          # Track was banned/skipped
SCROBBLE_MODE_SKIPPED = "S"                         # Track was skipped

# Scrobbling requirements and validation

"""
Scrobbling Requirements:
- Must have valid API key and secret
- Must have valid session key (authenticated user)
- Track must be at least 30 seconds long
- Track must be played for at least 240 seconds OR at least half the track length
- Timestamp must be accurate (not future time)
- Artist and title are required fields

Validation Notes:
- Album, track number, MusicBrainz ID, and duration are optional but recommended
- MusicBrainz IDs provide better matching accuracy
- Duration should be in seconds for scrobbling, milliseconds for track objects
- Timestamps should be Unix timestamps (seconds since epoch)

Rate Limiting:
- Maximum 50 scrobbles per request with scrobble_many()
- API rate limit of 5 requests per second (handled automatically if rate limiting enabled)
- Failed scrobbles can be retried with exponential backoff
"""

def md5(text: str) -> str:
    """
    Create MD5 hash of text (used for password hashing).
    
    Args:
        text (str): Text to hash
        
    Returns:
        str: MD5 hash in hexadecimal format
        
    Example:
        password_hash = pylast.md5("my_password")
    """

# Time utility for timestamp generation
import time

def get_current_timestamp() -> int:
    """
    Get current Unix timestamp for scrobbling.
    
    Returns:
        int: Current Unix timestamp
    """
    return int(time.time())

from collections import namedtuple

PlayedTrack = namedtuple('PlayedTrack', ['track', 'album', 'playback_date', 'timestamp'])
"""
Represents a played track with timestamp information.

Fields:
    track (Track): The played track object
    album (Album): Album containing the track (may be None)
    playback_date (str): Human-readable playback date
    timestamp (int): Unix timestamp of when track was played
"""

import pylast
import time

# Setup authenticated network
network = pylast.LastFMNetwork(
    api_key=API_KEY,
    api_secret=API_SECRET,
    username=username,
    password_hash=pylast.md5(password)
)

# Scrobble a track that just finished playing
timestamp = int(time.time()) - 300  # 5 minutes ago

network.scrobble(
    artist="Pink Floyd",
    title="Wish You Were Here",
    timestamp=timestamp,
    album="Wish You Were Here",
    track_number=1,
    duration=334  # 5 minutes 34 seconds
)

print("Track scrobbled successfully!")

# Update now playing for current track
network.update_now_playing(
    artist="Led Zeppelin",
    title="Stairway to Heaven",
    album="Led Zeppelin IV",
    duration=482
)

print("Now playing updated!")

# Scrobble a listening session
listening_session = [
    {
        'artist': 'The Beatles',
        'title': 'Come Together',
        'timestamp': int(time.time()) - 1200,  # 20 minutes ago
        'album': 'Abbey Road',
        'track_number': 1,
        'duration': 259
    },
    {
        'artist': 'The Beatles',
        'title': 'Something',
        'timestamp': int(time.time()) - 941,   # ~15 minutes ago
        'album': 'Abbey Road',
        'track_number': 2,
        'duration': 182
    },
    {
        'artist': 'The Beatles',
        'title': 'Maxwell\'s Silver Hammer',
        'timestamp': int(time.time()) - 759,   # ~12 minutes ago
        'album': 'Abbey Road',
        'track_number': 3,
        'duration': 207
    },
    {
        'artist': 'The Beatles',
        'title': 'Oh! Darling',
        'timestamp': int(time.time()) - 552,   # ~9 minutes ago
        'album': 'Abbey Road',
        'track_number': 4,
        'duration': 206
    }
]

try:
    network.scrobble_many(listening_session)
    print(f"Successfully scrobbled {len(listening_session)} tracks!")
except pylast.WSError as e:
    print(f"Scrobbling failed: {e}")
    print(f"Error code: {e.get_id()}")

# Scrobble with MusicBrainz IDs for better accuracy
network.scrobble(
    artist="Radiohead",
    title="Paranoid Android",
    timestamp=int(time.time()) - 383,  # ~6 minutes ago
    album="OK Computer",
    track_number=2,
    mbid="8cc5d2a6-1e31-4ca0-8da4-b23a8c4c0bf3",  # MusicBrainz track ID
    duration=383
)

# Batch scrobble with MusicBrainz data
mb_tracks = [
    {
        'artist': 'Pink Floyd',
        'title': 'Time',
        'timestamp': int(time.time()) - 800,
        'album': 'The Dark Side of the Moon',
        'track_number': 4,
        'mbid': '0c9aa3ab-b8ab-4ae2-b777-7b139fd94f34',
        'duration': 421
    },
    {
        'artist': 'Pink Floyd',
        'title': 'The Great Gig in the Sky',
        'timestamp': int(time.time()) - 379,
        'album': 'The Dark Side of the Moon',
        'track_number': 5,
        'mbid': '1cc3ecb0-2fdc-4fc4-a7fb-d48b4949b36a',
        'duration': 284
    }
]

network.scrobble_many(mb_tracks)

def safe_scrobble(network, artist, title, timestamp, **kwargs):
    """Safely scrobble with error handling and validation"""
    
    # Basic validation
    if not artist or not title:
        print("Error: Artist and title are required")
        return False
    
    if timestamp > time.time():
        print("Error: Cannot scrobble future timestamps")
        return False
    
    # Check if timestamp is too old (Last.fm has limits)
    two_weeks_ago = time.time() - (14 * 24 * 60 * 60)
    if timestamp < two_weeks_ago:
        print("Warning: Timestamp is older than 2 weeks, may be rejected")
    
    try:
        network.scrobble(
            artist=artist,
            title=title,
            timestamp=timestamp,
            **kwargs
        )
        print(f"Successfully scrobbled: {artist} - {title}")
        return True
        
    except pylast.WSError as e:
        error_id = e.get_id()
        if error_id == pylast.STATUS_AUTH_FAILED:
            print("Authentication failed - check session key")
        elif error_id == pylast.STATUS_INVALID_PARAMS:
            print("Invalid parameters - check artist/title")
        elif error_id == pylast.STATUS_RATE_LIMIT_EXCEEDED:
            print("Rate limit exceeded - wait before retrying")
        else:
            print(f"API error {error_id}: {e}")
        return False
        
    except pylast.NetworkError as e:
        print(f"Network error: {e}")
        return False

# Use safe scrobbling
safe_scrobble(
    network,
    artist="Queen",
    title="Bohemian Rhapsody",
    timestamp=int(time.time()) - 355,
    album="A Night at the Opera",
    duration=355
)

import csv
import datetime

def scrobble_from_csv(network, csv_file):
    """Scrobble tracks from CSV export (e.g., from music player)"""
    
    tracks_to_scrobble = []
    
    with open(csv_file, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        
        for row in reader:
            # Parse timestamp (assuming ISO format)
            played_at = datetime.datetime.fromisoformat(row['played_at'])
            timestamp = int(played_at.timestamp())
            
            track_data = {
                'artist': row['artist'],
                'title': row['title'],
                'timestamp': timestamp
            }
            
            # Add optional fields if present
            if row.get('album'):
                track_data['album'] = row['album']
            if row.get('duration'):
                track_data['duration'] = int(row['duration'])
            if row.get('track_number'):
                track_data['track_number'] = int(row['track_number'])
            
            tracks_to_scrobble.append(track_data)
    
    # Scrobble in batches of 50 (Last.fm limit)
    batch_size = 50
    for i in range(0, len(tracks_to_scrobble), batch_size):
        batch = tracks_to_scrobble[i:i + batch_size]
        
        try:
            network.scrobble_many(batch)
            print(f"Scrobbled batch {i//batch_size + 1}: {len(batch)} tracks")
            
            # Be nice to the API
            time.sleep(1)
            
        except Exception as e:
            print(f"Failed to scrobble batch {i//batch_size + 1}: {e}")
    
    print(f"Finished scrobbling {len(tracks_to_scrobble)} tracks")

# Example CSV format:
# artist,title,album,played_at,duration,track_number
# "The Beatles","Come Together","Abbey Road","2023-01-01T10:00:00","00:04:19",1
# scrobble_from_csv(network, "listening_history.csv")

class ScrobbleManager:
    """Manage real-time scrobbling for music applications"""
    
    def __init__(self, network):
        self.network = network
        self.current_track = None
        self.track_start_time = None
        self.min_scrobble_time = 30  # Minimum seconds to scrobble
        self.scrobble_threshold = 0.5  # Scrobble at 50% completion
    
    def start_track(self, artist, title, album=None, duration=None):
        """Start playing a new track"""
        self.current_track = {
            'artist': artist,
            'title': title,
            'album': album,
            'duration': duration
        }
        self.track_start_time = time.time()
        
        # Update now playing
        try:
            self.network.update_now_playing(
                artist=artist,
                title=title,
                album=album,
                duration=duration
            )
            print(f"Now playing: {artist} - {title}")
        except Exception as e:
            print(f"Failed to update now playing: {e}")
    
    def stop_track(self):
        """Stop current track and scrobble if criteria met"""
        if not self.current_track or not self.track_start_time:
            return
        
        played_time = time.time() - self.track_start_time
        duration = self.current_track.get('duration', 0)
        
        # Check scrobbling criteria
        should_scrobble = False
        
        if played_time >= self.min_scrobble_time:
            if duration:
                # Scrobble if played at least 50% or 4 minutes, whichever is less
                threshold_time = min(duration * self.scrobble_threshold, 240)
                should_scrobble = played_time >= threshold_time
            else:
                # No duration info, scrobble if played more than 30 seconds
                should_scrobble = played_time >= self.min_scrobble_time
        
        if should_scrobble:
            try:
                self.network.scrobble(
                    artist=self.current_track['artist'],
                    title=self.current_track['title'],
                    timestamp=int(self.track_start_time),
                    album=self.current_track.get('album'),
                    duration=duration
                )
                print(f"Scrobbled: {self.current_track['artist']} - {self.current_track['title']}")
            except Exception as e:
                print(f"Failed to scrobble: {e}")
        else:
            print(f"Track not scrobbled: played {played_time:.1f}s of {duration or 'unknown'}s")
        
        # Clear current track
        self.current_track = None
        self.track_start_time = None

# Usage example
scrobbler = ScrobbleManager(network)

# Simulate music playback
scrobbler.start_track("Led Zeppelin", "Stairway to Heaven", "Led Zeppelin IV", 482)
time.sleep(5)  # Simulate 5 seconds of playback
scrobbler.stop_track()  # Won't scrobble (too short)

scrobbler.start_track("Queen", "Bohemian Rhapsody", "A Night at the Opera", 355)
time.sleep(60)  # Simulate 1 minute of playback
scrobbler.stop_track()  # Will scrobble (>30 seconds and >50% of 355s)

def analyze_scrobbling_stats(user):
    """Analyze user's scrobbling patterns"""
    
    # Get recent scrobbles
    recent_tracks = user.get_recent_tracks(limit=100)
    
    if not recent_tracks:
        print("No recent tracks found")
        return
    
    # Analyze listening patterns
    hourly_stats = {}
    daily_stats = {}
    artist_counts = {}
    
    for played_track in recent_tracks:
        # Parse timestamp
        timestamp = played_track.timestamp
        dt = datetime.datetime.fromtimestamp(timestamp)
        
        # Hour analysis
        hour = dt.hour
        hourly_stats[hour] = hourly_stats.get(hour, 0) + 1
        
        # Day analysis
        day = dt.strftime('%A')
        daily_stats[day] = daily_stats.get(day, 0) + 1
        
        # Artist analysis
        artist = played_track.track.get_artist().get_name()
        artist_counts[artist] = artist_counts.get(artist, 0) + 1
    
    # Print analysis
    print(f"Scrobbling Analysis (last {len(recent_tracks)} tracks):")
    
    print("\nMost active hours:")
    sorted_hours = sorted(hourly_stats.items(), key=lambda x: x[1], reverse=True)
    for hour, count in sorted_hours[:5]:
        print(f"  {hour:02d}:00 - {count} scrobbles")
    
    print("\nMost active days:")
    sorted_days = sorted(daily_stats.items(), key=lambda x: x[1], reverse=True)
    for day, count in sorted_days:
        print(f"  {day} - {count} scrobbles")
    
    print("\nTop artists:")
    sorted_artists = sorted(artist_counts.items(), key=lambda x: x[1], reverse=True)
    for artist, count in sorted_artists[:10]:
        print(f"  {artist} - {count} scrobbles")

# Analyze scrobbling patterns
user = network.get_authenticated_user()
analyze_scrobbling_stats(user)