[music] Fix historical LFM imports

vrobbler/apps/scrobbles/importers/lastfm.py

@@ -1,9 +1,14 @@
+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__)
@@ -22,6 +27,8 @@ PYLAST_ERRORS = tuple(
 
 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"),
@@ -30,24 +37,28 @@ class LastFM:
                 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}")
+            raise
 
-    def import_from_lastfm(self, last_processed=None):
+    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)
+        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"),
-                enrich=True,
+                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(
@@ -92,6 +103,20 @@ class LastFM:
         )
         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"""
@@ -102,14 +127,12 @@ class LastFM:
         if time_to:
             lfm_params["time_to"] = int(time_to.timestamp())
 
-        # if not time_from and not time_to:
-        lfm_params["limit"] = None
+        lfm_params["limit"] = 1 if check else None
 
         found_scrobbles = self.user.get_recent_tracks(**lfm_params)
-        # TOOD spin this out into a celery task over certain threshold of found scrobbles?
 
-        if check and found_scrobbles:
-            return True
+        if check:
+            return bool(found_scrobbles)
 
         for scrobble in found_scrobbles:
             logger.info(f"Processing {scrobble}")
@@ -163,3 +186,112 @@ class LastFM:
                 }
             )
         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

vrobbler/apps/scrobbles/management/commands/import_lastfm_historical.py

@@ -1,5 +1,5 @@
-from django.core.management.base import BaseCommand
 from django.contrib.auth import get_user_model
+from django.core.management.base import BaseCommand
 
 User = get_user_model()
 
@@ -11,11 +11,17 @@ class Command(BaseCommand):
             type=int,
             help="User ID to import historical scrobbles for",
         )
+        parser.add_argument(
+            "--async",
+            action="store_true",
+            dest="async",
+            help="Dispatch the task to Celery instead of running inline",
+        )
 
     def handle(self, *args, **options):
-        from scrobbles.models import LastFmImport
+        from scrobbles.importers.lastfm import dispatch_historical_imports
 
-        user_id = options.get("user_id")
+        user_id = options["user_id"]
         user = User.objects.filter(id=user_id).first()
 
         if not user:
@@ -29,17 +35,15 @@ class Command(BaseCommand):
             )
             return
 
-        lfm_import = LastFmImport.objects.create(
-            user=user,
-            earliest_timestamp=None,
-            last_batch_imported=None,
-            import_in_progress=False,
-        )
-
-        lfm_import.start_historical_import()
-
-        self.stdout.write(
-            self.style.SUCCESS(
-                f"Started historical LastFM import for {user} (import ID: {lfm_import.id})"
+        if options["async"]:
+            dispatch_historical_imports.delay(user_id)
+            self.stdout.write(
+                self.style.SUCCESS(f"Dispatched historical import task for user {user}")
+            )
+        else:
+            count = dispatch_historical_imports(user_id)
+            self.stdout.write(
+                self.style.SUCCESS(
+                    f"Dispatched {count} monthly batch(es) for user {user}"
+                )
             )
-        )

vrobbler/apps/scrobbles/migrations/0086_add_import_in_progress_to_lastfmimport.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.29 on 2026-05-24 19:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("scrobbles", "0085_scale_csv_import_original_filename"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="lastfmimport",
+            name="import_in_progress",
+            field=models.BooleanField(default=False),
+        ),
+    ]

vrobbler/apps/scrobbles/models.py

@@ -10,9 +10,9 @@ from zoneinfo import ZoneInfo
 import pendulum
 import pytz
 from beers.models import Beer
+from birds.models import BirdingLocation
 from boardgames.models import BoardGame
 from books.koreader import process_koreader_sqlite_file
-from birds.models import BirdingLocation
 from books.models import Book, BookLogData, BookPageLogData, Paper
 from bricksets.models import BrickSet
 from charts.utils import build_charts
@@ -343,6 +343,8 @@ class TrailGPXImport(BaseFileImportMixin):
 
 
 class LastFmImport(BaseFileImportMixin):
+    import_in_progress = models.BooleanField(default=False)
+
     class Meta:
         verbose_name = "Last.FM Import"
 
@@ -353,7 +355,7 @@ class LastFmImport(BaseFileImportMixin):
     def get_absolute_url(self):
         return reverse("scrobbles:lastfm-import-detail", kwargs={"slug": self.uuid})
 
-    def process(self, import_all=False):
+    def process(self, import_all=False, time_from=None, time_to=None):
         """Import scrobbles found on LastFM"""
 
         if self.user.id == 1:
@@ -363,27 +365,32 @@ class LastFmImport(BaseFileImportMixin):
             logger.info(f"{self} already processed on {self.processed_finished}")
             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_finished
+
+        if time_from is not None or time_to is not None:
+            last_processed = time_from
+        else:
+            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
+
+            if last_import:
+                last_processed = last_import.processed_finished
 
         self.mark_started()
 
-        scrobbles = lastfm.import_from_lastfm(last_processed)
+        scrobbles = lastfm.import_from_lastfm(last_processed, time_to=time_to)
 
         self.record_log(scrobbles)
         self.mark_finished()
@@ -419,8 +426,8 @@ class RetroarchImport(BaseFileImportMixin):
         self.mark_started()
 
         if self.lrtl_file:
-            import tempfile
             import os
+            import tempfile
             import zipfile
 
             tmpdir = tempfile.mkdtemp()

vrobbler/apps/scrobbles/tasks.py

@@ -1,5 +1,5 @@
 import logging
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from celery import shared_task
 from charts.utils import (
@@ -118,13 +118,14 @@ def process_bgstats_import(import_id):
 
 
 @shared_task
-def process_lastfm_import(import_id):
+def process_lastfm_import(import_id, time_from=None, time_to=None):
     LastFmImport = apps.get_model("scrobbles", "LastFMImport")
     lastfm_import = LastFmImport.objects.filter(id=import_id).first()
     if not lastfm_import:
         logger.warn(f"LastFmImport not found with id {import_id}")
+        return
 
-    lastfm_import.process()
+    lastfm_import.process(time_from=time_from, time_to=time_to)
 
 
 @shared_task
@@ -277,7 +278,7 @@ def _retention_files_to_delete(remote_files, now):
     """
     import re
     from collections import defaultdict
