[music] Add metadata cleanup command

vrobbler/apps/music/management/commands/cleanup_track_metadata.py

@@ -0,0 +1,251 @@
+import logging
+import time
+
+import musicbrainzngs
+from django.core.management.base import BaseCommand
+
+from music.musicbrainz import get_track_metadata_with_artist
+
+logger = logging.getLogger(__name__)
+
+VARIOUS_ARTISTS_NAMES = ["various artists", "va", "various"]
+
+
+def _is_various_artists(artist_name: str) -> bool:
+    return artist_name.strip().casefold() in VARIOUS_ARTISTS_NAMES
+
+
+def _get_artist_from_scrobble_logs(track_id: int):
+    from scrobbles.models import Scrobble
+
+    scrobbles = Scrobble.objects.filter(track_id=track_id)[:10]
+    for scrobble in scrobbles:
+        raw = scrobble.log.get("raw_data", {})
+        if not raw:
+            continue
+
+        provider_mbid = raw.get("Provider_musicbrainztrack")
+        if provider_mbid:
+            return {"artist_name": raw.get("Artist", ""), "recording_mbid": provider_mbid}
+
+        artist = raw.get("Artist", "")
+        if artist:
+            return {"artist_name": artist, "recording_mbid": None}
+    return None
+
+
+def _lookup_by_title_only(track_title: str):
+    if not track_title:
+        return None
+    try:
+        result = musicbrainzngs.search_recordings(
+            recording=track_title, limit=10
+        )
+        query_title = track_title.strip().casefold()
+        for recording in result.get("recording-list", []):
+            rec_title = recording["title"].strip()
+            if rec_title.casefold() == query_title:
+                length_ms = recording.get("length")
+                return {
+                    "recording_mbid": recording["id"],
+                    "length_ms": int(length_ms) if length_ms else None,
+                }
+        return None
+    except (musicbrainzngs.WebServiceError, musicbrainzngs.ResponseError) as e:
+        logger.warning(f"MusicBrainz error searching by title: {e}")
+        return None
+
+
+def _lookup_recording_by_mbid(mbid: str):
+    try:
+        result = musicbrainzngs.get_recording_by_id(mbid, includes=["artists"])
+        recording = result["recording"]
+        length_ms = recording.get("length")
+        artist_credit = recording.get("artist-credit", [{}])
+        primary_artist = artist_credit[0].get("artist", {}) if artist_credit else {}
+        return {
+            "recording_mbid": mbid,
+            "length_ms": int(length_ms) if length_ms else None,
+            "artist_name": primary_artist.get("name", ""),
+            "artist_mbid": primary_artist.get("id", ""),
+        }
+    except (musicbrainzngs.WebServiceError, musicbrainzngs.ResponseError) as e:
+        logger.warning(f"MusicBrainz error looking up {mbid}: {e}")
+        return None
+
+
+def _lookup_length_by_mbid(mbid: str):
+    try:
+        result = musicbrainzngs.get_recording_by_id(mbid)
+        recording = result["recording"]
+        length_ms = recording.get("length")
+        if length_ms:
+            return int(length_ms)
+        return None
+    except (musicbrainzngs.WebServiceError, musicbrainzngs.ResponseError) as e:
+        logger.warning(f"MusicBrainz error looking up {mbid}: {e}")
+        return None
+
+
+class Command(BaseCommand):
+    help = "Clean up metadata on music tracks from MusicBrainz"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--commit",
+            action="store_true",
+            help="Commit changes to the database",
+        )
+        parser.add_argument(
+            "--batch-size",
+            type=int,
+            default=100,
+            help="Number of tracks to process per batch (default: 100)",
+        )
+        parser.add_argument(
+            "--only-missing-mbid",
+            action="store_true",
+            help="Only process tracks missing musicbrainz_id",
+        )
+        parser.add_argument(
+            "--only-missing-length",
+            action="store_true",
+            help="Only process tracks missing base_run_time_seconds",
+        )
+
+    def handle(self, *args, **options):
+        from music.models import Track
+
+        commit = options["commit"]
+        batch_size = options["batch_size"]
+
+        if not commit:
+            self.stdout.write("Dry run — no changes will be saved. Use --commit to apply.")
+
+        musicbrainzngs.set_useragent("vrobbler", "0.3.0")
+
+        qs = Track.objects.all()
+
+        if options["only_missing_mbid"]:
+            qs = qs.filter(musicbrainz_id__isnull=True)
+        if options["only_missing_length"]:
+            qs = qs.filter(base_run_time_seconds__isnull=True)
+
+        total = qs.count()
+        self.stdout.write(f"Processing {total} tracks")
+
+        missing_mbid_qs = qs.filter(musicbrainz_id__isnull=True)
+        missing_length_qs = qs.filter(
+            musicbrainz_id__isnull=False, base_run_time_seconds__isnull=True
+        )
+
+        self.stdout.write(
+            f"  {missing_mbid_qs.count()} tracks without musicbrainz_id"
+        )
+        self.stdout.write(
+            f"  {missing_length_qs.count()} tracks with mbid but no run time"
+        )
+
+        if not commit:
+            self.stdout.write("\nSkipping API lookups in dry-run mode. Use --commit to run against MusicBrainz.")
+            return
+
+        found_count = 0
+        not_found_count = 0
+        length_fixed_count = 0
+        va_fixed_count = 0
+
+        for i, track in enumerate(missing_mbid_qs.iterator(chunk_size=batch_size)):
+            artist_name = ""
+            artist_obj = track.artist
+            if artist_obj:
+                artist_name = artist_obj.name
+
+            is_va = _is_various_artists(artist_name)
+            result = None
+
+            if is_va:
+                scrobble_info = _get_artist_from_scrobble_logs(track.id)
+                if scrobble_info:
+                    if scrobble_info["recording_mbid"]:
+                        result = _lookup_recording_by_mbid(
+                            scrobble_info["recording_mbid"]
+                        )
+                        if result:
+                            self.stdout.write(
+                                f"  VA track '{track.title}' matched via provider MBID "
+                                f"{scrobble_info['recording_mbid']} (artist: {scrobble_info['artist_name']})"
+                            )
+                    else:
+                        real_artist = scrobble_info["artist_name"]
+                        result = get_track_metadata_with_artist(
+                            track.title, real_artist
+                        )
+                        if result:
+                            self.stdout.write(
+                                f"  VA track '{track.title}' matched via log artist "
+                                f"'{real_artist}' as {result['recording_mbid']}"
+                            )
+
+                if not result:
+                    result = _lookup_by_title_only(track.title)
+                    if result:
+                        self.stdout.write(
+                            f"  VA track '{track.title}' matched recording {result['recording_mbid']}"
+                        )
+
+            if not result and artist_name:
+                result = get_track_metadata_with_artist(
+                    track.title, artist_name
+                )
+
+            if not result and track.title:
+                result = _lookup_by_title_only(track.title)
+
+            if result and result.get("recording_mbid"):
+                track.musicbrainz_id = result["recording_mbid"]
+                length_ms = result.get("length_ms")
+                if length_ms and not track.base_run_time_seconds:
+                    track.base_run_time_seconds = int(length_ms / 1000)
+                track.save(update_fields=["musicbrainz_id", "base_run_time_seconds"])
+                found_count += 1
+            else:
+                track.tags.add("not-in-musicbrainz")
+                not_found_count += 1
+
+            if is_va and result:
+                va_fixed_count += 1
+
+            if (i + 1) % batch_size == 0:
+                self.stdout.write(
+                    f"  Progress: {i + 1}/{total} processed, "
+                    f"found: {found_count}, not found: {not_found_count}"
+                )
+                time.sleep(1)
+
+        for i, track in enumerate(missing_length_qs.iterator(chunk_size=batch_size)):
+            length_ms = _lookup_length_by_mbid(track.musicbrainz_id)
+            if length_ms:
+                track.base_run_time_seconds = int(length_ms / 1000)
+                track.save(update_fields=["base_run_time_seconds"])
+                length_fixed_count += 1
+
+            if (i + 1) % batch_size == 0:
+                self.stdout.write(
+                    f"  Length lookup progress: {i + 1}/{missing_length_qs.count()}"
+                )
+                time.sleep(1)
+
+        remaining = Track.objects.filter(
+            musicbrainz_id__isnull=True,
+        ).exclude(tags__name__in=["not-in-musicbrainz"])
+        for track in remaining:
+            track.tags.add("missing-metadata")
+
+        self.stdout.write(
+            f"\nResults (commit={commit}):\n"
+            f"  Found on MusicBrainz: {found_count}\n"
+            f"  Not found / tagged not-in-musicbrainz: {not_found_count}\n"
+            f"  Run times filled from MBID: {length_fixed_count}\n"
+            f"  Various Artists tracks resolved: {va_fixed_count}"
+        )