from __future__ import annotations import sys from typing import Any, Dict, List, Optional from ProviderCore.base import Provider, SearchResult from SYS.logger import log class YouTube(Provider): """Search provider for YouTube using the yt_dlp Python package.""" def search( self, query: str, limit: int = 10, filters: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> List[SearchResult]: # Use the yt_dlp Python module (installed via requirements.txt). try: import yt_dlp # type: ignore ydl_opts: Dict[str, Any] = { "quiet": True, "skip_download": True, "extract_flat": True } with yt_dlp.YoutubeDL(ydl_opts) as ydl: # type: ignore[arg-type] search_query = f"ytsearch{limit}:{query}" info = ydl.extract_info(search_query, download=False) entries = info.get("entries") or [] results: List[SearchResult] = [] for video_data in entries[:limit]: title = video_data.get("title", "Unknown") video_id = video_data.get("id", "") url = video_data.get( "url" ) or f"https://youtube.com/watch?v={video_id}" uploader = video_data.get("uploader", "Unknown") duration = video_data.get("duration", 0) view_count = video_data.get("view_count", 0) duration_str = ( f"{int(duration // 60)}:{int(duration % 60):02d}" if duration else "" ) views_str = f"{view_count:,}" if view_count else "" results.append( SearchResult( table="youtube", title=title, path=url, detail=f"By: {uploader}", annotations=[duration_str, f"{views_str} views"], media_kind="video", columns=[ ("Title", title), ("Uploader", uploader), ("Duration", duration_str), ("Views", views_str), ], full_metadata={ "video_id": video_id, "uploader": uploader, "duration": duration, "view_count": view_count, }, ) ) return results except Exception: log("[youtube] yt_dlp import failed", file=sys.stderr) return [] def validate(self) -> bool: try: import yt_dlp # type: ignore return True except Exception: return False