114 lines
4.3 KiB
Python
114 lines
4.3 KiB
Python
"""
|
|
Music metadata search across multiple sources (iTunes, Deezer).
|
|
Both are free and require no API key.
|
|
"""
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
ITUNES_URL = "https://itunes.apple.com/search"
|
|
DEEZER_SEARCH_URL = "https://api.deezer.com/search"
|
|
DEEZER_ALBUM_URL = "https://api.deezer.com/album"
|
|
|
|
SOURCES = ("itunes", "deezer")
|
|
|
|
|
|
async def search_music(query: str, source: str = "itunes", limit: int = 10) -> list[dict[str, Any]]:
|
|
if source == "deezer":
|
|
return await _search_deezer(query, limit)
|
|
return await _search_itunes(query, limit)
|
|
|
|
|
|
# ── iTunes ───────────────────────────────────────────────────────────────────
|
|
|
|
async def _search_itunes(query: str, limit: int) -> list[dict[str, Any]]:
|
|
params = {
|
|
"term": query,
|
|
"media": "music",
|
|
"entity": "song",
|
|
"limit": limit,
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.get(ITUNES_URL, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
results = []
|
|
for item in data.get("results", []):
|
|
# Upgrade artwork URL to 600x600 (iTunes returns 100x100 by default)
|
|
art_url = item.get("artworkUrl100", "")
|
|
if art_url:
|
|
art_url = art_url.replace("100x100bb", "600x600bb")
|
|
|
|
results.append({
|
|
"trackId": item.get("trackId"),
|
|
"trackName": item.get("trackName", ""),
|
|
"artistName": item.get("artistName", ""),
|
|
"collectionName": item.get("collectionName", ""),
|
|
"trackNumber": item.get("trackNumber"),
|
|
"discNumber": item.get("discNumber"),
|
|
"releaseDate": item.get("releaseDate", "")[:4], # just the year
|
|
"genre": item.get("primaryGenreName", ""),
|
|
"artworkUrl": art_url,
|
|
"previewUrl": item.get("previewUrl", ""),
|
|
})
|
|
|
|
return results
|
|
|
|
|
|
# ── Deezer ───────────────────────────────────────────────────────────────────
|
|
|
|
async def _search_deezer(query: str, limit: int) -> list[dict[str, Any]]:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.get(DEEZER_SEARCH_URL, params={"q": query, "limit": limit})
|
|
response.raise_for_status()
|
|
tracks = response.json().get("data", [])
|
|
|
|
# Deezer's search results don't include genre/release date/track number —
|
|
# those live on the album, so fetch each unique album in parallel.
|
|
album_ids = {t["album"]["id"] for t in tracks if t.get("album")}
|
|
albums = await asyncio.gather(
|
|
*(_fetch_deezer_album(client, aid) for aid in album_ids),
|
|
return_exceptions=True,
|
|
)
|
|
album_map = {a["id"]: a for a in albums if isinstance(a, dict) and "id" in a}
|
|
|
|
results = []
|
|
for item in tracks:
|
|
album = item.get("album") or {}
|
|
album_info = album_map.get(album.get("id"), {})
|
|
|
|
genres = album_info.get("genres", {}).get("data", [])
|
|
release_date = album_info.get("release_date", "")
|
|
|
|
results.append({
|
|
"trackId": item.get("id"),
|
|
"trackName": item.get("title", ""),
|
|
"artistName": (item.get("artist") or {}).get("name", ""),
|
|
"collectionName": album.get("title", ""),
|
|
"trackNumber": _deezer_track_position(album_info, item.get("id")),
|
|
"discNumber": None,
|
|
"releaseDate": release_date[:4] if release_date else "",
|
|
"genre": genres[0]["name"] if genres else "",
|
|
"artworkUrl": album.get("cover_xl") or album.get("cover_big") or "",
|
|
"previewUrl": item.get("preview", ""),
|
|
})
|
|
|
|
return results
|
|
|
|
|
|
async def _fetch_deezer_album(client: httpx.AsyncClient, album_id: int) -> dict[str, Any]:
|
|
response = await client.get(f"{DEEZER_ALBUM_URL}/{album_id}")
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def _deezer_track_position(album_info: dict[str, Any], track_id: Any) -> int | None:
|
|
for track in album_info.get("tracks", {}).get("data", []):
|
|
if track.get("id") == track_id:
|
|
return track.get("track_position")
|
|
return None
|