feat(search): add multi-source search across Yandex, Apple, YouTube with direct download & import

app/core/search.py

@@ -0,0 +1,156 @@
+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[:6]:
+                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
+                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}"
+                })
+
+        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
+                    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")
+                    })
+    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)
+                    tracks.append({
+                        "service": "youtube",
+                        "service_name": "YouTube",
+                        "id": entry.get("id"),
+                        "title": entry.get("title"),
+                        "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')}"
+                    })
+    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
+    }