298 lines
10 KiB
Python
298 lines
10 KiB
Python
import calendar
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
|
||
import pylast
|
||
import pytz
|
||
from celery import shared_task
|
||
from dateutil.relativedelta import relativedelta
|
||
from django.conf import settings
|
||
from django.contrib.auth import get_user_model
|
||
from django.utils import timezone
|
||
from music.models import 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):
|
||
self.user = None
|
||
self.vrobbler_user = 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)
|
||
except PYLAST_ERRORS as e:
|
||
logger.error(f"Error during Last.fm setup: {e}")
|
||
raise
|
||
|
||
def import_from_lastfm(self, last_processed=None, time_to=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, time_to=time_to
|
||
)
|
||
|
||
for lfm_scrobble in lastfm_scrobbles:
|
||
track = Track.find_or_create(
|
||
title=lfm_scrobble.get("title"),
|
||
artist_name=lfm_scrobble.get("artist"),
|
||
album_name=lfm_scrobble.get("album", ""),
|
||
run_time_seconds=lfm_scrobble.get("run_time_seconds"),
|
||
mbid=lfm_scrobble.get("mbid"),
|
||
enrich=False,
|
||
)
|
||
|
||
tz_timestamp = self.vrobbler_user.profile.get_timestamp_with_tz(
|
||
lfm_scrobble.get("timestamp")
|
||
)
|
||
timestamp = lfm_scrobble.get("timestamp")
|
||
stop_timestamp = timestamp + timedelta(seconds=track.run_time_seconds)
|
||
new_scrobble = Scrobble(
|
||
user=self.vrobbler_user,
|
||
timestamp=timestamp,
|
||
stop_timestamp=stop_timestamp,
|
||
source=source,
|
||
track=track,
|
||
played_to_completion=True,
|
||
in_progress=False,
|
||
media_type=Scrobble.MediaType.TRACK,
|
||
timezone=tz_timestamp.tzinfo.name,
|
||
visibility="private",
|
||
)
|
||
# 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"Last.fm import fnished",
|
||
extra={
|
||
"scrobbles_created": len(created),
|
||
"user_id": self.vrobbler_user,
|
||
"lastfm_user": self.user,
|
||
},
|
||
)
|
||
return created
|
||
|
||
def get_earliest_scrobble_timestamp(self) -> datetime | None:
|
||
"""Return the user's Last.fm registration date as the earliest
|
||
possible scrobble timestamp.
|
||
|
||
Uses ``get_unixtime_registered()`` which is a single API call.
|
||
Returns ``None`` if the user has no scrobbles at all.
|
||
"""
|
||
if not self.get_last_scrobbles(check=True):
|
||
return None
|
||
registered_ts = self.user.get_unixtime_registered()
|
||
earliest = datetime.fromtimestamp(registered_ts, tz=pytz.utc)
|
||
logger.info("Last.fm user %s registered on %s", self.user.name, earliest)
|
||
return earliest
|
||
|
||
def get_last_scrobbles(self, time_from=None, time_to=None, check=False):
|
||
"""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())
|
||
|
||
lfm_params["limit"] = 1 if check else None
|
||
|
||
found_scrobbles = self.user.get_recent_tracks(**lfm_params)
|
||
|
||
if check:
|
||
return bool(found_scrobbles)
|
||
|
||
for scrobble in found_scrobbles:
|
||
logger.info(f"Processing {scrobble}")
|
||
run_time = None
|
||
mbid = None
|
||
artist = None
|
||
|
||
log_dict = {"scrobble": scrobble}
|
||
try:
|
||
run_time = int(scrobble.track.get_duration() / 1000)
|
||
mbid = scrobble.track.get_mbid()
|
||
artist = scrobble.track.get_artist().name
|
||
log_dict["artist"] = artist
|
||
log_dict["mbid"] = mbid
|
||
log_dict["run_time"] = run_time
|
||
except pylast.MalformedResponseError as e:
|
||
logger.warning(e)
|
||
except pylast.WSError as e:
|
||
logger.info(
|
||
"LastFM barfed trying to get the track for {scrobble.track}",
|
||
extra=log_dict,
|
||
)
|
||
except pylast.NetworkError as e:
|
||
logger.info(
|
||
"LastFM barfed trying to get the track for {scrobble.track}",
|
||
extra=log_dict,
|
||
)
|
||
|
||
if not artist:
|
||
logger.info(
|
||
f"Silly LastFM, no artist found for scrobble",
|
||
extra=log_dict,
|
||
)
|
||
continue
|
||
|
||
# TODO figure out if this will actually work
|
||
# timestamp = datetime.fromtimestamp(int(scrobble.timestamp), UTC)
|
||
timestamp = datetime.utcfromtimestamp(int(scrobble.timestamp)).replace(
|
||
tzinfo=pytz.utc
|
||
)
|
||
|
||
logger.info(f"Scrobble appended to list for bulk create", extra=log_dict)
|
||
scrobbles.append(
|
||
{
|
||
"artist": artist,
|
||
"album": scrobble.album,
|
||
"title": scrobble.track.title,
|
||
"mbid": mbid,
|
||
"run_time_seconds": run_time,
|
||
"timestamp": timestamp,
|
||
}
|
||
)
|
||
return scrobbles
|
||
|
||
|
||
@shared_task
|
||
def dispatch_historical_imports(user_id):
|
||
"""Find every un-imported month and dispatch a sub-task for each.
|
||
|
||
Walks backward month by month from the user's earliest completed
|
||
``LastFmImport`` (or the current month if none exists) to their
|
||
first Last.fm scrobble, creating a ``LastFmImport`` record and
|
||
dispatching ``process_lastfm_import`` for each month that has
|
||
scrobbles.
|
||
"""
|
||
from scrobbles.models import LastFmImport
|
||
from scrobbles.tasks import process_lastfm_import
|
||
|
||
user = get_user_model().objects.filter(id=user_id).first()
|
||
if not user:
|
||
logger.error("User %s not found", user_id)
|
||
return 0
|
||
|
||
try:
|
||
lastfm = LastFM(user)
|
||
except Exception:
|
||
logger.exception("Failed to set up LastFM client for user %s", user_id)
|
||
return 0
|
||
|
||
logger.info("Looking up earliest Last.fm scrobble for user %s", user_id)
|
||
earliest_scrobble = lastfm.get_earliest_scrobble_timestamp()
|
||
if earliest_scrobble is None:
|
||
logger.info("User %s has no Last.fm scrobbles", user_id)
|
||
return 0
|
||
logger.info("Earliest Last.fm scrobble for user %s: %s", user_id, earliest_scrobble)
|
||
|
||
in_progress = LastFmImport.objects.filter(
|
||
user_id=user_id, processed_finished__isnull=True
|
||
).exists()
|
||
if in_progress:
|
||
logger.info(
|
||
"User %s has an import still in progress, skipping dispatch",
|
||
user_id,
|
||
)
|
||
return 0
|
||
|
||
def _first_of_month(dt):
|
||
return dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
||
|
||
def _last_of_month(dt):
|
||
_, last_day = calendar.monthrange(dt.year, dt.month)
|
||
return dt.replace(
|
||
day=last_day, hour=23, minute=59, second=59, microsecond=999999
|
||
)
|
||
|
||
earliest = (
|
||
LastFmImport.objects.filter(user_id=user_id, processed_finished__isnull=False)
|
||
.order_by("processed_finished")
|
||
.first()
|
||
)
|
||
if earliest:
|
||
earliest_log_scrobble = earliest.scrobbles().order_by("timestamp").first()
|
||
cursor = (
|
||
_first_of_month(earliest_log_scrobble.timestamp)
|
||
if earliest_log_scrobble
|
||
else earliest.processed_finished
|
||
)
|
||
logger.info(
|
||
"Found existing import; earliest scrobble %s, cursor set to %s",
|
||
earliest_log_scrobble.timestamp if earliest_log_scrobble else None,
|
||
cursor,
|
||
)
|
||
else:
|
||
cursor = timezone.now()
|
||
logger.info("No existing imports; cursor set to %s", cursor)
|
||
|
||
dispatched = 0
|
||
while True:
|
||
cursor = _first_of_month(cursor) - relativedelta(days=1)
|
||
month_start = _first_of_month(cursor)
|
||
month_end = _last_of_month(cursor)
|
||
|
||
if month_end < earliest_scrobble:
|
||
logger.info("Reached earliest scrobble at %s, stopping", earliest_scrobble)
|
||
break
|
||
|
||
logger.info(
|
||
"Checking for Last.fm scrobbles in range %s to %s",
|
||
month_start.date(),
|
||
month_end.date(),
|
||
)
|
||
|
||
has = lastfm.get_last_scrobbles(
|
||
time_from=month_start, time_to=month_end, check=True
|
||
)
|
||
if not has:
|
||
continue
|
||
|
||
lfm_import = LastFmImport.objects.create(user_id=user_id)
|
||
process_lastfm_import.delay(
|
||
lfm_import.id, time_from=month_start, time_to=month_end
|
||
)
|
||
logger.info(
|
||
"%s – %s dispatched (import %s)",
|
||
month_start.date(),
|
||
month_end.date(),
|
||
lfm_import.id,
|
||
)
|
||
dispatched += 1
|
||
|
||
logger.info("Dispatched %d monthly batch(es) for user %s", dispatched, user_id)
|
||
return dispatched
|