95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from Provider._base import SearchProvider, SearchResult
|
|
from SYS.logger import log
|
|
|
|
|
|
class YouTube(SearchProvider):
|
|
"""Search provider for YouTube using yt-dlp."""
|
|
|
|
def search(
|
|
self,
|
|
query: str,
|
|
limit: int = 10,
|
|
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]
|
|
|
|
try:
|
|
process = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
|
|
if process.returncode != 0:
|
|
log(f"[youtube] yt-dlp failed: {process.stderr}", file=sys.stderr)
|
|
return []
|
|
|
|
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,
|
|
},
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
except Exception as exc:
|
|
log(f"[youtube] Error: {exc}", file=sys.stderr)
|
|
return []
|
|
|
|
def validate(self) -> bool:
|
|
return shutil.which("yt-dlp") is not None
|