#!/usr/bin/env python3 """ YouTube service wrapper around YouTubeCLI - Async implementation """ import asyncio from typing import Any, Dict, List, Optional, Set from rich.console import Console from youtube_cli.main import YouTubeCLI from youtube_tui.models.video import Video console = Console() class YouTubeServiceError(Exception): """Base exception for YouTubeService errors""" pass class SearchError(YouTubeServiceError): """Exception raised during search operations""" pass class DownloadError(YouTubeServiceError): """Exception raised during download operations""" pass class ArchiveError(YouTubeServiceError): """Exception raised during archive operations""" pass class YouTubeService: """Service class that wraps YouTubeCLI for TUI integration with async support""" def __init__(self, config_path: Optional[str] = None): """Initialize the YouTube service""" self.cli = YouTubeCLI(config_path=config_path) self.console = Console() async def search_videos( self, query: str, page: int = 1, per_page: int = 15, ) -> List[Video]: """ Search for videos on YouTube (async) Args: query: Search query string page: Page number (1-indexed) per_page: Number of videos per page (ignored, hardcoded to 15 in CLI) Returns: List of Video objects Raises: SearchError: If search fails """ # Use asyncio.to_thread to run blocking subprocess calls def _search() -> List[Video]: try: # Call the search_videos method with return_results=True results = self.cli.search_videos( query, self.cli.config, page, return_results=True ) if not results: return [] # Convert results to Video objects return [self._create_video_from_result(r) for r in results] except Exception as e: self.console.print(f"[red]Error searching videos: {e}[/red]") raise SearchError(f"Failed to search videos: {e}") from e return await asyncio.to_thread(_search) async def download_video( self, video: Video, category: Optional[str] = None, network_folder: Optional[str] = None, progress_callback=None, ) -> bool: """ Download a video (async) Args: video: Video object to download category: Category folder for download location network_folder: Optional network share folder progress_callback: Optional callback to report progress (percentage: int) Returns: True if download succeeded, False otherwise Raises: DownloadError: If download fails """ def _download() -> bool: try: success = self.cli.download_video( video.url, self.cli.config, category=category, network_folder=network_folder, progress_callback=progress_callback, ) return success is not False # download_video returns None on error except Exception as e: self.console.print(f"[red]Error downloading video: {e}[/red]") raise DownloadError(f"Failed to download video: {e}") from e return await asyncio.to_thread(_download) async def download_playlist( self, video: Video, category: Optional[str] = None, network_folder: Optional[str] = None, progress_callback=None, ) -> bool: """ Download a playlist (async) Args: video: Video object containing playlist URL category: Category folder for download location network_folder: Optional network share folder progress_callback: Optional callback to report progress (percentage: int) Returns: True if download succeeded, False otherwise Raises: DownloadError: If download fails """ def _download_playlist() -> bool: try: success = self.cli.download_playlist( video.url, self.cli.config, category=category, network_folder=network_folder, progress_callback=progress_callback, ) return success is not False # download_playlist returns None on error except Exception as e: self.console.print(f"[red]Error downloading playlist: {e}[/red]") raise DownloadError(f"Failed to download playlist: {e}") from e return await asyncio.to_thread(_download_playlist) async def get_categories(self) -> List[str]: """ Get available download categories (async) Returns: List of category names """ def _get_categories() -> List[str]: return self.cli.get_categories(self.cli.config) return await asyncio.to_thread(_get_categories) async def is_video_downloaded(self, video_id: str) -> bool: """ Check if a video has already been downloaded (async) Args: video_id: YouTube video ID Returns: True if video is in archive, False otherwise """ def _check_archive() -> bool: return self.cli.is_video_downloaded(video_id) return await asyncio.to_thread(_check_archive) async def add_to_archive(self, video: Video) -> None: """ Add a video to the archive (async) Args: video: Video object to add Raises: ArchiveError: If archive operation fails """ def _add_to_archive() -> None: try: self.cli.add_to_archive( { "url": video.url, "id": video.video_id, "title": video.title, } ) except Exception as e: self.console.print(f"[red]Error adding to archive: {e}[/red]") raise ArchiveError(f"Failed to add video to archive: {e}") from e await asyncio.to_thread(_add_to_archive) async def get_archive(self) -> Dict[str, Any]: """ Load the entire archive (async) Returns: Archive dictionary containing all downloaded videos """ def _load_archive() -> Dict[str, Any]: return self.cli.load_archive() return await asyncio.to_thread(_load_archive) async def get_downloaded_video_ids(self) -> Set[str]: """ Get set of all downloaded video IDs (async) Returns: Set of video IDs that have been downloaded """ archive = await self.get_archive() return set(archive.keys()) async def remove_from_archive(self, video_id: str) -> bool: """ Remove a video from the archive (async) Args: video_id: YouTube video ID to remove Returns: True if video was removed, False if not found """ def _remove_from_archive() -> bool: try: archive = self.cli.load_archive() if video_id in archive: del archive[video_id] self.cli.save_archive(archive) return True return False except Exception: return False return await asyncio.to_thread(_remove_from_archive) def _create_video_from_result(self, result: Dict[str, Any]) -> Video: """ Create a Video object from yt-dlp result Args: result: yt-dlp search result dictionary Returns: Video object """ duration = result.get("length", "0:00") if duration: duration = str(duration) else: duration = "0:00" # Extract channel from author channel = result.get("author", result.get("channel", "Unknown")) return Video( video_id=result.get("id", ""), title=result.get("title", "Unknown"), channel=channel, channel_id=result.get("channel_id", ""), duration=duration, view_count=str(result.get("view_count", "0")), upload_date=result.get("upload_date", ""), description=result.get("description", ""), thumbnail_url=result.get("thumbnail"), url=result.get("url", ""), ) def format_duration(self, seconds: int) -> str: """ Format duration in seconds to MM:SS or HH:MM:SS format Args: seconds: Duration in seconds Returns: Formatted duration string """ return self.cli.format_duration(seconds)