250 lines
10 KiB
Python
250 lines
10 KiB
Python
import logging
|
|
import time
|
|
|
|
import musicbrainzngs
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
|
|
from music.musicbrainz import resolve_track, get_track_metadata_with_artist, extract_featured_artists
|
|
|
|
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_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
|
|
track_ids = list(missing_mbid_qs.values_list("pk", flat=True))
|
|
i = 0
|
|
|
|
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
|
|
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:
|
|
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, 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" [{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 is_va and result:
|
|
va_fixed_count += 1
|
|
|
|
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)
|
|
|
|
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
|
|
|
|
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 musicbrainz-notfound: {not_found_count}\n"
|
|
f" Run times filled from MBID: {length_fixed_count}\n"
|
|
f" Various Artists tracks resolved: {va_fixed_count}"
|
|
)
|