198 lines
8.3 KiB
Python
198 lines
8.3 KiB
Python
import asyncio
|
|
import urllib.parse
|
|
from typing import List, Dict, Any, Optional
|
|
import httpx
|
|
import yt_dlp
|
|
from app.config import settings
|
|
from app.providers.yandex import YandexMusicProvider
|
|
|
|
yandex_provider = YandexMusicProvider()
|
|
|
|
def format_duration(ms: Optional[int]) -> str:
|
|
if not ms:
|
|
return "00:00"
|
|
total_sec = ms // 1000
|
|
m = total_sec // 60
|
|
s = total_sec % 60
|
|
return f"{m:02d}:{s:02d}"
|
|
|
|
async def search_yandex(query: str) -> Dict[str, List[Dict[str, Any]]]:
|
|
"""Search tracks and albums via Yandex Music."""
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(None, _search_yandex_sync, query)
|
|
|
|
def _search_yandex_sync(query: str) -> Dict[str, List[Dict[str, Any]]]:
|
|
tracks = []
|
|
albums = []
|
|
try:
|
|
client = yandex_provider.get_client()
|
|
res = client.search(query, type_="all")
|
|
|
|
if res.tracks and res.tracks.results:
|
|
for t in res.tracks.results[:8]:
|
|
if not t:
|
|
continue
|
|
artists = ", ".join(a.name for a in t.artists) if t.artists else "Unknown"
|
|
album_title = t.albums[0].title if t.albums else "Single"
|
|
year = t.albums[0].year if t.albums and t.albums[0].year else None
|
|
cover = "https://" + t.cover_uri.replace("%%", "400x400") if t.cover_uri else None
|
|
|
|
content_warning = getattr(t, "content_warning", None)
|
|
version = getattr(t, "version", None)
|
|
title_lower = (t.title or "").lower()
|
|
version_lower = (version or "").lower()
|
|
|
|
if content_warning == "explicit" or any(w in title_lower or w in version_lower for w in ["explicit", "18+", "uncensored"]):
|
|
explicit_status = "explicit"
|
|
elif content_warning == "clean" or any(w in title_lower or w in version_lower for w in ["clean", "censored", "радио", "radio edit"]):
|
|
explicit_status = "clean"
|
|
else:
|
|
explicit_status = "neutral"
|
|
|
|
tracks.append({
|
|
"service": "yandex",
|
|
"service_name": "Яндекс Музыка",
|
|
"id": str(t.id),
|
|
"title": t.title,
|
|
"artist": artists,
|
|
"album": album_title,
|
|
"year": year,
|
|
"duration": format_duration(t.duration_ms),
|
|
"cover": cover,
|
|
"url": f"https://music.yandex.ru/track/{t.id}",
|
|
"explicit_status": explicit_status,
|
|
"version": version,
|
|
"preview_url": None
|
|
})
|
|
|
|
if res.albums and res.albums.results:
|
|
for a in res.albums.results[:4]:
|
|
if not a:
|
|
continue
|
|
artists = ", ".join(art.name for art in a.artists) if a.artists else "Unknown"
|
|
cover = "https://" + a.cover_uri.replace("%%", "400x400") if a.cover_uri else None
|
|
albums.append({
|
|
"service": "yandex",
|
|
"service_name": "Яндекс Музыка",
|
|
"id": str(a.id),
|
|
"title": a.title,
|
|
"artist": artists,
|
|
"year": a.year,
|
|
"track_count": a.track_count,
|
|
"cover": cover,
|
|
"url": f"https://music.yandex.ru/album/{a.id}"
|
|
})
|
|
except Exception as e:
|
|
print(f"Yandex search error: {e}")
|
|
return {"tracks": tracks, "albums": albums}
|
|
|
|
async def search_apple(query: str) -> Dict[str, List[Dict[str, Any]]]:
|
|
"""Search tracks and albums via Apple Music / iTunes public API."""
|
|
tracks = []
|
|
albums = []
|
|
try:
|
|
encoded = urllib.parse.quote(query)
|
|
url = f"https://itunes.apple.com/search?term={encoded}&entity=song&limit=5"
|
|
async with httpx.AsyncClient(timeout=8.0) as client:
|
|
resp = await client.get(url)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
for r in data.get("results", []):
|
|
cover = r.get("artworkUrl100", "").replace("100x100bb", "400x400bb")
|
|
duration_ms = r.get("trackTimeMillis", 0)
|
|
release_date = r.get("releaseDate", "")
|
|
year = int(release_date[:4]) if release_date else None
|
|
|
|
track_explicitness = (r.get("trackExplicitness") or "").lower()
|
|
if track_explicitness == "explicit":
|
|
explicit_status = "explicit"
|
|
elif track_explicitness == "cleaned":
|
|
explicit_status = "clean"
|
|
else:
|
|
explicit_status = "neutral"
|
|
|
|
tracks.append({
|
|
"service": "apple",
|
|
"service_name": "Apple Music",
|
|
"id": str(r.get("trackId")),
|
|
"title": r.get("trackName"),
|
|
"artist": r.get("artistName"),
|
|
"album": r.get("collectionName"),
|
|
"year": year,
|
|
"duration": format_duration(duration_ms),
|
|
"cover": cover,
|
|
"url": r.get("trackViewUrl"),
|
|
"explicit_status": explicit_status,
|
|
"version": None,
|
|
"preview_url": r.get("previewUrl")
|
|
})
|
|
except Exception as e:
|
|
print(f"Apple search error: {e}")
|
|
return {"tracks": tracks, "albums": albums}
|
|
|
|
async def search_youtube(query: str) -> Dict[str, List[Dict[str, Any]]]:
|
|
"""Search tracks via YouTube Music."""
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(None, _search_youtube_sync, query)
|
|
|
|
def _search_youtube_sync(query: str) -> Dict[str, List[Dict[str, Any]]]:
|
|
tracks = []
|
|
ydl_opts = {
|
|
"extract_flat": True,
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
}
|
|
try:
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(f"ytsearch3:{query}", download=False)
|
|
if info and "entries" in info:
|
|
for entry in info["entries"]:
|
|
if not entry:
|
|
continue
|
|
sec = entry.get("duration", 0)
|
|
title_raw = entry.get("title") or ""
|
|
title_lower = title_raw.lower()
|
|
|
|
if any(w in title_lower for w in ["explicit", "uncensored", "18+"]):
|
|
explicit_status = "explicit"
|
|
elif any(w in title_lower for w in ["clean", "censored", "radio edit", "радиоверсия"]):
|
|
explicit_status = "clean"
|
|
else:
|
|
explicit_status = "neutral"
|
|
|
|
tracks.append({
|
|
"service": "youtube",
|
|
"service_name": "YouTube",
|
|
"id": entry.get("id"),
|
|
"title": title_raw,
|
|
"artist": entry.get("uploader") or entry.get("channel") or "Unknown",
|
|
"album": "YouTube Audio",
|
|
"year": None,
|
|
"duration": f"{sec // 60:02d}:{sec % 60:02d}" if sec else "00:00",
|
|
"cover": entry.get("thumbnail"),
|
|
"url": f"https://www.youtube.com/watch?v={entry.get('id')}",
|
|
"explicit_status": explicit_status,
|
|
"version": None,
|
|
"preview_url": None
|
|
})
|
|
except Exception as e:
|
|
print(f"YouTube search error: {e}")
|
|
return {"tracks": tracks, "albums": []}
|
|
|
|
async def search_all(query: str) -> Dict[str, Any]:
|
|
"""Run concurrent search across all platforms."""
|
|
yandex_task = search_yandex(query)
|
|
apple_task = search_apple(query)
|
|
youtube_task = search_youtube(query)
|
|
|
|
y_res, a_res, yt_res = await asyncio.gather(yandex_task, apple_task, youtube_task)
|
|
|
|
all_tracks = y_res["tracks"] + a_res["tracks"] + yt_res["tracks"]
|
|
all_albums = y_res["albums"] + a_res["albums"]
|
|
|
|
return {
|
|
"query": query,
|
|
"tracks": all_tracks,
|
|
"albums": all_albums
|
|
}
|