618 lines
20 KiB
Python
618 lines
20 KiB
Python
import logging
|
|
import re
|
|
from datetime import datetime
|
|
|
|
import musicbrainzngs
|
|
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")
|
|
|
|
|
|
def lookup_album_from_mb(musicbrainz_id: str) -> dict:
|
|
release_dict = {}
|
|
|
|
release_data = musicbrainzngs.get_release_by_id(
|
|
musicbrainz_id,
|
|
includes=["artists", "release-groups", "recordings"],
|
|
).get("release")
|
|
|
|
if not release_data:
|
|
return release_dict
|
|
|
|
primary_artist = release_data.get("artist-credit")[0]
|
|
release_dict = {
|
|
"artist": {
|
|
"name": primary_artist.get("name"),
|
|
"musicbrainz_id": primary_artist.get("id"),
|
|
},
|
|
"album": {
|
|
"name": release_data.get("title"),
|
|
"musicbrainz_id": musicbrainz_id,
|
|
"musicbrainz_releasegroup_id": release_data.get("release-group").get("id"),
|
|
"musicbrainz_albumaritist_id": primary_artist.get("id"),
|
|
"year": release_data.get("year")[0:4],
|
|
},
|
|
}
|
|
|
|
release_dict["tracks"] = []
|
|
for track in release_data.get("medium-list")[0]["track-list"]:
|
|
recording = track["recording"]
|
|
release_dict["tracks"].append(
|
|
{
|
|
"title": recording["title"],
|
|
"musicbrainz_id": recording["id"],
|
|
"run_time": track["length"] / 1000,
|
|
}
|
|
)
|
|
|
|
return release_dict
|
|
|
|
|
|
def lookup_album_dict_from_mb(release_name: str, artist_name: str) -> dict:
|
|
|
|
top_result = {}
|
|
|
|
try:
|
|
top_result = musicbrainzngs.search_releases(release_name, artist=artist_name)[
|
|
"release-list"
|
|
][0]
|
|
except IndexError:
|
|
logger.info(f"No release found on MB for {artist_name} and {release_name}")
|
|
|
|
score = int(top_result.get("ext:score", 0))
|
|
if score < 85:
|
|
logger.debug(
|
|
"Album lookup score below 85 threshold",
|
|
extra={"result": top_result},
|
|
)
|
|
return {}
|
|
year = None
|
|
if top_result.get("date"):
|
|
year = parse(top_result["date"]).year
|
|
|
|
return {
|
|
"year": year,
|
|
"title": top_result["title"],
|
|
"mb_id": top_result["id"],
|
|
"mb_group_id": top_result["release-group"]["id"],
|
|
}
|
|
|
|
|
|
def lookup_artist_from_mb(artist_name: str) -> dict:
|
|
|
|
try:
|
|
top_result = musicbrainzngs.search_artists(artist=artist_name)["artist-list"][0]
|
|
except IndexError:
|
|
return {}
|
|
score = int(top_result.get("ext:score"))
|
|
if score < 85:
|
|
logger.debug(
|
|
"Artist lookup score below 85 threshold",
|
|
extra={"result": top_result},
|
|
)
|
|
return {}
|
|
|
|
return top_result
|
|
|
|
|
|
def lookup_track_from_mb(
|
|
track_name: str, artist_mb_id: str, album_mb_id: str = ""
|
|
) -> dict:
|
|
logger.info(
|
|
"[lookup_track_from_mb] called",
|
|
extra={
|
|
"track_name": track_name,
|
|
"artist_mb_id": artist_mb_id,
|
|
"album_mb_id": album_mb_id,
|
|
},
|
|
)
|
|
|
|
try:
|
|
results = musicbrainzngs.search_recordings(
|
|
query=track_name, artist=artist_mb_id, release=album_mb_id
|
|
)["recording-list"]
|
|
logger.info(
|
|
"[lookup_track_from_mb] musicbrainz recordings search results",
|
|
extra={"results": results},
|
|
)
|
|
top_result = results[0]
|
|
except IndexError:
|
|
logger.error("[lookup_track_from_mb] no results found")
|
|
return {}
|
|
|
|
score = int(top_result.get("ext:score"))
|
|
if score < 85:
|
|
logger.info(
|
|
"[lookup_track_from_mb] no results above 85% certainty ",
|
|
extra={"results": results},
|
|
)
|
|
return {}
|
|
|
|
return top_result
|
|
|
|
|
|
def get_album_metadata(album_name, artist_name, strict=True) -> dict:
|
|
"""
|
|
Get detailed metadata for an album from MusicBrainz.
|
|
|
|
:param album_name: Name of the album
|
|
:param artist_name: Name of the artist
|
|
:param strict: If True, only exact matches on album and artist (case-insensitive)
|
|
:return: dict with album metadata, or None if not found
|
|
"""
|
|
try:
|
|
result = musicbrainzngs.search_releases(
|
|
release=album_name, artist=artist_name, limit=5
|
|
)
|
|
|
|
for release in result.get("release-list", []):
|
|
title = release["title"]
|
|
primary_artist = release["artist-credit"][0]["artist"]["name"]
|
|
|
|
title_match = title.lower() == album_name.lower()
|
|
artist_match = primary_artist.lower() == artist_name.lower()
|
|
|
|
if not strict or (title_match and artist_match):
|
|
all_artists = [
|
|
ac["artist"]["name"]
|
|
for ac in release["artist-credit"]
|
|
if isinstance(ac, dict) and "artist" in ac
|
|
]
|
|
|
|
return {
|
|
"album_title": title,
|
|
"primary_artist": primary_artist,
|
|
"all_artists": all_artists,
|
|
"mbid": release["id"],
|
|
"release_date": release.get(
|
|
"date"
|
|
), # May be partial (e.g., just year)
|
|
"release_group_mbid": release["release-group"]["id"],
|
|
}
|
|
|
|
return {}
|
|
|
|
except musicbrainzngs.WebServiceError as e:
|
|
print("MusicBrainz error:", e)
|
|
return {}
|
|
|
|
|
|
def get_recording_mbid_exact(
|
|
track_title: str, artist_name: str, album_name: str
|
|
) -> tuple[str, int]:
|
|
try:
|
|
result = musicbrainzngs.search_releases(
|
|
artist=artist_name, release=album_name, limit=1
|
|
)
|
|
releases = result.get("release-list", [])
|
|
if not releases:
|
|
raise Exception("No releases found")
|
|
|
|
release_id = releases[0]["id"]
|
|
|
|
release_data = musicbrainzngs.get_release_by_id(
|
|
release_id, includes=["recordings"]
|
|
)
|
|
tracks = release_data["release"]["medium-list"][0]["track-list"]
|
|
|
|
for track in tracks:
|
|
if track["recording"]["title"].lower() == track_title.lower():
|
|
return track["recording"]["id"], int(track["recording"]["length"])
|
|
|
|
raise Exception("No recording found")
|
|
except musicbrainzngs.WebServiceError as e:
|
|
print(f"MusicBrainz error: {e}")
|
|
raise Exception(e)
|
|
|
|
|
|
def get_artist_metadata_extended(artist_name, strict=True):
|
|
"""
|
|
Fetch artist metadata including MBID, name, origin, tags, and description.
|
|
|
|
:param artist_name: The artist's name
|
|
:param strict: If True, only return exact name match
|
|
:return: dict with metadata, or None if not found
|
|
"""
|
|
try:
|
|
# Step 1: Search for artist
|
|
search_results = musicbrainzngs.search_artists(artist=artist_name, limit=5)
|
|
for artist in search_results.get("artist-list", []):
|
|
if not strict or artist["name"].lower() == artist_name.lower():
|
|
mbid = artist["id"]
|
|
|
|
# Step 2: Get detailed info about the artist
|
|
details = musicbrainzngs.get_artist_by_id(
|
|
mbid, includes=["tags", "url-rels"]
|
|
)["artist"]
|
|
|
|
begin_date = details.get("life-span", {}).get("begin")
|
|
area = details.get("area", {}).get("name")
|
|
disambiguation = details.get("disambiguation")
|
|
tags = [t["name"] for t in details.get("tag-list", [])]
|
|
|
|
# Step 3: Try to find a Wikipedia or Wikidata link
|
|
description_url = None
|
|
for rel in details.get("url-relation-list", []):
|
|
if rel["type"] == "wikipedia":
|
|
description_url = rel["target"]
|
|
break
|
|
elif rel["type"] == "wikidata":
|
|
description_url = rel["target"]
|
|
|
|
return {
|
|
"mbid": mbid,
|
|
"name": details["name"],
|
|
"disambiguation": disambiguation,
|
|
"begin_date": begin_date,
|
|
"area": area,
|
|
"tags": tags,
|
|
"description_url": description_url, # user can fetch summary if needed
|
|
}
|
|
|
|
return None
|
|
except musicbrainzngs.WebServiceError as e:
|
|
print("MusicBrainz error:", e)
|
|
return None
|
|
|
|
|
|
def get_artist_metadata_brief(artist_id):
|
|
"""Fetch basic artist metadata by MBID."""
|
|
try:
|
|
details = musicbrainzngs.get_artist_by_id(
|
|
artist_id, includes=["tags", "aliases", "url-rels"]
|
|
)["artist"]
|
|
|
|
begin_date = details.get("life-span", {}).get("begin")
|
|
area = details.get("area", {}).get("name")
|
|
disambiguation = details.get("disambiguation")
|
|
tags = [t["name"] for t in details.get("tag-list", [])]
|
|
|
|
description_url = None
|
|
for rel in details.get("url-relation-list", []):
|
|
if rel["type"] == "wikipedia":
|
|
description_url = rel["target"]
|
|
break
|
|
elif rel["type"] == "wikidata" and not description_url:
|
|
description_url = rel["target"]
|
|
|
|
return {
|
|
"mbid": artist_id,
|
|
"name": details["name"],
|
|
"disambiguation": disambiguation,
|
|
"begin_date": begin_date,
|
|
"area": area,
|
|
"tags": tags,
|
|
"description_url": description_url,
|
|
}
|
|
|
|
except musicbrainzngs.WebServiceError as e:
|
|
print("MusicBrainz error (artist lookup):", e)
|
|
return None
|
|
|
|
|
|
def parse_date(date_str):
|
|
"""Parse MusicBrainz date format into sortable datetime object."""
|
|
if not date_str:
|
|
return None
|
|
for fmt in ("%Y-%m-%d", "%Y-%m", "%Y"):
|
|
try:
|
|
return datetime.strptime(date_str, fmt)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def get_album_metadata_with_artist(album_name, artist_name, strict=True):
|
|
"""
|
|
Get metadata for the earliest release of an album and its primary artist.
|
|
|
|
:param album_name: Album title
|
|
:param artist_name: Name of the artist
|
|
:param strict: If True, enforce exact match for album and artist
|
|
:return: dict with album and primary artist metadata
|
|
"""
|
|
try:
|
|
result = musicbrainzngs.search_releases(
|
|
release=album_name, artist=artist_name, limit=100
|
|
)
|
|
|
|
query_album = album_name.strip().casefold()
|
|
query_artist = artist_name.strip().casefold()
|
|
|
|
valid_releases = []
|
|
for release in result.get("release-list", []):
|
|
release_title = release["title"].strip()
|
|
primary_artist = release["artist-credit"][0]["artist"]
|
|
artist_name_actual = primary_artist["name"].strip()
|
|
|
|
if strict:
|
|
if release_title.casefold() != query_album:
|
|
continue
|
|
if artist_name_actual.casefold() != query_artist:
|
|
continue
|
|
|
|
release_date = parse_date(release.get("date"))
|
|
valid_releases.append((release, release_date))
|
|
|
|
if not valid_releases:
|
|
return None
|
|
|
|
# Sort releases by earliest release date
|
|
valid_releases.sort(key=lambda x: x[1] or datetime.max)
|
|
release, _ = valid_releases[0]
|
|
|
|
primary_artist = release["artist-credit"][0]["artist"]
|
|
all_artists = [
|
|
ac["artist"]["name"] for ac in release["artist-credit"] if "artist" in ac
|
|
]
|
|
|
|
artist_metadata = get_artist_metadata_brief(primary_artist["id"])
|
|
|
|
return {
|
|
"album_title": release["title"],
|
|
"primary_artist_name": primary_artist["name"],
|
|
"all_artists": all_artists,
|
|
"mbid": release["id"],
|
|
"release_group_mbid": release["release-group"]["id"],
|
|
"release_date": release.get("date"),
|
|
"primary_artist": artist_metadata,
|
|
}
|
|
|
|
except musicbrainzngs.WebServiceError as e:
|
|
print("MusicBrainz error (album lookup):", e)
|
|
return None
|
|
|
|
|
|
def get_artist_metadata_brief(artist_id):
|
|
try:
|
|
details = musicbrainzngs.get_artist_by_id(
|
|
artist_id, includes=["tags", "aliases", "url-rels"]
|
|
)["artist"]
|
|
|
|
begin_date = details.get("life-span", {}).get("begin")
|
|
area = details.get("area", {}).get("name")
|
|
disambiguation = details.get("disambiguation")
|
|
tags = [t["name"] for t in details.get("tag-list", [])]
|
|
|
|
description_url = None
|
|
for rel in details.get("url-relation-list", []):
|
|
if rel["type"] == "wikipedia":
|
|
description_url = rel["target"]
|
|
break
|
|
elif rel["type"] == "wikidata" and not description_url:
|
|
description_url = rel["target"]
|
|
|
|
return {
|
|
"mbid": artist_id,
|
|
"name": details["name"],
|
|
"disambiguation": disambiguation,
|
|
"begin_date": begin_date,
|
|
"area": area,
|
|
"tags": tags,
|
|
"description_url": description_url,
|
|
}
|
|
|
|
except musicbrainzngs.WebServiceError as e:
|
|
print("MusicBrainz error (artist lookup):", e)
|
|
return None
|
|
|
|
|
|
def get_track_metadata_with_artist(track_title, artist_name, strict=True):
|
|
"""
|
|
Get metadata for the earliest-known recording of a track, including artist info.
|
|
|
|
:param track_title: Track title
|
|
:param artist_name: Artist name
|
|
:param strict: If True, match exactly (case-insensitive)
|
|
:return: dict with track + release + artist metadata
|
|
"""
|
|
try:
|
|
result = musicbrainzngs.search_recordings(
|
|
recording=track_title, artist=artist_name, limit=100
|
|
)
|
|
|
|
query_track = track_title.strip().casefold()
|
|
query_artist = artist_name.strip().casefold()
|
|
|
|
valid_candidates = []
|
|
|
|
for recording in result.get("recording-list", []):
|
|
rec_title = recording["title"].strip()
|
|
artist_credit = recording["artist-credit"][0]["artist"]
|
|
artist_name_actual = artist_credit["name"].strip()
|
|
|
|
if strict:
|
|
if rec_title.casefold() != query_track:
|
|
continue
|
|
if artist_name_actual.casefold() != query_artist:
|
|
continue
|
|
|
|
if "release-list" not in recording:
|
|
continue
|
|
|
|
for release in recording["release-list"]:
|
|
release_date = parse_date(release.get("date"))
|
|
if release_date:
|
|
valid_candidates.append((recording["id"], release, release_date))
|
|
|
|
if not valid_candidates:
|
|
return None
|
|
|
|
# Pick the earliest release
|
|
valid_candidates.sort(key=lambda x: x[2])
|
|
recording_id, release, _ = valid_candidates[0]
|
|
|
|
# Fetch full recording info
|
|
full_recording = musicbrainzngs.get_recording_by_id(
|
|
recording_id, includes=["artists", "releases"]
|
|
)["recording"]
|
|
|
|
primary_artist = full_recording["artist-credit"][0]["artist"]
|
|
all_artists = [
|
|
ac["artist"]["name"]
|
|
for ac in full_recording["artist-credit"]
|
|
if "artist" in ac
|
|
]
|
|
artist_metadata = get_artist_metadata_brief(primary_artist["id"])
|
|
|
|
return {
|
|
"track_title": full_recording["title"],
|
|
"length_ms": full_recording.get("length"),
|
|
"recording_mbid": recording_id,
|
|
"release_title": release["title"],
|
|
"release_date": release.get("date"),
|
|
"release_group_mbid": release["release-group"]["id"],
|
|
"primary_artist_name": primary_artist["name"],
|
|
"all_artists": all_artists,
|
|
"primary_artist": artist_metadata,
|
|
}
|
|
|
|
except musicbrainzngs.WebServiceError as e:
|
|
print("MusicBrainz error (track lookup):", e)
|
|
return None
|