114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""MusicBrainz, for audio dramas Audible does not carry.
|
||
|
||
Often has the episode number in the title and the original release date instead
|
||
of a later reissue. One request per second and a descriptive user agent are
|
||
required, which makes it slower than Audible.
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
import re
|
||
import time
|
||
|
||
import httpx
|
||
|
||
from .base import MetadataResult
|
||
|
||
BASE = "https://musicbrainz.org/ws/2/release"
|
||
USER_AGENT = os.environ.get(
|
||
"WORDARR_MUSICBRAINZ_UA", "wordarr/0.1 ( https://github.com/wordarr )"
|
||
)
|
||
# audio drama / audiobook only, otherwise "Lady Bedfort" returns 8000 pop songs
|
||
MEDIA_FILTER = '(secondarytype:"audio drama" OR secondarytype:audiobook)'
|
||
LANGUAGES = {"german": "deu", "english": "eng"}
|
||
MIN_INTERVAL = 1.1 # seconds between requests, per their rate limit
|
||
RESULT_LIMIT = 15
|
||
|
||
# "Die drei ??? 100: Toteninsel" -> series, number, title
|
||
TITLE_RE = re.compile(r"^(?P<series>.+?)\s+(?P<num>\d{1,4})\s*[:.\-–]\s*(?P<title>.+)$")
|
||
|
||
_lock = asyncio.Lock()
|
||
_last_call = 0.0
|
||
|
||
|
||
async def _get(client: httpx.AsyncClient, query: str) -> dict:
|
||
"""One rate limited request; a 503 means we were still too fast."""
|
||
global _last_call
|
||
async with _lock:
|
||
wait = MIN_INTERVAL - (time.monotonic() - _last_call)
|
||
if wait > 0:
|
||
await asyncio.sleep(wait)
|
||
for attempt in (1, 2):
|
||
resp = await client.get(
|
||
BASE,
|
||
params={"query": query, "fmt": "json", "limit": RESULT_LIMIT},
|
||
headers={"User-Agent": USER_AGENT},
|
||
)
|
||
_last_call = time.monotonic()
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
if resp.status_code != 503 or attempt == 2:
|
||
return {}
|
||
await asyncio.sleep(2)
|
||
return {}
|
||
|
||
|
||
def _to_result(release: dict) -> MetadataResult:
|
||
artists = [a.get("name", "") for a in release.get("artist-credit") or []
|
||
if isinstance(a, dict) and a.get("name")]
|
||
artist = artists[0] if artists else ""
|
||
title = release.get("title", "")
|
||
series, volume = artist, None
|
||
match = TITLE_RE.match(title)
|
||
if match:
|
||
series = match.group("series").strip()
|
||
volume = int(match.group("num"))
|
||
title = match.group("title").strip()
|
||
language = (release.get("text-representation") or {}).get("language", "")
|
||
date = release.get("date") or ""
|
||
return MetadataResult(
|
||
media_type="audiobook",
|
||
title=title,
|
||
# for audio dramas the credited "artist" is the series, not an author
|
||
authors="",
|
||
series=series,
|
||
volume=volume,
|
||
language={v: k for k, v in LANGUAGES.items()}.get(language, ""),
|
||
external_id=f"mb:{release.get('id', '')}",
|
||
year=int(date[:4]) if date[:4].isdigit() else None,
|
||
source="musicbrainz",
|
||
)
|
||
|
||
|
||
# Lucene metacharacters, "???" in a series name would act as wildcards
|
||
LUCENE_SPECIAL = re.compile(r'[+\-&|!(){}\[\]^"~*?:\\/]+')
|
||
|
||
|
||
def _terms(query: str) -> str:
|
||
"""Words joined with AND; a quoted phrase would miss "… 100: Toteninsel"."""
|
||
words = [w for w in LUCENE_SPECIAL.sub(" ", query).split() if len(w) > 1][:8]
|
||
return " AND ".join(f"release:{w}" for w in words)
|
||
|
||
|
||
async def search(query: str, language: str = "") -> list[MetadataResult]:
|
||
words = _terms(query)
|
||
if not words:
|
||
return []
|
||
terms = [f"({words})", MEDIA_FILTER]
|
||
if language in LANGUAGES:
|
||
terms.append(f"lang:{LANGUAGES[language]}")
|
||
async with httpx.AsyncClient(timeout=25, follow_redirects=True) as client:
|
||
data = await _get(client, " AND ".join(terms))
|
||
|
||
# many pressings per drama, keep the oldest: the original release
|
||
best: dict[tuple, MetadataResult] = {}
|
||
for release in data.get("releases", []):
|
||
result = _to_result(release)
|
||
if not result.title:
|
||
continue
|
||
key = (result.title.lower(), result.series.lower(), result.volume)
|
||
current = best.get(key)
|
||
if not current or (result.year and (not current.year or result.year < current.year)):
|
||
best[key] = result
|
||
return sorted(best.values(), key=lambda r: (r.volume is None, r.volume or 0, r.title))
|