lksjalk

Provider/youtube.py

@@ -1,8 +1,5 @@
 from __future__ import annotations
 
-import json
-import shutil
-import subprocess
 import sys
 from typing import Any, Dict, List, Optional
 
@@ -11,7 +8,7 @@ from SYS.logger import log
 
 
 class YouTube(Provider):
-    """Search provider for YouTube using yt-dlp."""
+    """Search provider for YouTube using the yt_dlp Python package."""
 
     def search(
         self,
@@ -20,75 +17,56 @@ class YouTube(Provider):
         filters: Optional[Dict[str, Any]] = None,
         **kwargs: Any,
     ) -> List[SearchResult]:
-        ytdlp_path = shutil.which("yt-dlp")
-        if not ytdlp_path:
-            log("[youtube] yt-dlp not found in PATH", file=sys.stderr)
-            return []
-
-        search_query = f"ytsearch{limit}:{query}"
-        cmd = [ytdlp_path, "--dump-json", "--flat-playlist", "--no-warnings", search_query]
-
+        # Use the yt_dlp Python module (installed via requirements.txt).
         try:
-            process = subprocess.run(
-                cmd,
-                capture_output=True,
-                text=True,
-                encoding="utf-8",
-                errors="replace",
-            )
+            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)
 
-            if process.returncode != 0:
-                log(f"[youtube] yt-dlp failed: {process.stderr}", file=sys.stderr)
-                return []
+                    duration_str = f"{int(duration // 60)}:{int(duration % 60):02d}" if duration else ""
+                    views_str = f"{view_count:,}" if view_count else ""
 
-            results: List[SearchResult] = []
-            for line in process.stdout.splitlines():
-                if not line.strip():
-                    continue
-
-                try:
-                    video_data = json.loads(line)
-                except json.JSONDecodeError:
-                    continue
-
-                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,
-                        },
+                    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 as exc:
-            log(f"[youtube] Error: {exc}", file=sys.stderr)
+                return results
+        except Exception:
+            log("[youtube] yt_dlp import failed", file=sys.stderr)
             return []
 
     def validate(self) -> bool:
-        return shutil.which("yt-dlp") is not None
+        try:
+            import yt_dlp  # type: ignore
+            return True
+        except Exception:
+            return False