Jarian Cottingham
06877fb2fc
- New TUI using Textual framework with responsive interface - Search, results, download, and queue screens for YouTube video management - Download queue system with sequential background downloads - Status bar with queue count and download progress indicators - Comprehensive test suite (97 tests) with 100% pass rate - Modern Python packaging with pyproject.toml and uv support - Added requirements-tui.txt for TUI-specific dependencies - Updated setup.py and install.sh for TUI integration - Enhanced README.md with TUI usage documentation
312 lines
9.0 KiB
Python
312 lines
9.0 KiB
Python
#!/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)
|