[music] Add better track fetching

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)):
+        for batch_num, offset in enumerate(range(0, len(track_ids), batch_size)):
-            artist_name = ""
+            batch_pks = track_ids[offset : offset + batch_size]
-            artist_obj = track.artist
+            with transaction.atomic():
-            if artist_obj:
+                for track in Track.objects.filter(pk__in=batch_pks).iterator():
-                artist_name = artist_obj.name
+                    i += 1
+                    artist_name = ""
+                    artist_obj = track.artist
+                    if artist_obj:
+                        artist_name = artist_obj.name
 
-            is_va = _is_various_artists(artist_name)
+                    is_va = _is_various_artists(artist_name)
-            result = None
+                    result = None
+                    method = ""
 
-            if is_va:
+                    if is_va:
-                scrobble_info = _get_artist_from_scrobble_logs(track.id)
+                        scrobble_info = _get_artist_from_scrobble_logs(track.id)
-                if scrobble_info:
+                        if scrobble_info:
-                    if scrobble_info["recording_mbid"]:
+                            if scrobble_info["recording_mbid"]:
-                        result = _lookup_recording_by_mbid(
+                                result = _lookup_recording_by_mbid(
-                            scrobble_info["recording_mbid"]
+                                    scrobble_info["recording_mbid"]
-                        )
+                                )
-                        if result:
+                                if result:
-                            self.stdout.write(
+                                    method = "provider"
-                                f"  VA track '{track.title}' matched via provider MBID "
+                                    self.stdout.write(
-                                f"{scrobble_info['recording_mbid']} (artist: {scrobble_info['artist_name']})"
+                                        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"]
+                            else:
-                        result = get_track_metadata_with_artist(
+                                result, method = resolve_track(
-                            track.title, real_artist
+                                    track.title, scrobble_info["artist_name"]
-                        )
+                                )
-                        if result:
+                                if result:
-                            self.stdout.write(
+                                    self.stdout.write(
-                                f"  VA track '{track.title}' matched via log artist "
+                                        f"  VA track '{track.title}' matched via log artist "
-                                f"'{real_artist}' as {result['recording_mbid']}"
+                                        f"'{scrobble_info['artist_name']}' as {result['recording_mbid']} ({method})"
-                            )
+                                    )
 
-                if not result:
+                        if not result:
-                    result = _lookup_by_title_only(track.title)
+                            result, method = resolve_track(track.title)
-                    if result:
+                            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:
+                    if is_va and result:
-                result = get_track_metadata_with_artist(
+                        va_fixed_count += 1
-                    track.title, artist_name
-                )
 
-            if not result and track.title:
+            self.stdout.write(
-                result = _lookup_by_title_only(track.title)
+                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"):
+        length_ids = list(missing_length_qs.values_list("pk", flat=True))
-                track.musicbrainz_id = result["recording_mbid"]
+        j = 0
-                length_ms = result.get("length_ms")
+        for batch_num, offset in enumerate(range(0, len(length_ids), batch_size)):
-                if length_ms and not track.base_run_time_seconds:
+            batch_pks = length_ids[offset : offset + batch_size]
-                    track.base_run_time_seconds = int(length_ms / 1000)
+            with transaction.atomic():
-                track.save(update_fields=["musicbrainz_id", "base_run_time_seconds"])
+                for track in Track.objects.filter(pk__in=batch_pks).iterator():
-                found_count += 1
+                    j += 1
-            else:
+                    length_ms = _lookup_length_by_mbid(track.musicbrainz_id)
-                track.tags.add("not-in-musicbrainz")
+                    if length_ms:
-                not_found_count += 1
+                        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:
+            self.stdout.write(
-                va_fixed_count += 1
+                f"  Batch {batch_num + 1}: {j}/{len(length_ids)} length lookups"
-
+            )
-            if (i + 1) % batch_size == 0:
+            time.sleep(1)
-                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"  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}"
         )

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:
+                result, method = resolve_track(title, artist_name_str, album_name)
-                    mbid, length = get_recording_mbid_exact(
+                if result and result.get("recording_mbid"):
-                        title, artist_name_str, album_name
+                    track.musicbrainz_id = result["recording_mbid"]
-                    )
+                    length_ms = result.get("length_ms")
-                except Exception:
+                    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()
 

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")