From 034c7cb4134b5cf3c3827918d43a427e0e2fe6e0 Mon Sep 17 00:00:00 2001 From: Colin Powell Date: Fri, 29 May 2026 19:47:09 -0400 Subject: [PATCH] [music] Add better track fetching --- .../commands/cleanup_track_metadata.py | 200 +++++++++--------- vrobbler/apps/music/models.py | 25 ++- vrobbler/apps/music/musicbrainz.py | 144 +++++++++++++ 3 files changed, 261 insertions(+), 108 deletions(-) diff --git a/vrobbler/apps/music/management/commands/cleanup_track_metadata.py b/vrobbler/apps/music/management/commands/cleanup_track_metadata.py index 915fca8..9fcd95a 100644 --- a/vrobbler/apps/music/management/commands/cleanup_track_metadata.py +++ b/vrobbler/apps/music/management/commands/cleanup_track_metadata.py @@ -3,8 +3,9 @@ import time import musicbrainzngs from django.core.management.base import BaseCommand +from django.db import transaction -from music.musicbrainz import get_track_metadata_with_artist +from music.musicbrainz import resolve_track, get_track_metadata_with_artist, extract_featured_artists logger = logging.getLogger(__name__) @@ -34,28 +35,6 @@ def _get_artist_from_scrobble_logs(track_id: int): 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"]) @@ -154,98 +133,117 @@ class Command(BaseCommand): not_found_count = 0 length_fixed_count = 0 va_fixed_count = 0 + track_ids = list(missing_mbid_qs.values_list("pk", flat=True)) + i = 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 + for batch_num, offset in enumerate(range(0, len(track_ids), batch_size)): + batch_pks = track_ids[offset : offset + batch_size] + with transaction.atomic(): + for track in Track.objects.filter(pk__in=batch_pks).iterator(): + i += 1 + artist_name = "" + artist_obj = track.artist + if artist_obj: + artist_name = artist_obj.name - is_va = _is_various_artists(artist_name) - result = None + is_va = _is_various_artists(artist_name) + result = None + method = "" - 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 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: + method = "provider" + self.stdout.write( + f" VA track '{track.title}' matched via provider MBID " + f"{scrobble_info['recording_mbid']} (artist: {scrobble_info['artist_name']})" + ) + else: + result, method = resolve_track( + track.title, scrobble_info["artist_name"] + ) + if result: + self.stdout.write( + f" VA track '{track.title}' matched via log artist " + f"'{scrobble_info['artist_name']}' as {result['recording_mbid']} ({method})" + ) - if not result: - result = _lookup_by_title_only(track.title) - if result: + if not result: + result, method = resolve_track(track.title) + if result: + self.stdout.write( + f" VA track '{track.title}' matched by title alone " + f"as {result['recording_mbid']} ({method})" + ) + + if not result: + result, method = resolve_track(track.title, artist_name) + + 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(int(length_ms) / 1000) + track.save(update_fields=["musicbrainz_id", "base_run_time_seconds"]) + method_tag = f"musicbrainz-{method}" if method else "musicbrainz-enriched" + track.tags.add(method_tag, "musicbrainz-enriched") + from music.models import Artist + cleaned_title, featured_names = extract_featured_artists(track.title) + for feat_name in featured_names: + artist = Artist.find_or_create(feat_name, track_name=track.title) + if artist: + track.artists.add(artist) self.stdout.write( - f" VA track '{track.title}' matched recording {result['recording_mbid']}" + f" [{i}] FOUND {track} — mbid={track.musicbrainz_id} ({method})" ) + found_count += 1 + else: + track.tags.add("musicbrainz-notfound") + self.stdout.write( + f" [{i}] NOTFOUND {track}" + ) + not_found_count += 1 - if not result and artist_name: - result = get_track_metadata_with_artist( - track.title, artist_name - ) + if is_va and result: + va_fixed_count += 1 - if not result and track.title: - result = _lookup_by_title_only(track.title) + self.stdout.write( + f" Batch {batch_num + 1}: {offset + len(batch_pks)}/{total} processed, " + f"found: {found_count}, not found: {not_found_count}" + ) + time.sleep(1) - 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 + length_ids = list(missing_length_qs.values_list("pk", flat=True)) + j = 0 + for batch_num, offset in enumerate(range(0, len(length_ids), batch_size)): + batch_pks = length_ids[offset : offset + batch_size] + with transaction.atomic(): + for track in Track.objects.filter(pk__in=batch_pks).iterator(): + j += 1 + length_ms = _lookup_length_by_mbid(track.musicbrainz_id) + if length_ms: + track.base_run_time_seconds = int(int(length_ms) / 1000) + track.tags.add("musicbrainz-enriched") + track.save(update_fields=["base_run_time_seconds"]) + self.stdout.write( + f" [{j}] LENGTH {track} — {track.base_run_time_seconds}s" + ) + length_fixed_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" Batch {batch_num + 1}: {j}/{len(length_ids)} length lookups" + ) + time.sleep(1) 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" Not found / tagged musicbrainz-notfound: {not_found_count}\n" f" Run times filled from MBID: {length_fixed_count}\n" f" Various Artists tracks resolved: {va_fixed_count}" ) diff --git a/vrobbler/apps/music/models.py b/vrobbler/apps/music/models.py index 8c90516..4b92b19 100644 --- a/vrobbler/apps/music/models.py +++ b/vrobbler/apps/music/models.py @@ -16,9 +16,11 @@ from imagekit.processors import ResizeToFit from music.allmusic import get_allmusic_slug, scrape_data_from_allmusic from music.bandcamp import get_bandcamp_slug from music.musicbrainz import ( + extract_featured_artists, get_album_metadata_with_artist, get_recording_mbid_exact, get_track_metadata_with_artist, + resolve_track, ) from music.theaudiodb import lookup_album_from_tadb, lookup_artist_from_tadb from music.utils import clean_artist_name @@ -694,6 +696,7 @@ class Track(ScrobblableMixin): if mbid and run_time_seconds: track.base_run_time_seconds = run_time_seconds track.musicbrainz_id = mbid + track.tags.add("musicbrainz-provider", "musicbrainz-enriched") else: artist_name_str = " & ".join(artist_names) if artist_names else "" logger.info( @@ -704,15 +707,23 @@ class Track(ScrobblableMixin): "track_id": track.id, }, ) - try: - mbid, length = get_recording_mbid_exact( - title, artist_name_str, album_name - ) - except Exception: + result, method = resolve_track(title, artist_name_str, album_name) + 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(int(length_ms) / 1000) + method_tag = f"musicbrainz-{method}" if method else "" + track.tags.add(method_tag, "musicbrainz-enriched") + cleaned_title, featured_names = extract_featured_artists(title) + for feat_name in featured_names: + artist = Artist.find_or_create(feat_name, track_name=title) + if artist: + track.artists.add(artist) + else: print("No musicbrainz result found, cannot enrich") + track.tags.add("musicbrainz-notfound") return track - track.base_run_time_seconds = run_time_seconds or int(length / 1000) - track.musicbrainz_id = mbid if commit: track.save() diff --git a/vrobbler/apps/music/musicbrainz.py b/vrobbler/apps/music/musicbrainz.py index 8eec8a7..a32fb9f 100644 --- a/vrobbler/apps/music/musicbrainz.py +++ b/vrobbler/apps/music/musicbrainz.py @@ -1,4 +1,5 @@ import logging +import re from datetime import datetime import musicbrainzngs @@ -6,6 +7,149 @@ from dateutil.parser import parse logger = logging.getLogger(__name__) +TRACK_SUFFIXES = [ + r" - Bonus Track", + r" - Bonus", + r" - Remix", + r" - Live", + r" - Radio Edit", + r" - Extended Mix", + r" - Instrumental", + r" - Acoustic", + r" - Mix", + r" - Edit", + r" \[Explicit\]", +] + +FEATURED_PATTERNS = [ + (r"\(feat\. (.+?)\)", re.IGNORECASE), + (r"\(ft\. (.+?)\)", re.IGNORECASE), + (r"featuring (.+)", re.IGNORECASE), + (r"feat\. (.+)", re.IGNORECASE), + (r"ft\. (.+)", re.IGNORECASE), +] + + +def strip_track_suffixes(title: str) -> str: + for suffix in TRACK_SUFFIXES: + title = re.sub(suffix, "", title, flags=re.IGNORECASE) + return title.strip() + + +def extract_featured_artists(title: str) -> tuple[str, list[str]]: + featured = [] + cleaned = title + for pattern, flags in FEATURED_PATTERNS: + m = re.search(pattern, cleaned, flags) + if m: + featured.append(m.group(1).strip()) + cleaned = re.sub(pattern, "", cleaned, flags=flags).strip() + return cleaned, featured + + +def search_recordings(title: str, artist: str = "") -> list[dict]: + kwargs: dict = {"recording": title, "limit": 10} + if artist: + kwargs["artist"] = artist + try: + return musicbrainzngs.search_recordings(**kwargs).get("recording-list", []) + except (musicbrainzngs.WebServiceError, musicbrainzngs.ResponseError) as e: + logger.warning(f"MusicBrainz search error: {e}") + return [] + + +def pick_best_recording( + recordings: list[dict], title: str, artist: str = "" +) -> tuple[dict | None, str]: + query_title = title.strip().casefold() + query_artist = artist.strip().casefold() if artist else "" + best_score: tuple[dict | None, str] = (None, "") + + for rec in recordings: + rec_title = rec.get("title", "").strip() + score = int(rec.get("ext:score", 0)) + + if rec_title.casefold() != query_title: + if score >= 85 and not best_score[0]: + best_score = (rec, "score") + continue + + if query_artist: + artist_credit = rec.get("artist-credit", []) + rec_artist = ( + artist_credit[0]["artist"]["name"].strip() + if artist_credit + else "" + ) + if rec_artist.casefold() != query_artist: + if score >= 85 and not best_score[0]: + best_score = (rec, "score") + continue + + return (rec, "exact") + + return best_score + + +def resolve_track( + title: str, artist: str = "", album: str = "" +) -> tuple[dict | None, str]: + if not title: + return (None, "") + + titles_to_try: list[tuple[str, str]] = [(title, "")] + stripped = strip_track_suffixes(title) + if stripped != title: + titles_to_try.append((stripped, "stripped-")) + + for attempt_title, prefix in titles_to_try: + if album and not prefix: + try: + mbid, length = get_recording_mbid_exact( + attempt_title, artist, album + ) + return ( + { + "recording_mbid": mbid, + "length_ms": length, + }, + "exact", + ) + except Exception: + pass + + if artist: + result = get_track_metadata_with_artist(attempt_title, artist) + if result and result.get("recording_mbid"): + return (result, f"{prefix}exact") + + recordings = search_recordings(attempt_title, artist) + result, tag = pick_best_recording(recordings, attempt_title, artist) + if result: + length_ms = result.get("length") + return ( + { + "recording_mbid": result["id"], + "length_ms": int(length_ms) if length_ms else None, + }, + f"{prefix}{tag}", + ) + + recordings = search_recordings(attempt_title) + result, tag = pick_best_recording(recordings, attempt_title) + if result: + tag = "title-only" if not prefix else f"{prefix}title-only" + length_ms = result.get("length") + return ( + { + "recording_mbid": result["id"], + "length_ms": int(length_ms) if length_ms else None, + }, + tag, + ) + + return (None, "") + musicbrainzngs.set_useragent("Vrobbler", "1.0", "help@unbl.ink")