diff --git a/PROJECT.org b/PROJECT.org index 3c894b8..2e689f3 100644 --- a/PROJECT.org +++ b/PROJECT.org @@ -93,7 +93,7 @@ fetching and simple saving. :LOGBOOK: CLOCK: [2025-07-09 Wed 09:55]--[2025-07-09 Wed 10:15] => 0:20 :END: -* Backlog [33/49] :vrobbler:project:personal: +* Backlog [34/51] :vrobbler:project:personal: ** TODO [#C] Add sentiment parsing for Scrobbles with notes :vrobbler:project:scrobbles:sentiment: :PROPERTIES: :ID: 37781d6a-f3b0-48b2-bf98-33c2c791cf85 @@ -501,7 +501,10 @@ Also, we should create a data migration to clean up albums with matching musicbrainz_releasegroup_id fields by consolidating tracks around the first album found. -** TODO [#A] Clean up metadata on music tracks :music:tracks:metadata:musicbrainz: +** DONE [#A] Clean up metadata on music tracks :music:tracks:metadata:musicbrainz: +:PROPERTIES: +:ID: 3a007bd7-b901-43cb-892e-102d39fffc85 +:END: *** Description diff --git a/vrobbler/apps/music/management/__init__.py b/vrobbler/apps/music/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/vrobbler/apps/music/management/commands/__init__.py b/vrobbler/apps/music/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/vrobbler/apps/music/management/commands/cleanup_track_metadata.py b/vrobbler/apps/music/management/commands/cleanup_track_metadata.py new file mode 100644 index 0000000..915fca8 --- /dev/null +++ b/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}" + )