vrobbler/vrobbler/apps/music/lastfm.py

import logging
import time
from datetime import datetime, timedelta

import pylast
import pytz
from django.conf import settings
from django.utils import timezone
from music.utils import (
    get_or_create_album,
    get_or_create_artist,
    get_or_create_track,
)

logger = logging.getLogger(__name__)

PYLAST_ERRORS = tuple(
    getattr(pylast, exc_name)
    for exc_name in (
        "ScrobblingError",
        "NetworkError",
        "MalformedResponseError",
        "WSError",
    )
    if hasattr(pylast, exc_name)
)


class LastFM:
    def __init__(self, user):
        try:
            self.client = pylast.LastFMNetwork(
                api_key=getattr(settings, "LASTFM_API_KEY"),
                api_secret=getattr(settings, "LASTFM_SECRET_KEY"),
                username=user.profile.lastfm_username,
                password_hash=pylast.md5(user.profile.lastfm_password),
            )
            self.user = self.client.get_user(user.profile.lastfm_username)
            self.vrobbler_user = user
        except PYLAST_ERRORS as e:
            logger.error(f"Error during Last.fm setup: {e}")

    def import_from_lastfm(self, last_processed=None):
        """Given a last processed time, import all scrobbles from LastFM since then"""
        from scrobbles.models import Scrobble

        new_scrobbles = []
        source = "Last.fm"
        lastfm_scrobbles = self.get_last_scrobbles(time_from=last_processed)

        for lfm_scrobble in lastfm_scrobbles:
            track = get_or_create_track(
                lfm_scrobble,
                {
                    "TRACK_TITLE": "track",
                    "ARTIST_NAME": "artist",
                    "ALBUM_NAME": "album",
                    "TIMESTAMP": "timestamp",
                },
            )

            timezone = settings.TIME_ZONE
            if self.vrobbler_user.profile:
                timezone = self.vrobbler_user.profile.timezone

            new_scrobble = Scrobble(
                user=self.vrobbler_user,
                timestamp=timestamp,
                source=source,
                track=track,
                timezone=timezone,
                played_to_completion=True,
                in_progress=False,
                media_type=Scrobble.MediaType.TRACK,
            )
            # Vrobbler scrobbles on finish, LastFM scrobbles on start
            seconds_eariler = timestamp - timedelta(seconds=20)
            seconds_later = timestamp + timedelta(seconds=20)
            existing = Scrobble.objects.filter(
                created__gte=seconds_eariler,
                created__lte=seconds_later,
                track=track,
            ).first()
            if existing:
                logger.debug(f"Skipping existing scrobble {new_scrobble}")
                continue
            logger.debug(f"Queued scrobble {new_scrobble} for creation")
            new_scrobbles.append(new_scrobble)

        created = Scrobble.objects.bulk_create(new_scrobbles)
        logger.info(
            f"Created {len(created)} scrobbles",
            extra={"created_scrobbles": created},
        )
        return created

    def get_last_scrobbles(self, time_from=None, time_to=None):
        """Given a user, Last.fm api key, and secret key, grab a list of scrobbled
        tracks"""
        lfm_params = {}
        scrobbles = []
        if time_from:
            lfm_params["time_from"] = int(time_from.timestamp())
        if time_to:
            lfm_params["time_to"] = int(time_to.timestamp())

        # if not time_from and not time_to:
        lfm_params["limit"] = None

        found_scrobbles = self.user.get_recent_tracks(**lfm_params)
        # TOOD spin this out into a celery task over certain threshold of found scrobbles?

        for scrobble in found_scrobbles:
            logger.debug(f"Processing {scrobble}")
            run_time = None
            mbid = None
            artist = None

            try:
                run_time = int(scrobble.track.get_duration() / 1000)
                mbid = scrobble.track.get_mbid()
                artist = scrobble.track.get_artist().name
            except pylast.MalformedResponseError as e:
                logger.warn(e)
            except pylast.WSError as e:
                logger.warn(
                    "LastFM barfed trying to get the track for {scrobble.track}"
                )
            except pylast.NetworkError as e:
                logger.warn(
                    "LastFM barfed trying to get the track for {scrobble.track}"
                )

            if not artist:
                logger.warn(f"Silly LastFM, no artist found for {scrobble}")
                continue

            timestamp = datetime.utcfromtimestamp(
                int(scrobble.timestamp)
            ).replace(tzinfo=pytz.utc)

            logger.info(f"{artist},{scrobble.track.title},{timestamp}")
            scrobbles.append(
                {
                    "artist": artist,
                    "album": scrobble.album,
                    "title": scrobble.track.title,
                    "mbid": mbid,
                    "run_time_seconds": run_time,
                    "timestamp": timestamp,
                }
            )
        return scrobbles