Add Last.fm importing
This commit is contained in:
@ -1,10 +1,12 @@
|
||||
#!/usr/bin/env python3
|
||||
from typing import Optional
|
||||
import logging
|
||||
|
||||
from scrobbles.musicbrainz import (
|
||||
lookup_album_dict_from_mb,
|
||||
lookup_artist_id_from_mb,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
from music.models import Artist, Album, Track
|
||||
|
||||
@ -20,7 +22,7 @@ def get_or_create_artist(name: str) -> Artist:
|
||||
def get_or_create_album(name: str, artist: Artist) -> Album:
|
||||
album = None
|
||||
album_created = False
|
||||
albums = Album.objects.filter(name=name)
|
||||
albums = Album.objects.filter(name__iexact=name)
|
||||
if albums.count() == 1:
|
||||
album = albums.first()
|
||||
else:
|
||||
@ -62,16 +64,24 @@ def get_or_create_track(
|
||||
run_time=None,
|
||||
run_time_ticks=None,
|
||||
) -> Track:
|
||||
track, track_created = Track.objects.get_or_create(
|
||||
title=title,
|
||||
artist=artist,
|
||||
musicbrainz_id=mbid,
|
||||
)
|
||||
track = None
|
||||
if mbid:
|
||||
track = Track.objects.filter(
|
||||
musicbrainz_id=mbid,
|
||||
).first()
|
||||
if not track:
|
||||
track = Track.objects.filter(
|
||||
title=title, artist=artist, album=album
|
||||
).first()
|
||||
|
||||
if track_created:
|
||||
track.album = album
|
||||
track.run_time = run_time
|
||||
track.run_time_ticks = run_time_ticks
|
||||
track.save(update_fields=['album', 'run_time', 'run_time_ticks'])
|
||||
if not track:
|
||||
track = Track.objects.create(
|
||||
title=title,
|
||||
artist=artist,
|
||||
album=album,
|
||||
musicbrainz_id=mbid,
|
||||
run_time=run_time,
|
||||
run_time_ticks=run_time_ticks,
|
||||
)
|
||||
|
||||
return track
|
||||
|
||||
@ -0,0 +1,24 @@
|
||||
# Generated by Django 4.1.5 on 2023-02-12 22:26
|
||||
|
||||
from django.db import migrations, models
|
||||
import encrypted_field.fields
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('profiles', '0001_initial'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='userprofile',
|
||||
name='lastfm_password',
|
||||
field=encrypted_field.fields.EncryptedField(blank=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='userprofile',
|
||||
name='lastfm_username',
|
||||
field=models.CharField(blank=True, max_length=255, null=True),
|
||||
),
|
||||
]
|
||||
@ -5,7 +5,10 @@ from django.db import models
|
||||
from django_extensions.db.models import TimeStampedModel
|
||||
from profiles.constants import PRETTY_TIMEZONE_CHOICES
|
||||
|
||||
from encrypted_field import EncryptedField
|
||||
|
||||
User = get_user_model()
|
||||
BNULL = {"blank": True, "null": True}
|
||||
|
||||
|
||||
class UserProfile(TimeStampedModel):
|
||||
@ -15,6 +18,8 @@ class UserProfile(TimeStampedModel):
|
||||
timezone = models.CharField(
|
||||
max_length=255, choices=PRETTY_TIMEZONE_CHOICES
|
||||
)
|
||||
lastfm_username = models.CharField(max_length=255, **BNULL)
|
||||
lastfm_password = EncryptedField(**BNULL)
|
||||
|
||||
def __str__(self):
|
||||
return f"User profile for {self.user}"
|
||||
|
||||
@ -1,6 +1,10 @@
|
||||
from django.contrib import admin
|
||||
|
||||
from scrobbles.models import AudioScrobblerTSVImport, Scrobble
|
||||
from scrobbles.models import (
|
||||
AudioScrobblerTSVImport,
|
||||
ChartRecord,
|
||||
LastFmImport,
|
||||
Scrobble,
|
||||
)
|
||||
|
||||
|
||||
class ScrobbleInline(admin.TabularInline):
|
||||
@ -17,6 +21,38 @@ class AudioScrobblerTSVImportAdmin(admin.ModelAdmin):
|
||||
ordering = ("-created",)
|
||||
|
||||
|
||||
@admin.register(LastFmImport)
|
||||
class LastFmImportAdmin(admin.ModelAdmin):
|
||||
date_hierarchy = "created"
|
||||
list_display = ("uuid", "created", "process_count", "processed_on")
|
||||
ordering = ("-created",)
|
||||
|
||||
|
||||
@admin.register(ChartRecord)
|
||||
class ChartRecordAdmin(admin.ModelAdmin):
|
||||
date_hierarchy = "created"
|
||||
list_display = (
|
||||
"user",
|
||||
"rank",
|
||||
"year",
|
||||
"week",
|
||||
"month",
|
||||
"day",
|
||||
"media_name",
|
||||
)
|
||||
ordering = ("-created",)
|
||||
|
||||
def media_name(self, obj):
|
||||
if obj.video:
|
||||
return obj.video
|
||||
if obj.track:
|
||||
return obj.track
|
||||
if obj.podcast_episode:
|
||||
return obj.podcast_episode
|
||||
if obj.sport_event:
|
||||
return obj.sport_event
|
||||
|
||||
|
||||
@admin.register(Scrobble)
|
||||
class ScrobbleAdmin(admin.ModelAdmin):
|
||||
date_hierarchy = "timestamp"
|
||||
|
||||
143
vrobbler/apps/scrobbles/lastfm.py
Normal file
143
vrobbler/apps/scrobbles/lastfm.py
Normal file
@ -0,0 +1,143 @@
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pylast
|
||||
import pytz
|
||||
from django.conf import settings
|
||||
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"
|
||||
source_id = ""
|
||||
latest_scrobbles = self.get_last_scrobbles(time_from=last_processed)
|
||||
|
||||
for scrobble in latest_scrobbles:
|
||||
timestamp = scrobble.pop('timestamp')
|
||||
|
||||
artist = get_or_create_artist(scrobble.pop('artist'))
|
||||
album = get_or_create_album(scrobble.pop('album'), artist)
|
||||
|
||||
scrobble['artist'] = artist
|
||||
scrobble['album'] = album
|
||||
track = get_or_create_track(**scrobble)
|
||||
|
||||
new_scrobble = Scrobble(
|
||||
user=self.vrobbler_user,
|
||||
timestamp=timestamp,
|
||||
source=source,
|
||||
source_id=source_id,
|
||||
track=track,
|
||||
played_to_completion=True,
|
||||
in_progress=False,
|
||||
)
|
||||
# Vrobbler scrobbles on finish, LastFM scrobbles on start
|
||||
ten_seconds_eariler = timestamp - timedelta(seconds=15)
|
||||
ten_seconds_later = timestamp + timedelta(seconds=15)
|
||||
existing = Scrobble.objects.filter(
|
||||
created__gte=ten_seconds_eariler,
|
||||
created__lte=ten_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
|
||||
|
||||
@staticmethod
|
||||
def undo_lastfm_import(process_log, dryrun=True):
|
||||
"""Given a newline separated list of scrobbles, delete them"""
|
||||
from scrobbles.models import Scrobble
|
||||
|
||||
if not process_log:
|
||||
logger.warning("No lines in process log found to undo")
|
||||
return
|
||||
|
||||
for line in process_log.split('\n'):
|
||||
scrobble_id = line.split("\t")[0]
|
||||
scrobble = Scrobble.objects.filter(id=scrobble_id).first()
|
||||
if not scrobble:
|
||||
logger.warning(
|
||||
f"Could not find scrobble {scrobble_id} to undo"
|
||||
)
|
||||
continue
|
||||
logger.info(f"Removing scrobble {scrobble_id}")
|
||||
if not dryrun:
|
||||
scrobble.delete()
|
||||
|
||||
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"""
|
||||
scrobbles = []
|
||||
if time_from:
|
||||
time_from = int(time_from.timestamp())
|
||||
if time_to:
|
||||
time_to = int(time_to.timestamp())
|
||||
|
||||
if not time_from and not time_to:
|
||||
found_scrobbles = self.user.get_recent_tracks(limit=None)
|
||||
else:
|
||||
found_scrobbles = self.user.get_recent_tracks(
|
||||
time_from=time_from, time_to=time_to
|
||||
)
|
||||
for scrobble in found_scrobbles:
|
||||
run_time_ticks = scrobble.track.get_duration()
|
||||
run_time = run_time_ticks / 1000
|
||||
scrobbles.append(
|
||||
{
|
||||
"artist": scrobble.track.get_artist().name,
|
||||
"album": scrobble.album,
|
||||
"title": scrobble.track.title,
|
||||
"mbid": scrobble.track.get_mbid(),
|
||||
"run_time": int(run_time),
|
||||
"run_time_ticks": run_time_ticks,
|
||||
"timestamp": datetime.utcfromtimestamp(
|
||||
int(scrobble.timestamp)
|
||||
).replace(tzinfo=pytz.utc),
|
||||
}
|
||||
)
|
||||
return scrobbles
|
||||
61
vrobbler/apps/scrobbles/migrations/0018_lastfmimport.py
Normal file
61
vrobbler/apps/scrobbles/migrations/0018_lastfmimport.py
Normal file
@ -0,0 +1,61 @@
|
||||
# Generated by Django 4.1.5 on 2023-02-13 06:43
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
import django_extensions.db.fields
|
||||
import uuid
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
('scrobbles', '0017_audioscrobblertsvimport_user'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='LastFmImport',
|
||||
fields=[
|
||||
(
|
||||
'id',
|
||||
models.BigAutoField(
|
||||
auto_created=True,
|
||||
primary_key=True,
|
||||
serialize=False,
|
||||
verbose_name='ID',
|
||||
),
|
||||
),
|
||||
(
|
||||
'created',
|
||||
django_extensions.db.fields.CreationDateTimeField(
|
||||
auto_now_add=True, verbose_name='created'
|
||||
),
|
||||
),
|
||||
(
|
||||
'modified',
|
||||
django_extensions.db.fields.ModificationDateTimeField(
|
||||
auto_now=True, verbose_name='modified'
|
||||
),
|
||||
),
|
||||
('uuid', models.UUIDField(default=uuid.uuid4, editable=False)),
|
||||
('processed_on', models.DateTimeField(blank=True, null=True)),
|
||||
('process_log', models.TextField(blank=True, null=True)),
|
||||
('process_count', models.IntegerField(blank=True, null=True)),
|
||||
(
|
||||
'user',
|
||||
models.ForeignKey(
|
||||
blank=True,
|
||||
null=True,
|
||||
on_delete=django.db.models.deletion.DO_NOTHING,
|
||||
to=settings.AUTH_USER_MODEL,
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
'get_latest_by': 'modified',
|
||||
'abstract': False,
|
||||
},
|
||||
),
|
||||
]
|
||||
@ -12,6 +12,7 @@ from scrobbles.utils import check_scrobble_for_finish
|
||||
from sports.models import SportEvent
|
||||
from videos.models import Series, Video
|
||||
from vrobbler.apps.profiles.utils import now_user_timezone
|
||||
from vrobbler.apps.scrobbles.lastfm import LastFM
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
User = get_user_model()
|
||||
@ -78,6 +79,66 @@ class AudioScrobblerTSVImport(TimeStampedModel):
|
||||
undo_audioscrobbler_tsv_import(self.process_log, dryrun)
|
||||
|
||||
|
||||
class LastFmImport(TimeStampedModel):
|
||||
user = models.ForeignKey(User, on_delete=models.DO_NOTHING, **BNULL)
|
||||
uuid = models.UUIDField(editable=False, default=uuid4)
|
||||
processed_on = models.DateTimeField(**BNULL)
|
||||
process_log = models.TextField(**BNULL)
|
||||
process_count = models.IntegerField(**BNULL)
|
||||
|
||||
def __str__(self):
|
||||
return f"LastFM Import: {self.uuid}"
|
||||
|
||||
def process(self, import_all=False):
|
||||
"""Import scrobbles found on LastFM"""
|
||||
if self.processed_on:
|
||||
logger.info(f"{self} already processed on {self.processed_on}")
|
||||
return
|
||||
|
||||
last_import = None
|
||||
if not import_all:
|
||||
try:
|
||||
last_import = LastFmImport.objects.exclude(id=self.id).last()
|
||||
except:
|
||||
pass
|
||||
|
||||
if not import_all and not last_import:
|
||||
logger.warn(
|
||||
"No previous import, to import all Last.fm scrobbles, pass import_all=True"
|
||||
)
|
||||
return
|
||||
|
||||
lastfm = LastFM(self.user)
|
||||
last_processed = None
|
||||
if last_import:
|
||||
last_processed = last_import.processed_on
|
||||
|
||||
scrobbles = lastfm.import_from_lastfm(last_processed)
|
||||
self.process_log = ""
|
||||
if scrobbles:
|
||||
for count, scrobble in enumerate(scrobbles):
|
||||
scrobble_str = f"{scrobble.id}\t{scrobble.timestamp}\t{scrobble.track.title}"
|
||||
log_line = f"{scrobble_str}"
|
||||
if count > 0:
|
||||
log_line = "\n" + log_line
|
||||
self.process_log += log_line
|
||||
self.process_count = len(scrobbles)
|
||||
else:
|
||||
self.process_log = f"Created no new scrobbles"
|
||||
self.process_count = 0
|
||||
|
||||
self.processed_on = timezone.now()
|
||||
self.save(
|
||||
update_fields=['processed_on', 'process_count', 'process_log']
|
||||
)
|
||||
|
||||
def undo(self, dryrun=False):
|
||||
"""Undo import of scrobbles from LastFM"""
|
||||
LastFM.undo_lastfm_import(self.process_log, dryrun)
|
||||
self.processed_on = None
|
||||
self.save(update_fields=['processed_on'])
|
||||
|
||||
|
||||
class ChartRecord(TimeStampedModel):
|
||||
"""Sort of like a materialized view for what we could dynamically generate,
|
||||
but would kill the DB as it gets larger. Collects time-based records
|
||||
|
||||
Reference in New Issue
Block a user