From aa6c605d14a7467b7cca53648bc30935a29b123c Mon Sep 17 00:00:00 2001 From: Antigravity Date: Mon, 12 Oct 2026 14:23:50 +0300 Subject: [PATCH] feat(search): add SoundCloud source and cross-platform relevance ranking --- app/core/search.py | 261 ++++++++++++++++++++++++++++--------- app/providers/universal.py | 12 +- app/static/style.css | 2 + app/templates/index.html | 2 +- 4 files changed, 214 insertions(+), 63 deletions(-) diff --git a/app/core/search.py b/app/core/search.py index b5eb0a3..9886b01 100644 --- a/app/core/search.py +++ b/app/core/search.py @@ -1,4 +1,6 @@ import asyncio +import re +import difflib import urllib.parse from typing import List, Dict, Any, Optional import httpx @@ -11,11 +13,63 @@ yandex_provider = YandexMusicProvider() def format_duration(ms: Optional[int]) -> str: if not ms: return "00:00" - total_sec = ms // 1000 + total_sec = int(ms) // 1000 m = total_sec // 60 s = total_sec % 60 return f"{m:02d}:{s:02d}" +def normalize_text(s: str) -> str: + if not s: + return "" + s = s.lower() + s = re.sub(r"[^\w\s]", " ", s) + return " ".join(s.split()) + +def calculate_track_relevance(query: str, item: Dict[str, Any]) -> float: + """Calculate match score between search query and track metadata.""" + q_norm = normalize_text(query) + q_words = set(q_norm.split()) + if not q_words: + return 0.0 + + art_norm = normalize_text(item.get("artist", "")) + tit_norm = normalize_text(item.get("title", "")) + full_norm = f"{art_norm} {tit_norm}" + full_words = set(full_norm.split()) + + # 1. Overlap of query words (0..100) + overlap = len(q_words & full_words) / len(q_words) + score = overlap * 100.0 + + # 2. Sequence similarity ratio (0..50) + seq_ratio = difflib.SequenceMatcher(None, q_norm, full_norm).ratio() + score += seq_ratio * 50.0 + + # 3. Exact query substring in title or combined string + if q_norm in full_norm or full_norm in q_norm: + score += 30.0 + + # 4. Handle "Artist - Title" query pattern + if " - " in query or " — " in query: + parts = re.split(r"\s+[-—]\s+", query, maxsplit=1) + if len(parts) == 2: + q_art = normalize_text(parts[0]) + q_tit = normalize_text(parts[1]) + if q_art and (q_art in art_norm or art_norm in q_art): + score += 35.0 + if q_tit and (q_tit in tit_norm or tit_norm in q_tit): + score += 45.0 + + # 5. Exact title match bonus + if tit_norm and (tit_norm == q_norm or (len(q_words) > 1 and q_norm.endswith(tit_norm))): + score += 25.0 + + # Slight penalty for overly long titles (e.g. video descriptions) + if len(full_norm) > len(q_norm) * 3: + score -= 10.0 + + return max(0.0, score) + async def search_yandex(query: str) -> Dict[str, List[Dict[str, Any]]]: """Search tracks and albums via Yandex Music.""" loop = asyncio.get_event_loop() @@ -86,6 +140,124 @@ def _search_yandex_sync(query: str) -> Dict[str, List[Dict[str, Any]]]: print(f"Yandex search error: {e}") return {"tracks": tracks, "albums": albums} +async def search_soundcloud(query: str) -> Dict[str, List[Dict[str, Any]]]: + """Search tracks via SoundCloud.""" + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, _search_soundcloud_sync, query) + +def _search_soundcloud_sync(query: str) -> Dict[str, List[Dict[str, Any]]]: + tracks = [] + encoded_query = urllib.parse.quote(query) + ydl_opts = { + "extract_flat": True, + "quiet": True, + "no_warnings": True, + "ignoreerrors": True, + } + try: + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(f"scsearch5:{encoded_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 = urllib.parse.unquote(entry.get("title") or "") + uploader_raw = urllib.parse.unquote(entry.get("uploader") or entry.get("channel") or "Unknown") + 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" + + artist = uploader_raw + title = title_raw + if " - " in title_raw: + parts = title_raw.split(" - ", 1) + artist = parts[0].strip() + title = parts[1].strip() + + url = entry.get("webpage_url") or entry.get("url") or "" + tracks.append({ + "service": "soundcloud", + "service_name": "SoundCloud", + "id": str(entry.get("id")), + "title": title, + "artist": artist, + "album": "SoundCloud", + "year": None, + "duration": f"{int(sec) // 60:02d}:{int(sec) % 60:02d}" if sec else "00:00", + "cover": entry.get("thumbnail"), + "url": url, + "explicit_status": explicit_status, + "version": None, + "preview_url": None + }) + except Exception as e: + print(f"SoundCloud search error: {e}") + return {"tracks": tracks, "albums": []} + +async def search_youtube(query: str) -> Dict[str, List[Dict[str, Any]]]: + """Search tracks via YouTube Music / YouTube.""" + 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, + "ignoreerrors": True, + } + try: + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(f"ytsearch5:{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 "" + uploader_raw = entry.get("uploader") or entry.get("channel") or "Unknown" + 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" + + artist = uploader_raw + title = title_raw + if " - " in title_raw: + parts = title_raw.split(" - ", 1) + artist = parts[0].strip() + title = parts[1].strip() + + tracks.append({ + "service": "youtube", + "service_name": "YouTube", + "id": str(entry.get("id")), + "title": title, + "artist": artist, + "album": "YouTube Audio", + "year": None, + "duration": f"{int(sec) // 60:02d}:{int(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_apple(query: str) -> Dict[str, List[Dict[str, Any]]]: """Search tracks and albums via Apple Music / iTunes public API.""" tracks = [] @@ -123,72 +295,45 @@ async def search_apple(query: str) -> Dict[str, List[Dict[str, Any]]]: "cover": cover, "url": r.get("trackViewUrl"), "explicit_status": explicit_status, - "version": None, + "version": "30с превью", "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) + """Run concurrent search across all platforms with relevance ranking.""" + y_task = search_yandex(query) + sc_task = search_soundcloud(query) + yt_task = search_youtube(query) + a_task = search_apple(query) - y_res, a_res, yt_res = await asyncio.gather(yandex_task, apple_task, youtube_task) + y_res, sc_res, yt_res, a_res = await asyncio.gather( + y_task, sc_task, yt_task, a_task, return_exceptions=True + ) - all_tracks = y_res["tracks"] + a_res["tracks"] + yt_res["tracks"] - all_albums = y_res["albums"] + a_res["albums"] + all_tracks = [] + all_albums = [] + + for res in (y_res, sc_res, yt_res, a_res): + if isinstance(res, dict): + all_tracks.extend(res.get("tracks", [])) + all_albums.extend(res.get("albums", [])) + + # Calculate relevance score for each track + for t in all_tracks: + t["relevance_score"] = round(calculate_track_relevance(query, t), 2) + + # Sort tracks descending by relevance score: + # Tracks with the highest match (e.g. found on Yandex, SoundCloud, YouTube) appear at the top! + all_tracks.sort(key=lambda t: t.get("relevance_score", 0), reverse=True) + + # Sort albums by relevance as well + if all_albums: + for al in all_albums: + al["relevance_score"] = round(calculate_track_relevance(query, al), 2) + all_albums.sort(key=lambda a: a.get("relevance_score", 0), reverse=True) return { "query": query, diff --git a/app/providers/universal.py b/app/providers/universal.py index c0ccd83..df632d8 100644 --- a/app/providers/universal.py +++ b/app/providers/universal.py @@ -17,7 +17,10 @@ class UniversalProvider(BaseProvider): "music.apple.com", "youtube.com", "youtu.be", - "music.youtube.com" + "music.youtube.com", + "soundcloud.com", + "snd.sc", + "api.soundcloud.com" ]) async def resolve(self, url: str) -> List[TrackMetadata]: @@ -106,15 +109,16 @@ class UniversalProvider(BaseProvider): continue title = entry.get("title", "Unknown") uploader = entry.get("uploader") or entry.get("artist") or "Unknown" + service_name = "soundcloud" if ("soundcloud" in (entry.get("webpage_url") or url).lower() or "snd.sc" in url.lower()) else "youtube" tracks.append(TrackMetadata( - service="youtube", + service=service_name, service_id=str(entry.get("id", "")), title=title, artist=uploader, album=entry.get("album"), duration_ms=int(entry.get("duration", 0) * 1000) if entry.get("duration") else None, cover_url=entry.get("thumbnail"), - source_url=entry.get("webpage_url") or f"https://www.youtube.com/watch?v={entry.get('id')}" + source_url=entry.get("webpage_url") or url )) except Exception as e: print(f"yt-dlp resolve error: {e}") @@ -144,7 +148,7 @@ class UniversalProvider(BaseProvider): } try: - target_input = track.source_url if track.service == "youtube" and track.source_url else f"ytsearch1:{query}" + target_input = track.source_url if track.service in ("youtube", "soundcloud") and track.source_url else f"ytsearch1:{query}" with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([target_input]) diff --git a/app/static/style.css b/app/static/style.css index 751d4f0..cd8f774 100644 --- a/app/static/style.css +++ b/app/static/style.css @@ -400,8 +400,10 @@ input[type="text"]:focus { } .tag-yandex { background-color: rgba(239, 68, 68, 0.2); color: #f87171; border: 1px solid rgba(239, 68, 68, 0.4); } +.tag-soundcloud { background-color: rgba(249, 115, 22, 0.2); color: #fb923c; border: 1px solid rgba(249, 115, 22, 0.4); } .tag-apple { background-color: rgba(236, 72, 153, 0.2); color: #f472b6; border: 1px solid rgba(236, 72, 153, 0.4); } .tag-youtube { background-color: rgba(225, 29, 72, 0.2); color: #fb7185; border: 1px solid rgba(225, 29, 72, 0.4); } +.tag-spotify { background-color: rgba(34, 197, 94, 0.2); color: #4ade80; border: 1px solid rgba(34, 197, 94, 0.4); } .empty-state { text-align: center; diff --git a/app/templates/index.html b/app/templates/index.html index 73cfe94..2b5c9f0 100644 --- a/app/templates/index.html +++ b/app/templates/index.html @@ -54,7 +54,7 @@