-    from datetime import timedelta, date
+    from datetime import date, timedelta
 
     pattern = re.compile(r"vrobbler-backup-(\d{4})_(\d{2})_(\d{2})\.sql\.gz")
     parsed = []
@@ -312,12 +313,14 @@ def _run_remote_cleanup(ssh_key, ssh_host, remote_path):
 
     Returns a summary string (e.g. "pruned 3 old backup(s)") or None.
     """
-    import subprocess
     import re
+    import subprocess
 
     result = subprocess.run(
         ["ssh", "-i", ssh_key, ssh_host, "ls", "-1", remote_path],
-        capture_output=True, text=True, timeout=30,
+        capture_output=True,
+        text=True,
+        timeout=30,
     )
     if result.returncode != 0:
         logger.warning(
@@ -345,9 +348,7 @@ def _run_remote_cleanup(ssh_key, ssh_host, remote_path):
         rm_cmd.extend(f"{remote_path}/{f}" for f in batch)
         subprocess.run(rm_cmd, check=True, timeout=30)
 
-    logger.info(
-        "backup_database: pruned %d remote backup(s)", len(to_delete)
-    )
+    logger.info("backup_database: pruned %d remote backup(s)", len(to_delete))
     return f"pruned {len(to_delete)} old remote backup(s)"
 
 
@@ -355,8 +356,8 @@ def _run_remote_cleanup(ssh_key, ssh_host, remote_path):
 def backup_database():
     """pg_dump + gzip, scp to remote, retention cleanup, ntfy notification."""
     import os
-    import subprocess
     import re
+    import subprocess
     from datetime import datetime
     from pathlib import Path
 
@@ -380,9 +381,12 @@ def backup_database():
     pg_dump_cmd = [
         "pg_dump",
         "--no-blobs",
-        "-h", db.get("HOST", "localhost"),
-        "-U", db.get("USER", "postgres"),
-        "-d", db["NAME"],
+        "-h",
+        db.get("HOST", "localhost"),
+        "-U",
+        db.get("USER", "postgres"),
+        "-d",
+        db["NAME"],
     ]
 
     logger.info("backup_database: dumping %s to %s", db["NAME"], backup_path)
@@ -390,9 +394,7 @@ def backup_database():
     try:
         with open(backup_path, "wb") as f:
             dump_proc = subprocess.Popen(pg_dump_cmd, stdout=subprocess.PIPE, env=env)
-            gzip_proc = subprocess.Popen(
-                ["gzip"], stdin=dump_proc.stdout, stdout=f
-            )
+            gzip_proc = subprocess.Popen(["gzip"], stdin=dump_proc.stdout, stdout=f)
             dump_proc.stdout.close()
             gzip_proc.communicate()
 
@@ -402,9 +404,7 @@ def backup_database():
             return
 
         dump_size = backup_path.stat().st_size
-        logger.info(
-            "backup_database: dump complete (%.1f MB)", dump_size / 1_000_000
-        )
+        logger.info("backup_database: dump complete (%.1f MB)", dump_size / 1_000_000)
 
         size_mb = dump_size / 1_000_000
         locations = [str(backup_path)]
@@ -436,9 +436,7 @@ def backup_database():
                 ssh_host = f"{m.group(1)}@{m.group(2)}"
                 remote_path = m.group(3)
                 logger.info("backup_database: pruning old remote backups")
-                cleanup_summary = _run_remote_cleanup(
-                    ssh_key, ssh_host, remote_path
-                )
+                cleanup_summary = _run_remote_cleanup(ssh_key, ssh_host, remote_path)
         else:
             logger.warning(
                 "backup_database: DB_BACKUP_SSH_KEY and DB_BACKUP_SSH_DEST not set — "
@@ -498,7 +496,9 @@ def import_from_retroarch_all_users():
 @shared_task
 def import_from_webdav_all_users():
     """Import from WebDAV for all users (replaces */2 cron)."""
-    from vrobbler.apps.scrobbles.importers.webdav import import_from_webdav_for_all_users
+    from vrobbler.apps.scrobbles.importers.webdav import (
+        import_from_webdav_for_all_users,
+    )
 
     import_from_webdav_for_all_users()
 
@@ -519,7 +519,9 @@ def import_from_imap_all_users():
 @shared_task
 def import_from_lichess_all_users():
     """Import chess games from Lichess for all users (replaces */20 cron)."""
-    from vrobbler.apps.boardgames.sources.lichess import import_chess_games_for_all_users
+    from vrobbler.apps.boardgames.sources.lichess import (
+        import_chess_games_for_all_users,
+    )
 
     import_chess_games_for_all_users()
 
@@ -527,7 +529,9 @@ def import_from_lichess_all_users():
 @shared_task
 def send_notification_for_in_progress():
     """Send ntfy stop-notifications for in-progress scrobbles (replaces */3 cron)."""
-    from vrobbler.apps.scrobbles.utils import send_stop_notifications_for_in_progress_scrobbles
+    from vrobbler.apps.scrobbles.utils import (
+        send_stop_notifications_for_in_progress_scrobbles,
+    )
 
     send_stop_notifications_for_in_progress_scrobbles()