From 17f1413d5ecc1607e73f9ea28841cb9888091481 Mon Sep 17 00:00:00 2001 From: Jarian Cottingham Date: Fri, 21 Aug 2026 17:08:28 +0000 Subject: [PATCH] Initial commit: youtube-tui (extracted from youtube-cli) --- .gitignore | 4 + LICENSE | 21 ++ README.md | 35 +++ __init__.py | 5 + __main__.py | 9 + app.py | 296 ++++++++++++++++++++++++ models/__init__.py | 7 + models/queue_item.py | 144 ++++++++++++ models/video.py | 77 +++++++ pyproject.toml | 59 +++++ screens/__init__.py | 19 ++ screens/download.py | 204 +++++++++++++++++ screens/help.py | 183 +++++++++++++++ screens/history.py | 186 +++++++++++++++ screens/home.py | 192 ++++++++++++++++ screens/modal.py | 253 ++++++++++++++++++++ screens/queue.py | 433 +++++++++++++++++++++++++++++++++++ screens/results.py | 364 +++++++++++++++++++++++++++++ screens/search.py | 156 +++++++++++++ services/__init__.py | 7 + services/download_manager.py | 220 ++++++++++++++++++ services/queue.py | 268 ++++++++++++++++++++++ services/youtube.py | 344 ++++++++++++++++++++++++++++ test_tui.py | 152 ++++++++++++ widgets/__init__.py | 11 + widgets/command_palette.py | 199 ++++++++++++++++ widgets/footer.py | 106 +++++++++ widgets/status_bar.py | 119 ++++++++++ 28 files changed, 4073 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 __init__.py create mode 100644 __main__.py create mode 100644 app.py create mode 100644 models/__init__.py create mode 100644 models/queue_item.py create mode 100644 models/video.py create mode 100644 pyproject.toml create mode 100644 screens/__init__.py create mode 100644 screens/download.py create mode 100644 screens/help.py create mode 100644 screens/history.py create mode 100644 screens/home.py create mode 100644 screens/modal.py create mode 100644 screens/queue.py create mode 100644 screens/results.py create mode 100644 screens/search.py create mode 100644 services/__init__.py create mode 100644 services/download_manager.py create mode 100644 services/queue.py create mode 100644 services/youtube.py create mode 100644 test_tui.py create mode 100644 widgets/__init__.py create mode 100644 widgets/command_palette.py create mode 100644 widgets/footer.py create mode 100644 widgets/status_bar.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2ec1dba --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.pyc +.pytest_cache/ +*.egg-info/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..7c6c6d6 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 jarianc + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..de40151 --- /dev/null +++ b/README.md @@ -0,0 +1,35 @@ +# YouTube TUI + +Text-based User Interface (TUI) for YouTube CLI + +## Installation + +```bash +pip install -e . +``` + +## Usage + +```bash +youtube-tui +``` + +## Development + +```bash +# Install dependencies +pip install textual>=8.0 + +# Run the TUI +python -m youtube_tui +``` + +## Project Structure + +- `__init__.py` - Package initialization +- `__main__.py` - Entry point for `python -m youtube_tui` +- `app.py` - Main Textual application +- `models/` - Data models (Video, etc.) +- `services/` - Business logic services +- `widgets/` - Textual widgets +- `screens/` - TUI screens \ No newline at end of file diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..2d2e52c --- /dev/null +++ b/__init__.py @@ -0,0 +1,5 @@ +""" +YouTube TUI - A Text-based User Interface for browsing and downloading YouTube videos +""" + +__version__ = "0.1.0" diff --git a/__main__.py b/__main__.py new file mode 100644 index 0000000..26a312c --- /dev/null +++ b/__main__.py @@ -0,0 +1,9 @@ +#!/usr/bin/env python3 +""" +Entry point for the YouTube TUI application +""" + +from youtube_tui.app import main + +if __name__ == "__main__": + main() diff --git a/app.py b/app.py new file mode 100644 index 0000000..f520c29 --- /dev/null +++ b/app.py @@ -0,0 +1,296 @@ +#!/usr/bin/env python3 +""" +Main Textual Application class for YouTube TUI +Enhanced with command palette, help screen, status bar, and theme support +""" + +import json +import logging +import subprocess +from datetime import datetime +from logging.handlers import RotatingFileHandler +from pathlib import Path +from typing import Any, Optional + +from textual.app import App, ComposeResult +from textual.widgets import Header, Static + +# Configure logging +LOG_DIR = Path.home() / ".config" / "youtube_cli" / "logs" +LOG_DIR.mkdir(parents=True, exist_ok=True) +LOG_FILE = LOG_DIR / "app.log" + +# Use RotatingFileHandler for log rotation (10MB, 5 backups) +file_handler = RotatingFileHandler(LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5) +file_handler.setLevel(logging.DEBUG) +file_handler.setFormatter( + logging.Formatter( + "%(asctime)s | %(name)s | %(levelname)s | %(message)s", + "%Y-%m-%d %H:%M:%S", + ) +) + +# Create console handler +console_handler = logging.StreamHandler() +console_handler.setLevel(logging.INFO) +console_handler.setFormatter(logging.Formatter("%(message)s")) + +# Configure root logger +logging.basicConfig( + level=logging.DEBUG, + handlers=[ + file_handler, + console_handler, + ], +) +logger = logging.getLogger(__name__) + +from youtube_tui.models.video import Video +from youtube_tui.services.youtube import YouTubeService +from youtube_tui.services.queue import DownloadQueue +from youtube_tui.services.download_manager import DownloadManager +from youtube_tui.widgets.footer import CustomFooter + + +class YouTubeTUI(App): + """Main application class for YouTube TUI""" + + VERSION = "0.1.0" + + CSS = """ + Screen { + align: center middle; + } + + #header { + dock: top; + } + + #footer { + dock: bottom; + } + """ + + BINDINGS = [ + ("q", "quit", "Quit"), + ("escape", "cancel", "Cancel"), + ("ctrl+p", "command_palette", "Command Palette"), + ("ctrl+h", "show_help", "Help"), + ("ctrl+t", "toggle_theme", "Toggle Theme"), + ("ctrl+r", "refresh_screen", "Refresh"), + ("ctrl+f", "open_search", "Search"), + ("ctrl+l", "open_queue", "Queue"), + ] + + def __init__(self, *args: Any, **kwargs: Any) -> None: + # Set _theme_name directly to avoid property issues during init + object.__setattr__(self, "_theme_name", "textual-dark") + super().__init__(*args, **kwargs) + self.current_screen: Optional[object] = None + self.current_search_term = "" + self.search_history: list = [] + self.load_search_history() + self.yt_dlp_version = "unknown" + self._check_yt_dlp() + self.downloading = False + # Initialize queue and download manager + self.download_queue: Optional[DownloadQueue] = None + self.youtube_service: Optional[YouTubeService] = None + self.download_manager: Optional[DownloadManager] = None + + @property + def theme(self) -> str: + return getattr(self, "_theme_name", "textual-dark") + + @theme.setter + def theme(self, value: str) -> None: + self._theme_name = value + + def compose(self) -> ComposeResult: + """Compose the UI layout with enhanced components""" + yield Header() + yield Static("YouTube TUI - Browse and download videos", id="main-content") + yield CustomFooter(self) + + async def action_quit(self) -> None: + """Quit the application""" + self.exit() + + def action_cancel(self) -> None: + """Handle escape key""" + if hasattr(self.screen, "action_cancel"): + self.screen.action_cancel() + + def on_mount(self) -> None: + """Called when the app is mounted""" + # Initialize download queue and manager + self.youtube_service = YouTubeService() + self.download_queue = DownloadQueue() + self.download_manager = DownloadManager( + self.download_queue, self.youtube_service + ) + self.download_manager.start_processing() + + self.push_home_screen() + self.current_screen = self.screen + + def on_screen_stack_changed(self) -> None: + """Called when the screen stack changes""" + current_screen = self.screen + if hasattr(current_screen, "search_term"): + self.current_search_term = current_screen.search_term + self.current_screen = current_screen + + def push_home_screen(self) -> None: + """Push the home screen""" + from youtube_tui.screens.home import HomeScreen + + self.push_screen(HomeScreen()) + + def push_search_screen(self) -> None: + """Push the search screen""" + from youtube_tui.screens.search import SearchScreen + + self.push_screen(SearchScreen()) + + def push_results_screen(self, search_term: str, page: int = 1) -> None: + """Push the results screen""" + from youtube_tui.screens.results import ResultsScreen + + self.push_screen(ResultsScreen(search_term, page)) + + def push_download_screen(self, video: Video) -> None: + """Push the download screen""" + from youtube_tui.screens.download import DownloadScreen + + self.push_screen(DownloadScreen(video)) + + def push_category_modal(self) -> str: + """Push the category selection modal and return selected category""" + from youtube_tui.screens.modal import CategorySelectionModal + + # ModalScreen doesn't have request_screen in textual + self.push_screen(CategorySelectionModal()) + return "" + + def _check_yt_dlp(self) -> None: + """Check yt-dlp installation and version""" + try: + result = subprocess.run( + ["yt-dlp", "--version"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0: + self.yt_dlp_version = result.stdout.strip() + except Exception: + self.yt_dlp_version = "not installed" + + def load_search_history(self) -> None: + """Load search history from config file""" + config_dir = Path.home() / ".config" / "youtube_cli" + history_file = config_dir / "search_history.json" + + if history_file.exists(): + try: + with open(history_file, "r") as f: + self.search_history = json.load(f) + except Exception: + self.search_history = [] + + def save_search_history(self) -> None: + """Save search history to config file""" + config_dir = Path.home() / ".config" / "youtube_cli" + config_dir.mkdir(parents=True, exist_ok=True) + + history_file = config_dir / "search_history.json" + try: + with open(history_file, "w") as f: + json.dump(self.search_history, f, indent=2) + except Exception: + pass # Silently fail if we can't save + + def add_to_search_history(self, search_term: str) -> None: + """Add a search term to history""" + # Remove duplicates + self.search_history = [ + item + for item in self.search_history + if item.get("search_term") != search_term + ] + + # Add new entry + self.search_history.insert( + 0, + { + "search_term": search_term, + "timestamp": datetime.now().isoformat(), + }, + ) + + # Keep only last 50 searches + self.search_history = self.search_history[:50] + + self.save_search_history() + + def action_command_palette(self) -> None: + """Open the command palette""" + from youtube_tui.widgets.command_palette import CommandPalette + + self.push_screen(CommandPalette()) + + def action_show_help(self) -> None: + """Show the help screen""" + from youtube_tui.screens.help import HelpScreen + + self.push_screen(HelpScreen()) + + def action_toggle_theme(self) -> None: + """Toggle between dark and light themes""" + # Textual 0.43+ has built-in dark/light theme support + # We'll cycle through available themes + if self.theme == "css": + self.theme = "textual-dark" + elif self.theme == "textual-dark": + self.theme = "textual-light" + else: + self.theme = "css" + self.refresh() + + def action_refresh_screen(self) -> None: + """Refresh the current screen""" + current_screen = self.screen + if hasattr(current_screen, "refresh"): + current_screen.refresh() + + def action_open_search(self) -> None: + """Open the search screen from anywhere""" + from youtube_tui.screens.search import SearchScreen + + self.push_screen(SearchScreen()) + + def action_open_queue(self) -> None: + """Open the queue screen from anywhere""" + from youtube_tui.screens.queue import QueueScreen + + self.push_screen(QueueScreen()) + + def on_search_complete(self, search_term: str) -> None: + """Handle search completion""" + self.current_search_term = search_term + self.add_to_search_history(search_term) + + def set_downloading(self, downloading: bool) -> None: + """Set downloading state for status bar""" + self.downloading = downloading + + +def main() -> None: + """Main entry point for the TUI application""" + app = YouTubeTUI() + app.run() + + +if __name__ == "__main__": + main() diff --git a/models/__init__.py b/models/__init__.py new file mode 100644 index 0000000..63c406c --- /dev/null +++ b/models/__init__.py @@ -0,0 +1,7 @@ +""" +Models package for YouTube TUI +""" + +from youtube_tui.models.video import Video + +__all__ = ["Video"] diff --git a/models/queue_item.py b/models/queue_item.py new file mode 100644 index 0000000..8ca66d9 --- /dev/null +++ b/models/queue_item.py @@ -0,0 +1,144 @@ +#!/usr/bin/env python3 +""" +Download Queue System for YouTube TUI +""" + +import uuid +from enum import Enum +from typing import Optional + +from youtube_tui.models.video import Video + + +class QueueStatus(Enum): + """Status of a queue item""" + + PENDING = "pending" + DOWNLOADING = "downloading" + COMPLETED = "completed" + CANCELLED = "cancelled" + FAILED = "failed" + + +class QueueItem: + """Represents an item in the download queue""" + + def __init__( + self, + video: Optional[Video], + category: Optional[str] = None, + network_folder: Optional[str] = None, + ): + self._id = uuid.uuid4() # Unique identifier for tracking + self.video = video + self.category = category + self.network_folder = network_folder + self.status = QueueStatus.PENDING + self.progress = 0 + self.started_at: Optional[str] = None + self.completed_at: Optional[str] = None + self.error_message: Optional[str] = None # Error details when failed + + @property + def id(self) -> uuid.UUID: + """Get the unique ID of this queue item""" + return self._id + + @property + def is_active(self) -> bool: + """Check if this item is currently active (downloading or pending)""" + return self.status in ( + QueueStatus.PENDING, + QueueStatus.DOWNLOADING, + ) + + @property + def is_complete(self) -> bool: + """Check if this item has completed (successfully or not)""" + return self.status in ( + QueueStatus.COMPLETED, + QueueStatus.CANCELLED, + QueueStatus.FAILED, + ) + + def to_dict(self) -> dict: + """Convert to dictionary for JSON serialization""" + return { + "id": str(self._id), + "video": self.video.to_dict() if self.video else None, + "category": self.category, + "network_folder": self.network_folder, + "status": self.status.value, + "progress": self.progress, + "started_at": self.started_at, + "completed_at": self.completed_at, + "error_message": self.error_message, + } + + @classmethod + def from_dict(cls, data: dict) -> "QueueItem": + """Create QueueItem instance from dictionary""" + video_data = data.get("video") + video: Video = ( + Video.from_dict(video_data) + if video_data + else Video( + video_id="", + title="Unknown Video", + channel="Unknown Channel", + channel_id="", + duration="0:00", + view_count="0", + upload_date="", + description="", + ) + ) + + item = cls( + video=video, + category=data.get("category"), + network_folder=data.get("network_folder"), + ) + # Parse UUID from string if present + item_id = data.get("id") + if item_id: + try: + item._id = uuid.UUID(item_id) + except ValueError: + pass # Keep auto-generated UUID if parsing fails + item.status = QueueStatus(data.get("status", "pending")) + item.progress = data.get("progress", 0) + item.started_at = data.get("started_at") + item.completed_at = data.get("completed_at") + item.error_message = data.get("error_message") + return item + + def start_download(self) -> None: + """Mark item as downloading""" + from datetime import datetime + + self.status = QueueStatus.DOWNLOADING + self.started_at = datetime.now().isoformat() + + def update_progress(self, progress: int) -> None: + """Update download progress""" + self.progress = max(0, min(100, progress)) + + def complete(self) -> None: + """Mark item as completed""" + from datetime import datetime + + self.status = QueueStatus.COMPLETED + self.progress = 100 + self.completed_at = datetime.now().isoformat() + + def cancel(self) -> None: + """Mark item as cancelled""" + self.status = QueueStatus.CANCELLED + self.completed_at = None + + def fail(self, error_message: Optional[str] = None) -> None: + """Mark item as failed with optional error message""" + self.status = QueueStatus.FAILED + self.completed_at = None + self.error_message = error_message diff --git a/models/video.py b/models/video.py new file mode 100644 index 0000000..7434908 --- /dev/null +++ b/models/video.py @@ -0,0 +1,77 @@ +""" +Video data model for YouTube TUI +""" + +from dataclasses import dataclass +from typing import Optional + + +@dataclass +class Video: + """Represents a YouTube video""" + + video_id: str + title: str + channel: str + channel_id: str + duration: str + view_count: str + upload_date: str + description: str + thumbnail_url: Optional[str] = None + is_short: bool = False + url: str = "" + + def __post_init__(self) -> None: + """Post-initialization to set URL and detect shorts""" + if not self.url: + self.url = f"https://www.youtube.com/watch?v={self.video_id}" + + if "/shorts/" in self.url or self.duration == "0:00": + self.is_short = True + + @property + def display_title(self) -> str: + """Get title with short indicator""" + if self.is_short: + return f"(short) {self.title}" + return self.title + + @property + def display_duration(self) -> str: + """Get formatted duration""" + if self.is_short: + return "Short" + return self.duration + + def to_dict(self) -> dict: + """Convert to dictionary for JSON serialization""" + return { + "video_id": self.video_id, + "title": self.title, + "channel": self.channel, + "channel_id": self.channel_id, + "duration": self.duration, + "view_count": self.view_count, + "upload_date": self.upload_date, + "description": self.description, + "thumbnail_url": self.thumbnail_url, + "is_short": self.is_short, + "url": self.url, + } + + @classmethod + def from_dict(cls, data: dict) -> "Video": + """Create Video instance from dictionary""" + return cls( + video_id=data["video_id"], + title=data["title"], + channel=data["channel"], + channel_id=data["channel_id"], + duration=data["duration"], + view_count=data["view_count"], + upload_date=data["upload_date"], + description=data.get("description", ""), + thumbnail_url=data.get("thumbnail_url"), + url=data.get("url", ""), + ) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..db00062 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,59 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "youtube-tui" +version = "0.1.0" +description = "Text-based User Interface (TUI) for YouTube CLI" +readme = "README.md" +requires-python = ">=3.9" +authors = [ + {name = "Your Name", email = "your.email@example.com"}, +] +license = {text = "MIT"} +classifiers = [ + "Development Status :: 3 - Alpha", + "Environment :: Console", + "Intended Audience :: End Users/Desktop", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Multimedia :: Video", +] +dependencies = [ + "textual>=8.0", + "yt-dlp", + "rich", + "requests", +] + +[project.scripts] +youtube-tui = "youtube_tui.__main__:main" + +[project.urls] +Homepage = "https://github.com/yourusername/youtube-cli" +Issues = "https://github.com/yourusername/youtube-cli/issues" + +[tool.setuptools] +packages = ["youtube_tui", "youtube_tui.screens", "youtube_tui.widgets", "youtube_tui.models", "youtube_tui.services"] + +[tool.setuptools.package-data] +youtube_tui = ["py.typed"] + +[tool.ruff] +# Skip whitespace checks in CSS strings (Textual styling) +# These are intentional blank lines in CSS +lint.ignore = ["W293", "W291", "E402"] + +[tool.mypy] +python_version = "3.11" +warn_return_any = false +warn_unused_ignores = false +disallow_untyped_defs = false +check_untyped_defs = false +follow_imports = "skip" \ No newline at end of file diff --git a/screens/__init__.py b/screens/__init__.py new file mode 100644 index 0000000..a0366e5 --- /dev/null +++ b/screens/__init__.py @@ -0,0 +1,19 @@ +""" +Screens package for YouTube TUI +""" + +from youtube_tui.screens.download import DownloadScreen +from youtube_tui.screens.help import HelpScreen +from youtube_tui.screens.history import SearchHistoryScreen +from youtube_tui.screens.modal import CategorySelectionModal +from youtube_tui.screens.results import ResultsScreen +from youtube_tui.screens.search import SearchScreen + +__all__ = [ + "SearchScreen", + "ResultsScreen", + "DownloadScreen", + "CategorySelectionModal", + "HelpScreen", + "SearchHistoryScreen", +] diff --git a/screens/download.py b/screens/download.py new file mode 100644 index 0000000..f84416f --- /dev/null +++ b/screens/download.py @@ -0,0 +1,204 @@ +#!/usr/bin/env python3 +""" +Download Screen for YouTube TUI +""" + +import asyncio +from typing import Optional + +from textual.app import ComposeResult +from textual.containers import Container +from textual.screen import Screen +from textual.widgets import ( + Footer, + Header, + ProgressBar, + Static, +) + +from youtube_tui.models.video import Video +from youtube_tui.services.youtube import YouTubeService + + +class DownloadScreen(Screen): + """Screen for displaying download progress""" + + CSS = """ + DownloadScreen { + align: center middle; + } + + #download-container { + width: 70%; + height: auto; + border: double #555555; + padding: 2 3; + margin: 2 0; + } + + #video-title { + width: 100%; + height: 3; + content-align: center middle; + background: $surface; + margin-bottom: 1; + } + + #progress-container { + width: 100%; + height: 5; + margin: 2 0; + } + + #status-message { + width: 100%; + height: auto; + content-align: center middle; + margin: 1 0; + } + + #actions { + width: 100%; + height: auto; + dock: bottom; + margin-top: 1; + } + + Button { + width: 15; + margin: 1 1; + } + + #status-bar { + dock: bottom; + height: 1; + background: $surface; + color: $text-muted; + padding: 0 1; + } + """ + + BINDINGS = [ + ("escape", "cancel", "Cancel"), + ("ctrl+r", "refresh_screen", "Refresh"), + ] + + def __init__(self, video: Video, category: Optional[str] = None): + super().__init__() + self.youtube_service = YouTubeService() + self.video = video + self.category = category + self.download_complete = False + self.download_error = False + self.download_task: Optional[asyncio.Task] = None + + def compose(self) -> ComposeResult: + """Compose the download screen""" + yield Header() + yield Container( + Static(f"[bold]{self.video.display_title}[/bold]", id="video-title"), + Static("Preparing download...", id="status-message"), + Container( + ProgressBar(total=100, id="progress-bar"), + id="progress-container", + ), + id="download-container", + ) + yield Static(id="status-bar") + yield Footer() + + def on_mount(self) -> None: + """Called when screen is mounted""" + self.update_status("[blue]Starting download...[/blue]") + # Use asyncio.create_task instead of app.run_background + self.download_task = asyncio.create_task(self.start_download()) + + def action_refresh_screen(self) -> None: + """Refresh the screen""" + # For download screen, refresh just updates the status + self.update_status("[blue]Status: Download in progress...[/blue]") + + async def start_download(self) -> None: + """Start the download process""" + try: + async def progress_callback(percentage: int) -> bool: + """Update progress bar during download""" + self.update_progress(percentage) + self.update_status( + f"[blue]Downloading... {percentage}%[/blue]" + ) + return True + + success = await self.youtube_service.download_video( + self.video, self.category, progress_callback=progress_callback + ) + + if success: + self.download_complete = True + self.update_status("[green]Download completed![/green]") + self.update_progress(100) + # Wait briefly before returning + await asyncio.sleep(2) + self.app.pop_screen() + else: + self.download_error = True + self.update_status("[red]Download failed![/red]") + # Wait briefly before returning + await asyncio.sleep(2) + self.app.pop_screen() + + except asyncio.CancelledError: + # Task was cancelled + self.download_error = True + self.update_status("[yellow]Download cancelled[/yellow]") + self.app.pop_screen() + except Exception as e: + self.download_error = True + self.update_status(f"[red]Error: {e}[/red]") + # Wait briefly before returning + await asyncio.sleep(2) + self.app.pop_screen() + + def update_progress(self, percentage: int) -> None: + """Update the progress bar""" + progress_bar = self.query_one("#progress-bar", ProgressBar) + progress_bar.progress = percentage + + def update_status(self, message: str) -> None: + """Update the status message""" + status_message = self.query_one("#status-message", Static) + status_message.update(message) + + status_bar = self.query_one("#status-bar", Static) + status_bar.update(f"[bold white]{message}[/bold white]") + + def action_cancel(self) -> None: + """Cancel the download""" + # Check if there's a download manager and queue to cancel + if hasattr(self.app, "download_manager") and self.app.download_manager: + # Cancel via the download manager + self.app.download_manager.cancel_active_download() + + if self.download_task: + self.download_task.cancel() + self.download_error = True + self.update_status("[yellow]Download cancelled[/yellow]") + self.app.pop_screen() + + def on_unload(self) -> None: + """Called when screen is unloaded""" + if self.download_complete: + # Show success message briefly before returning + self.app.notify( + f"Downloaded: {self.video.display_title}", + title="Success", + severity="information", + timeout=3, + ) + elif self.download_error: + self.app.notify( + f"Failed to download: {self.video.display_title}", + title="Error", + severity="error", + timeout=3, + ) diff --git a/screens/help.py b/screens/help.py new file mode 100644 index 0000000..964c33d --- /dev/null +++ b/screens/help.py @@ -0,0 +1,183 @@ +""" +Help Screen for YouTube TUI +""" + +from textual.app import ComposeResult +from textual.containers import Container, VerticalScroll +from textual.screen import ModalScreen +from textual.widgets import Footer, Header, Static + + +class HelpScreen(ModalScreen): + """Help screen with keyboard shortcuts documentation""" + + CSS = """ + HelpScreen { + align: center middle; + } + + #help-container { + width: 80%; + height: 80%; + border: solid #555555; + background: $surface; + padding: 1; + } + + #help-title { + width: 100%; + height: 3; + dock: top; + background: $primary; + content-align: center middle; + color: $text; + } + + .section-title { + width: 100%; + height: 2; + margin: 1 0; + color: $primary; + text-style: bold; + } + + .shortcut-row { + height: 2; + padding: 0 1; + } + + .shortcut-key { + width: 20; + color: $accent; + text-style: bold; + } + + .shortcut-desc { + width: 100%; + color: $text; + } + + #app-description { + width: 100%; + height: 6; + margin: 1 0; + color: $text; + } + + #config-info { + width: 100%; + height: auto; + margin: 1 0; + color: $text-muted; + } + + #close-hint { + width: 100%; + height: 2; + dock: bottom; + text-align: center; + color: $text-muted; + } + """ + + BINDINGS = [ + ("escape", "close_help", "Close"), + ("q", "close_help", "Close"), + ] + + def compose(self) -> ComposeResult: + """Compose the help screen""" + yield Header() + yield Container( + Static("YouTube TUI Help", id="help-title"), + VerticalScroll( + Static( + "A Text-based User Interface for browsing and downloading YouTube videos", + id="app-description", + ), + Static("Global Keyboard Shortcuts", classes="section-title"), + Static( + "[key]q[/key] [desc]Quit application[/desc]", + classes="shortcut-row", + ), + Static( + "[key]escape[/key] [desc]Cancel current operation / go back[/desc]", + classes="shortcut-row", + ), + Static( + "[key]ctrl+p[/key] [desc]Open command palette[/desc]", + classes="shortcut-row", + ), + Static( + "[key]ctrl+h[/key] [desc]Show this help screen[/desc]", + classes="shortcut-row", + ), + Static( + "[key]ctrl+t[/key] [desc]Toggle theme (dark/light)[/desc]", + classes="shortcut-row", + ), + Static( + "[key]ctrl+r[/key] [desc]Refresh current screen[/desc]", + classes="shortcut-row", + ), + Static( + "[key]ctrl+f[/key] [desc]Open search from any screen[/desc]", + classes="shortcut-row", + ), + Static("Search Screen Shortcuts", classes="section-title"), + Static( + "[key]enter[/key] [desc]Perform search[/desc]", + classes="shortcut-row", + ), + Static("Results Screen Shortcuts", classes="section-title"), + Static( + "[key]n[/key] [desc]Next page[/desc]", + classes="shortcut-row", + ), + Static( + "[key]p[/key] [desc]Previous page[/desc]", + classes="shortcut-row", + ), + Static( + "[key]enter[/key] [desc]Download selected video[/desc]", + classes="shortcut-row", + ), + Static( + "[key]escape[/key] [desc]Go back to search[/desc]", + classes="shortcut-row", + ), + Static( + "Category Selection Modal Shortcuts", + classes="section-title", + ), + Static( + "[key]arrow keys[/key] [desc]Navigate options[/desc]", + classes="shortcut-row", + ), + Static( + "[key]enter[/key] [desc]Select category[/desc]", + classes="shortcut-row", + ), + Static( + "[key]escape[/key] [desc]Cancel[/desc]", + classes="shortcut-row", + ), + Static("Configuration", classes="section-title"), + Static( + "Configuration file: [path]~/.config/youtube_cli/config.json[/path]", + classes="config-info", + ), + Static( + "Archive file: [path]~/.config/youtube_cli/downloaded_videos.json[/path]", + classes="config-info", + ), + id="help-content", + ), + Static("Press [key]ESC[/key] or [key]Q[/key] to close", id="close-hint"), + id="help-container", + ) + yield Footer() + + def action_close_help(self) -> None: + """Close the help screen""" + self.app.pop_screen() diff --git a/screens/history.py b/screens/history.py new file mode 100644 index 0000000..31c8210 --- /dev/null +++ b/screens/history.py @@ -0,0 +1,186 @@ +""" +Search History Screen for YouTube TUI +""" + +import json +from pathlib import Path + +from textual.app import ComposeResult +from textual.containers import Container +from textual.screen import ModalScreen +from textual.widgets import Footer, Header, ListItem, ListView, Static + + +class SearchHistoryScreen(ModalScreen): + """Screen showing search history""" + + CSS = """ + SearchHistoryScreen { + align: center middle; + } + + #history-container { + width: 70%; + height: 70%; + border: solid #555555; + background: $surface; + padding: 1; + } + + #history-title { + width: 100%; + height: 3; + dock: top; + background: $primary; + content-align: center middle; + color: $text; + } + + ListView { + width: 100%; + height: 100%; + } + + ListItem { + height: 3; + padding: 0 1; + } + + ListItem:hover { + background: $primary-darken-2; + } + + ListItem.--highlight { + background: $primary; + } + + .history-item { + height: 3; + } + + .history-time { + color: $text-muted; + } + + #clear-btn { + margin: 1 1; + } + """ + + BINDINGS = [ + ("escape", "close_history", "Close"), + ("q", "close_history", "Close"), + ("d", "delete_selected", "Delete Selected"), + ("c", "clear_all", "Clear All"), + ] + + def __init__(self): + super().__init__() + self.history_file = ( + Path.home() / ".config" / "youtube_cli" / "search_history.json" + ) + self.search_history = [] + + def compose(self) -> ComposeResult: + """Compose the history screen""" + yield Header() + yield Container( + Static("Search History", id="history-title"), + ListView(id="history-list"), + id="history-container", + ) + yield Footer() + + def on_mount(self) -> None: + """Load history on mount""" + self.load_history() + + def load_history(self) -> None: + """Load search history from file""" + self.search_history = [] + + if self.history_file.exists(): + try: + with open(self.history_file, "r") as f: + self.search_history = json.load(f) + except Exception: + self.search_history = [] + + # Reverse to show newest first + self.search_history = list(reversed(self.search_history)) + + list_view = self.query_one("#history-list", ListView) + list_view.clear() + + for item in self.search_history: + search_term = item.get("search_term", "") + timestamp = item.get("timestamp", "") + + # Format timestamp + if timestamp: + try: + from datetime import datetime + + dt = datetime.fromisoformat(timestamp) + time_str = dt.strftime("%Y-%m-%d %H:%M") + except Exception: + time_str = timestamp + else: + time_str = "Unknown time" + + list_view.append( + ListItem(Static(f"[bold]{search_term}[/bold]\n[dim]{time_str}[/dim]")) + ) + + def action_delete_selected(self) -> None: + """Delete selected history item""" + list_view = self.query_one("#history-list", ListView) + if list_view.children: + # Get the selected item + selected_index = list_view.index + if ( + selected_index is not None + and selected_index >= 0 + and selected_index < len(self.search_history) + ): + # Remove from history + del self.search_history[selected_index] + self.save_history() + self.load_history() + + def action_clear_all(self) -> None: + """Clear all history""" + self.search_history = [] + self.save_history() + self.load_history() + self.notify("Search history cleared", timeout=2) + + def save_history(self) -> None: + """Save history to file""" + try: + # Ensure directory exists + self.history_file.parent.mkdir(parents=True, exist_ok=True) + + with open(self.history_file, "w") as f: + json.dump(list(reversed(self.search_history)), f, indent=2) + except Exception as e: + self.notify(f"Error saving history: {e}", severity="error", timeout=3) + + def action_close_history(self) -> None: + """Close the history screen""" + self.app.pop_screen() + + def on_list_view_selected(self, event: ListView.Selected) -> None: + """Handle item selection""" + # Use event.index directly, ignore list_view + item_index = event.index + + if 0 <= item_index < len(self.search_history): + # Get the search term + search_term = self.search_history[item_index].get("search_term", "") + + # Push search screen with the term + self.app.push_screen("search") + # Note: We'd need to expose the search screen to set the term + # For now, just notify + self.notify(f"Selected: {search_term}", timeout=2) diff --git a/screens/home.py b/screens/home.py new file mode 100644 index 0000000..01ca5a4 --- /dev/null +++ b/screens/home.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python3 +""" +Home Screen for YouTube TUI +Main dashboard with quick actions +""" + +from textual.app import ComposeResult +from textual.containers import Container +from textual.screen import Screen +from textual.widgets import ( + Button, + Footer, + Header, + Static, +) + + +class HomeScreen(Screen): + """Main dashboard screen with quick actions""" + + CSS = """ + HomeScreen { + align: center middle; + } + + #welcome-container { + width: 60%; + height: auto; + border: double #555555; + padding: 2 3; + margin: 2 0; + } + + #welcome-title { + width: 100%; + height: 3; + content-align: center middle; + margin-bottom: 1; + } + + #welcome-info { + width: 100%; + height: auto; + margin-bottom: 2; + color: $text-muted; + } + + #quick-actions { + width: 100%; + layout: grid; + grid-gutter: 1; + grid-columns: 2; + margin: 1 0; + } + + Button { + width: 100%; + margin: 0 1; + } + + #status-info { + width: 100%; + height: auto; + margin-top: 2; + color: $text-muted; + } + + #status-bar { + dock: bottom; + height: 1; + background: $surface; + color: $text-muted; + padding: 0 1; + } + """ + + BINDINGS = [ + ("s", "open_search", "Search"), + ("q", "open_queue", "Queue"), + ("h", "open_history", "History"), + ("ctrl+h", "show_help", "Help"), + ("q", "quit", "Quit"), + ] + + def __init__(self): + super().__init__() + self.download_queue = None + self.download_manager = None + + def compose(self) -> ComposeResult: + """Compose the home screen""" + yield Header() + yield Container( + Static( + "[bold cyan]YouTube CLI[/bold cyan] - Browse and download videos", + id="welcome-title", + ), + Static( + "Use keyboard shortcuts or buttons to navigate", + id="welcome-info", + ), + Container( + Button("Search Videos", id="search-btn"), + Button("Download Queue", id="queue-btn"), + Button("Search History", id="history-btn"), + Button("Help", id="help-btn"), + id="quick-actions", + ), + Static(id="status-info"), + id="welcome-container", + ) + yield Static(id="status-bar") + yield Footer() + + def on_mount(self) -> None: + """Called when screen is mounted""" + if hasattr(self.app, "download_queue"): + self.download_queue = self.app.download_queue + if hasattr(self.app, "download_manager"): + self.download_manager = self.app.download_manager + + self.update_queue_status() + self.update_status("[green]Welcome to YouTube TUI - Press 's' to search[/green]") + + def update_queue_status(self) -> None: + """Update queue status info""" + if self.download_manager: + status = self.download_manager.get_queue_status() + status_text = ( + f"Queue: {status['pending_count']} pending, " + f"{status['downloading_count']} downloading, " + f"{status['total_count']} total" + ) + else: + status_text = "Queue: No active downloads" + + status_info = self.query_one("#status-info", Static) + status_info.update(f"[dim]{status_text}[/dim]") + + def update_status(self, message: str) -> None: + """Update the status bar message""" + status_bar = self.query_one("#status-bar", Static) + status_bar.update(f"[bold white]{message}[/bold white]") + + def action_open_search(self) -> None: + """Open search screen""" + if hasattr(self.app, "push_search_screen"): + self.app.push_search_screen() + else: + from youtube_tui.screens.search import SearchScreen + + self.app.push_screen(SearchScreen()) + + def action_open_queue(self) -> None: + """Open queue screen""" + if hasattr(self.app, "action_open_queue"): + self.app.action_open_queue() + else: + from youtube_tui.screens.queue import QueueScreen + + self.app.push_screen(QueueScreen()) + + def action_open_history(self) -> None: + """Open search history""" + from youtube_tui.screens.history import SearchHistoryScreen + + self.app.push_screen(SearchHistoryScreen()) + + def action_show_help(self) -> None: + """Show help screen""" + from youtube_tui.screens.help import HelpScreen + + self.app.push_screen(HelpScreen()) + + def action_quit(self) -> None: + """Quit the application""" + self.app.exit() + + def action_cancel(self) -> None: + """Handle escape key""" + pass + + def on_button_pressed(self, event: Button.Pressed) -> None: + """Handle button presses""" + if event.button.id == "search-btn": + self.action_open_search() + elif event.button.id == "queue-btn": + self.action_open_queue() + elif event.button.id == "history-btn": + self.action_open_history() + elif event.button.id == "help-btn": + self.action_show_help() diff --git a/screens/modal.py b/screens/modal.py new file mode 100644 index 0000000..6e964f4 --- /dev/null +++ b/screens/modal.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +""" +Category Selection Modal for YouTube TUI +""" + +import logging +from typing import Optional + +from textual.app import ComposeResult +from textual.containers import Container, Vertical +from textual.screen import ModalScreen +from textual.widgets import ( + Button, + Footer, + Header, + Input, + ListItem, + ListView, + Static, +) + +from youtube_tui.services.youtube import YouTubeService + +logger = logging.getLogger(__name__) + + +class CategorySelectionModal(ModalScreen): + """Modal for selecting a download category""" + + CSS = """ + CategorySelectionModal { + align: center middle; + } + + #modal-container { + width: 60%; + height: auto; + border: solid #555555; + background: $surface; + padding: 1; + } + + #modal-title { + width: 100%; + height: 3; + dock: top; + background: $primary; + content-align: center middle; + color: $text; + } + + #categories-container { + width: 100%; + height: 20; + margin: 1 0; + } + + #custom-input { + width: 100%; + margin: 1 0; + } + + #modal-actions { + width: 100%; + height: auto; + dock: bottom; + margin-top: 1; + } + + Button { + width: 15; + margin: 1 1; + } + + ListItem { + height: 3; + padding: 0 1; + } + + ListItem:hover { + background: $primary-darken-2; + } + + ListItem.--highlight { + background: $primary; + } + """ + + BINDINGS = [ + ("escape", "close_modal", "Cancel"), + ("enter", "select_category", "Select"), + ("up", "cursor_up", "Cursor Up"), + ("down", "cursor_down", "Cursor Down"), + ] + + def __init__(self) -> None: + super().__init__() + self.youtube_service = YouTubeService() + self.selected_category: Optional[str] = None + self.selected_index = 0 + + def compose(self) -> ComposeResult: + """Compose the modal""" + yield Header() + yield Container( + Static("Select Download Category", id="modal-title"), + Vertical( + Static("Available Categories:", id="categories-label"), + ListView(id="categories-list"), + Static("Or type custom folder name:", id="custom-label"), + Input(placeholder="Enter custom folder name...", id="custom-input"), + id="categories-container", + ), + Container( + Button("Select", id="select-btn"), + Button("Cancel", id="cancel-btn"), + id="modal-actions", + ), + id="modal-container", + ) + yield Footer() + + def on_mount(self) -> None: + """Called when modal is mounted""" + self.load_categories() + self.update_status("Use arrow keys to select, Enter to confirm") + + def load_categories(self) -> None: + """Load available categories into the list""" + list_view = self.query_one("#categories-list", ListView) + list_view.clear() + + try: + # Note: This is called from on_mount which is sync + # In a real async context, this should be awaited + categories: list = self.youtube_service.cli.get_categories( + self.youtube_service.cli.config + ) + + logger.debug(f"load_categories: categories={categories}") + + for category in categories: + # Extract folder name for display + from pathlib import Path + + folder_name = Path(category).name if Path(category).name else "Root" + category_id = Path(category).name.replace(" ", "-").replace("/", "-") + logger.debug( + f"load_categories: category={category}, folder_name={folder_name}, category_id={category_id}" + ) + + # Create a custom widget for the item + item = ListItem( + Static(f" {folder_name}"), id=f"category-{category_id}" + ) + list_view.append(item) + + # Highlight first item + if list_view.children: + list_view.children[0].add_class("--highlight") + self.selected_index = 0 + logger.debug( + f"load_categories: first item highlighted, selected_index={self.selected_index}" + ) + + except Exception as e: + list_view.append( + ListItem(Static(f"[red]Error loading categories: {e}[/red]")) + ) + + def update_status(self, message: str) -> None: + """Update the modal status""" + # We could add a status line if needed + pass + + def action_select_category(self) -> None: + """Select the current category""" + list_view = self.query_one("#categories-list", ListView) + + # Debug logging + logger.debug( + f"action_select_category: selected_index={self.selected_index}, children_count={len(list_view.children)}" + ) + + if list_view.children and 0 <= self.selected_index < len(list_view.children): + # Get the selected item + item = list_view.children[self.selected_index] + category_id = item.id + logger.debug(f"action_select_category: item.id={category_id}") + if category_id and category_id.startswith("category-"): + self.selected_category = category_id.replace("category-", "") + logger.debug( + f"action_select_category: selected_category={self.selected_category}" + ) + self.dismiss(self.selected_category) + return + + # Check custom input + custom_input = self.query_one("#custom-input", Input) + custom_name = custom_input.value.strip() + if custom_name: + self.selected_category = custom_name + self.dismiss(self.selected_category) + return + + # No valid selection + self.update_status("[red]Please select a category or enter a custom name[/red]") + + def action_cursor_up(self) -> None: + """Move cursor up""" + list_view = self.query_one("#categories-list", ListView) + if list_view.children: + # Remove highlight from current item + if 0 <= self.selected_index < len(list_view.children): + list_view.children[self.selected_index].remove_class("--highlight") + + # Move up + self.selected_index = max(0, self.selected_index - 1) + + # Highlight new item + list_view.children[self.selected_index].add_class("--highlight") + + def action_cursor_down(self) -> None: + """Move cursor down""" + list_view = self.query_one("#categories-list", ListView) + if list_view.children: + # Remove highlight from current item + if 0 <= self.selected_index < len(list_view.children): + list_view.children[self.selected_index].remove_class("--highlight") + + # Move down + self.selected_index = min( + len(list_view.children) - 1, self.selected_index + 1 + ) + + # Highlight new item + list_view.children[self.selected_index].add_class("--highlight") + + def action_close_modal(self) -> None: + """Close modal without selecting""" + self.selected_category = None + self.dismiss(None) + + def on_button_pressed(self, event: Button.Pressed) -> None: + """Handle button presses""" + if event.button.id == "select-btn": + self.action_select_category() + elif event.button.id == "cancel-btn": + self.action_close_modal() + + def on_input_submitted(self, event: Input.Submitted) -> None: + """Handle enter key in custom input""" + self.action_select_category() diff --git a/screens/queue.py b/screens/queue.py new file mode 100644 index 0000000..7b76fb4 --- /dev/null +++ b/screens/queue.py @@ -0,0 +1,433 @@ +#!/usr/bin/env python3 +""" +Queue Screen for YouTube TUI +Displays and manages the download queue +""" + +from textual.app import ComposeResult +from textual.containers import Container +from textual.screen import Screen +from textual.widgets import ( + Button, + DataTable, + Footer, + Header, + Static, +) + +from youtube_tui.models.queue_item import QueueStatus + + +class QueueScreen(Screen): + """Screen for displaying and managing the download queue""" + + CSS = """ + QueueScreen { + align: center middle; + } + + #queue-container { + width: 95%; + height: 70%; + border: solid #555555; + margin: 1 0; + } + + #queue-title { + width: 100%; + height: 3; + dock: top; + background: $surface; + content-align: center middle; + } + + #stats-container { + width: 100%; + height: 3; + dock: top; + background: $surface; + content-align: center middle; + margin-bottom: 1; + } + + #status-bar { + dock: bottom; + height: 1; + background: $surface; + color: $text-muted; + padding: 0 1; + } + + Button { + width: 15; + margin: 0 1; + } + + DataTable { + width: 100%; + height: 100%; + } + + DataTable .datatable-row-highlight { + background: $primary; + } + + DataTable .datatable-header { + background: $primary-darken-2; + } + """ + + BINDINGS = [ + ("q", "go_back", "Back"), + ("escape", "go_back", "Back"), + ("ctrl+r", "refresh_screen", "Refresh"), + ("d", "download_selected", "Download Now"), + ("r", "remove_selected", "Remove"), + ("y", "retry_selected", "Retry"), + ("c", "clear_completed", "Clear Completed"), + ("f", "clear_failed", "Clear Failed"), + ("ctrl+f", "search_from_anywhere", "Search"), + ] + + def __init__(self): + super().__init__() + self.youtube_service = None + self.download_queue = None + self.download_manager = None + + def compose(self) -> ComposeResult: + """Compose the queue screen""" + yield Header() + yield Static("Download Queue", id="queue-title") + yield Container( + Static(id="stats-container"), + DataTable(id="queue-table", show_cursor=False), + id="queue-container", + ) + yield Container( + Button("← Back", id="back-btn"), + Button("Refresh", id="refresh-btn"), + Button("Remove", id="remove-btn"), + Button("Retry", id="retry-btn"), + Button("Clear Done", id="clear-done-btn"), + Button("Clear Failed", id="clear-failed-btn"), + id="queue-controls", + ) + yield Static(id="status-bar") + yield Footer() + + def on_mount(self) -> None: + """Called when screen is mounted""" + # Get references from app + if hasattr(self.app, "youtube_service"): + self.youtube_service = self.app.youtube_service + if hasattr(self.app, "download_queue"): + self.download_queue = self.app.download_queue + if hasattr(self.app, "download_manager"): + self.download_manager = self.app.download_manager + + self.update_table() + self.update_stats() + self.update_status("[green]Queue loaded[/green]") + + # Start auto-refresh if downloads active + if self.download_manager and self.download_manager.is_processing(): + self.set_interval(2, self._auto_refresh) + + def _auto_refresh(self) -> None: + """Auto-refresh queue table every 2 seconds""" + self.update_table() + self.update_stats() + + # Stop auto-refresh when no downloads active + if self.download_manager: + status = self.download_manager.get_queue_status() + if not status.get("has_active_download") and not status.get("pending_count"): + self.clear_interval(self._auto_refresh) + + def action_refresh_screen(self) -> None: + """Refresh the screen""" + self.update_table() + self.update_stats() + self.update_status("[blue]Refreshed queue[/blue]") + + def action_search_from_anywhere(self) -> None: + """Open search from anywhere""" + if hasattr(self.app, "action_open_search"): + self.app.action_open_search() + else: + self.app.pop_screen() + + def update_table(self) -> None: + """Update the DataTable with queue items""" + table = self.query_one("#queue-table", DataTable) + + # Get queue items + if self.download_queue: + items = self.download_queue.get_queue() + + # Clear and rebuild columns + table.clear() + + # Set up columns + table.add_columns("Status", "Title", "Category", "Progress") + table.add_columns("Started", "Completed") + table.add_columns("Actions") + + for item in items: + # Get status text with color + status = item.status.value + status_color = { + QueueStatus.PENDING: "yellow", + QueueStatus.DOWNLOADING: "blue", + QueueStatus.COMPLETED: "green", + QueueStatus.CANCELLED: "yellow", + QueueStatus.FAILED: "red", + }.get(item.status, "white") + + # Truncate long titles + title = item.video.display_title if item.video else "Unknown" + if len(title) > 40: + title = title[:37] + "..." + + # Get category + category = item.category or "Default" + + # Format progress + progress = f"{item.progress}%" + if item.status == QueueStatus.DOWNLOADING: + progress = f"[blue]{progress}[/blue]" + + # Format timestamps + started_at = item.started_at or "-" + completed_at = item.completed_at or "-" + + row_key = item.video.video_id if item.video else "" + + # Determine actions for this row + actions = "" + if item.status == QueueStatus.FAILED: + actions = "[yellow]Retry[/yellow]" + + # Use add_row with check for duplicate + try: + table.add_row( + f"[{status_color}]{status}[/{status_color}]", + title, + category, + progress, + started_at, + completed_at, + actions, + key=row_key, + ) + except Exception: + # Row already exists, skip it + pass + + # Focus the table + table.focus() + + def update_stats(self) -> None: + """Update the queue statistics""" + if self.download_queue: + stats = self.download_queue.get_stats() + stats_text = ( + f"Total: {stats['total']} | " + f"Pending: {stats['pending']} | " + f"Downloading: {stats['downloading']} | " + f"Completed: {stats['completed']} | " + f"Cancelled: {stats['cancelled']} | " + f"Failed: {stats['failed']}" + ) + + stats_container = self.query_one("#stats-container", Static) + stats_container.update(f"[bold]{stats_text}[/bold]") + + def update_status(self, message: str) -> None: + """Update the status bar message""" + status_bar = self.query_one("#status-bar", Static) + status_bar.update(f"[bold white]{message}[/bold white]") + + def action_download_selected(self) -> None: + """Download selected item immediately""" + table = self.query_one("#queue-table", DataTable) + selected_row = table.cursor_row + + if selected_row < 0: + self.update_status("[yellow]Select an item to download[/yellow]") + return + + items = self.download_queue.get_queue() if self.download_queue else [] + if selected_row >= len(items): + self.update_status("[yellow]Invalid selection[/yellow]") + return + + item = items[selected_row] + + if item.status == QueueStatus.PENDING: + # Start the download manager if not running + if self.download_manager and not self.download_manager.is_processing(): + self.download_manager.start() + + # Force immediate download by moving item to top + # In a real implementation, we'd have a priority queue + self.update_status( + f"[blue]Downloading: {item.video.display_title if item.video else 'Unknown'}[/blue]" + ) + else: + self.update_status("[yellow]Only pending items can be downloaded[/yellow]") + + def action_retry_selected(self) -> None: + """Retry selected failed item""" + table = self.query_one("#queue-table", DataTable) + selected_row = table.cursor_row + + if selected_row < 0: + self.update_status("[yellow]Select an item to retry[/yellow]") + return + + items = self.download_queue.get_queue() if self.download_queue else [] + if selected_row >= len(items): + return + + item = items[selected_row] + + # Only allow retrying failed items + if item.status != QueueStatus.FAILED: + self.update_status("[yellow]Only failed items can be retried[/yellow]") + return + + if item.video: + # Reset the item status to PENDING + item.status = QueueStatus.PENDING + item.started_at = None + item.completed_at = None + item.progress = 0 + + # Re-add to download queue + if self.download_queue.add_video(item.video, item.category): + self.update_status( + f"[green]Retrying: {item.video.display_title}[/green]" + ) + self.app.notify( + f"Retrying: {item.video.display_title}", + title="Queue", + severity="information", + timeout=3, + ) + else: + self.update_status("[red]Failed to retry item[/red]") + else: + self.update_status("[red]Invalid item for retry[/red]") + + self.update_table() + self.update_stats() + + def action_remove_selected(self) -> None: + """Remove selected item from queue""" + table = self.query_one("#queue-table", DataTable) + selected_row = table.cursor_row + + if selected_row < 0: + self.update_status("[yellow]Select an item to remove[/yellow]") + return + + items = self.download_queue.get_queue() if self.download_queue else [] + if selected_row >= len(items): + return + + item = items[selected_row] + + if item.video: + if self.download_queue.remove_video(item.video.video_id): + self.update_status( + f"[green]Removed: {item.video.display_title}[/green]" + ) + self.app.notify( + f"Removed: {item.video.display_title}", + title="Queue", + severity="information", + timeout=3, + ) + else: + self.update_status("[red]Failed to remove item[/red]") + + self.update_table() + self.update_stats() + + def action_clear_completed(self) -> None: + """Clear completed and cancelled items from queue""" + if self.download_queue: + removed = self.download_queue.clear_completed() + if removed > 0: + self.update_status( + f"[green]Cleared {removed} completed/cancelled items[/green]" + ) + self.app.notify( + f"Cleared {removed} items", + title="Queue", + severity="information", + timeout=3, + ) + else: + self.update_status( + "[yellow]No completed/cancelled items to clear[/yellow]" + ) + + self.update_table() + self.update_stats() + + def action_clear_failed(self) -> None: + """Clear failed items from queue""" + if self.download_queue: + removed = self.download_queue.clear_failed() + if removed > 0: + self.update_status(f"[green]Cleared {removed} failed items[/green]") + self.app.notify( + f"Cleared {removed} items", + title="Queue", + severity="information", + timeout=3, + ) + else: + self.update_status("[yellow]No failed items to clear[/yellow]") + + self.update_table() + self.update_stats() + + def action_go_back(self) -> None: + """Go back to results screen""" + self.app.pop_screen() + + def action_quit(self) -> None: + """Quit to search screen""" + self.app.pop_screen() + + def on_button_pressed(self, event: Button.Pressed) -> None: + """Handle button presses""" + if event.button.id == "back-btn": + self.action_go_back() + elif event.button.id == "refresh-btn": + self.action_refresh_screen() + elif event.button.id == "remove-btn": + self.action_remove_selected() + elif event.button.id == "retry-btn": + self.action_retry_selected() + elif event.button.id == "clear-done-btn": + self.action_clear_completed() + elif event.button.id == "clear-failed-btn": + self.action_clear_failed() + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + """Handle row selection""" + # Get the queue item that was selected + row_key = event.row_key + row_index = int(row_key.value) - 1 if row_key else -1 # type: ignore[arg-type] + + if self.download_queue: + items = self.download_queue.get_queue() + if 0 <= row_index < len(items): + item = items[row_index] + if item.video: + self.update_status(f"Selected: {item.video.display_title}") diff --git a/screens/results.py b/screens/results.py new file mode 100644 index 0000000..2836883 --- /dev/null +++ b/screens/results.py @@ -0,0 +1,364 @@ +#!/usr/bin/env python3 +""" +Results Screen for YouTube TUI +""" + +import asyncio +from typing import List + +from textual.app import ComposeResult +from textual.containers import Container +from textual.screen import Screen +from textual.widgets import ( + Button, + DataTable, + Footer, + Header, + Static, +) + +from youtube_tui.models.video import Video +from youtube_tui.screens.modal import CategorySelectionModal +from youtube_tui.services.youtube import YouTubeService + + +class ResultsScreen(Screen): + """Screen for displaying search results""" + + ALLOW_SELECT = True + + CSS = """ + ResultsScreen { + align: center middle; + } + + #results-container { + width: 95%; + height: 70%; + border: solid #555555; + margin: 1 0; + } + + #results-title { + width: 100%; + height: 3; + dock: top; + background: $surface; + content-align: center middle; + } + + #pagination-controls { + width: 100%; + height: 3; + dock: bottom; + background: $surface; + content-align: center middle; + } + + #status-bar { + dock: bottom; + height: 1; + background: $surface; + color: $text-muted; + padding: 0 1; + } + + Button { + width: 15; + margin: 0 1; + } + + DataTable { + width: 100%; + height: 100%; + } + + DataTable .datatable-row-highlight { + background: $primary; + } + + DataTable .datatable-header { + background: $primary-darken-2; + } + + .results-table { + background: $surface; + border: round #666; + } + + .results-table .datatable-row:hover { + background: $primary-lighten-2; + } + + .results-table .datatable-row-selected { + background: $primary; + } + + .results-table .datatable-row-active { + background: $primary-darken-2; + } + """ + + BINDINGS = [ + ("n", "next_page", "Next Page"), + ("p", "previous_page", "Previous Page"), + ("q", "go_back", "Back"), + ("enter", "download", "Download"), + ("escape", "go_back", "Back"), + ("ctrl+r", "refresh_screen", "Refresh"), + ("ctrl+f", "search_from_anywhere", "Search"), + ] + + def __init__(self, search_term: str, page: int = 1): + super().__init__() + self.youtube_service = YouTubeService() + self.search_term = search_term + self.page = page + self.videos: List[Video] = [] + self.total_pages: int = 1 + self.max_per_page: int = 15 + + def compose(self) -> ComposeResult: + """Compose the results screen""" + yield Header() + yield Static( + f"Results for: [bold cyan]{self.search_term}[/bold cyan] (Page {self.page})", + id="results-title", + ) + table = DataTable( + id="results-table", + show_cursor=True, + cursor_type="row", + show_row_labels=False, + classes="results-table", + ) + yield Container( + table, + id="results-container", + ) + yield Container( + Button("← Prev", id="prev-btn"), + Static(id="page-indicator"), + Button("Next →", id="next-btn"), + id="pagination-controls", + ) + yield Static(id="status-bar") + yield Footer() + + def on_mount(self) -> None: + """Called when screen is mounted""" + self.query_one("#results-table", DataTable).focus() + # Use asyncio.create_task to run the async load_results method + # since on_mount is synchronous but we need to fetch data asynchronously + self.load_task = asyncio.create_task(self.load_results()) + self.update_status("[blue]Loading search results...[/blue]") + + def action_refresh_screen(self) -> None: + """Refresh the screen""" + # Cancel any existing load task and start a new one + if hasattr(self, "load_task") and self.load_task: + self.load_task.cancel() + self.load_task = asyncio.create_task(self.load_results()) + self.update_status("[blue]Refreshing results...[/blue]") + + def action_search_from_anywhere(self) -> None: + """Open search from anywhere""" + # Use the app's action_open_search if available, otherwise go back + if hasattr(self.app, "action_open_search"): + self.app.action_open_search() + else: + self.app.pop_screen() + + async def load_results(self) -> None: + """Load search results from YouTube""" + # Clear stale results immediately to prevent mixing with new search + self.videos = [] + self.total_pages = 1 + self.update_table() + + try: + self.videos = await self.youtube_service.search_videos( + self.search_term, page=self.page, per_page=self.max_per_page + ) + + # Calculate total pages (simplified - yt-dlp returns 15 per page) + if len(self.videos) == self.max_per_page: + self.total_pages = self.page + 1 # There might be more pages + else: + self.total_pages = self.page + + # Update the table + self.update_table() + + # Update pagination controls + self.update_pagination() + + except Exception as e: + self.update_status(f"[red]Error loading results: {e}[/red]") + self.videos = [] + + def update_table(self) -> None: + """Update the DataTable with videos""" + table = self.query_one("#results-table", DataTable) + + # Clear existing data + table.clear(columns=True) + + # Set up columns + table.add_columns("#", "Title", "Author", "Duration", "Type") + + # Add rows + for i, video in enumerate(self.videos, 1): + # Determine video type + if video.is_short: + video_type = "Short" + elif "/playlist" in video.url: + video_type = "Playlist" + else: + video_type = "Video" + + # Truncate long titles + title = video.display_title + if len(title) > 50: + title = title[:47] + "..." + + author = video.channel + if len(author) > 20: + author = author[:17] + "..." + + table.add_row( + str(i), + title, + author, + video.display_duration, + video_type, + key=video.video_id, + ) + + # Focus the table + table.focus() + + def update_pagination(self) -> None: + """Update pagination controls""" + page_indicator = self.query_one("#page-indicator", Static) + page_indicator.update(f"Page {self.page} of {self.total_pages}") + + prev_btn = self.query_one("#prev-btn", Button) + next_btn = self.query_one("#next-btn", Button) + + # Disable previous button on first page + prev_btn.disabled = self.page <= 1 + + # Disable next button if we're on the last known page and have fewer results + if len(self.videos) < self.max_per_page: + next_btn.disabled = True + else: + next_btn.disabled = False + + def update_status(self, message: str) -> None: + """Update the status bar message""" + status_bar = self.query_one("#status-bar", Static) + status_bar.update(f"[bold white]{message}[/bold white]") + + def action_add_to_queue(self) -> None: + """Add selected video to queue""" + table = self.query_one("#results-table", DataTable) + selected_row = table.cursor_row + + if selected_row < 0 or selected_row >= len(self.videos): + self.update_status("[yellow]Select a video to add to queue[/yellow]") + return + + video = self.videos[selected_row] + + # Show modal for category selection + self.app.push_screen( + CategorySelectionModal(), + lambda category: self._add_to_queue_with_category(video, category), + ) + + def action_download(self) -> None: + """Download selected video - add to queue""" + self.action_add_to_queue() + + def action_next_page(self) -> None: + """Go to next page""" + if self.page < self.total_pages or len(self.videos) >= self.max_per_page: + self.page += 1 + # Cancel any existing load task and start a new one + if hasattr(self, "load_task") and self.load_task: + self.load_task.cancel() + self.load_task = asyncio.create_task(self.load_results()) + self.update_status(f"[blue]Loading page {self.page}...[/blue]") + + def action_previous_page(self) -> None: + """Go to previous page""" + if self.page > 1: + self.page -= 1 + # Cancel any existing load task and start a new one + if hasattr(self, "load_task") and self.load_task: + self.load_task.cancel() + self.load_task = asyncio.create_task(self.load_results()) + self.update_status(f"[blue]Loading page {self.page}...[/blue]") + + def action_go_back(self) -> None: + """Go back to search screen""" + self.app.pop_screen() + + def action_quit(self) -> None: + """Quit to search screen""" + self.app.pop_screen() + + def on_button_pressed(self, event: Button.Pressed) -> None: + """Handle button presses""" + if event.button.id == "prev-btn": + self.action_previous_page() + elif event.button.id == "next-btn": + self.action_next_page() + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + """Handle row selection (Enter key) - add to queue""" + self._add_selected_video_to_queue() + + def on_data_table_cell_selected(self, event: DataTable.CellSelected) -> None: + """Handle cell click - add to queue""" + self._add_selected_video_to_queue() + + def _add_to_queue_with_category(self, video: Video, category: str | None) -> None: + """Add video to queue with selected category (callback from modal)""" + if category is None: + self.update_status("[yellow]Category selection cancelled[/yellow]") + return + + try: + if hasattr(self.app, "download_queue") and self.app.download_queue: + self.app.download_queue.add_video(video, category=category) + self.update_status( + f"[green]Added to queue: {video.display_title}[/green]" + ) + self.app.notify( + f"Added to queue: {video.display_title}", + title="Queue", + severity="information", + timeout=3, + ) + else: + self.update_status("[yellow]Queue not available[/yellow]") + except Exception as e: + self.update_status(f"[red]Error adding to queue: {e}[/red]") + + def _add_selected_video_to_queue(self) -> None: + """Helper method to add selected video to queue""" + table = self.query_one("#results-table", DataTable) + row_index = table.cursor_row + + if row_index < 0 or row_index >= len(self.videos): + return + + video = self.videos[row_index] + + # Show modal for category selection + self.app.push_screen( + CategorySelectionModal(), + lambda category: self._add_to_queue_with_category(video, category), + ) diff --git a/screens/search.py b/screens/search.py new file mode 100644 index 0000000..fb3e013 --- /dev/null +++ b/screens/search.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +""" +Search Screen for YouTube TUI +""" + +from textual.app import ComposeResult +from textual.containers import Container +from textual.screen import Screen +from textual.widgets import ( + Button, + Footer, + Header, + Input, + Static, +) + +from youtube_tui.services.youtube import YouTubeService + + +class SearchScreen(Screen): + """Screen for searching YouTube videos""" + + CSS = """ + SearchScreen { + align: center middle; + } + + #search-container { + width: 80%; + height: auto; + border: double #555555; + padding: 1 2; + margin: 2 0; + } + + #search-input { + width: 100%; + margin: 1 0; + } + + #buttons { + width: 100%; + height: auto; + dock: bottom; + } + + Button { + width: 20; + margin: 1 1; + } + + #status-bar { + dock: bottom; + height: 1; + background: #333333; + color: #aaaaaa; + padding: 0 1; + } + + #loading-indicator { + height: 3; + margin: 1 0; + } + """ + + BINDINGS = [ + ("escape", "go_back", "Back"), + ("enter", "search", "Search"), + ("ctrl+f", "search_from_anywhere", "Search"), + ] + + def __init__(self): + super().__init__() + self.youtube_service = YouTubeService() + self.search_term = "" + self.is_searching = False + + def compose(self) -> ComposeResult: + """Compose the search screen""" + yield Header() + yield Container( + Static("YouTube Search", id="search-title", classes="title"), + Static("Enter a search term to find YouTube videos", id="search-hint"), + Input(placeholder="Search for videos...", id="search-input"), + Button("Search", id="search-button"), + Button("Cancel", id="cancel-button"), + id="search-container", + ) + yield Static(id="status-bar") + yield Footer() + + def on_mount(self) -> None: + """Called when screen is mounted""" + self.query_one(Input).focus() + self.update_status( + "Ready to search - Press Enter to search, Ctrl+F to search from anywhere" + ) + + def action_search_from_anywhere(self) -> None: + """Open search from anywhere""" + # This is a fallback for the key binding + self.action_search() + + def update_status(self, message: str) -> None: + """Update the status bar message""" + status_bar = self.query_one("#status-bar", Static) + status_bar.update(f"[bold white]{message}[/bold white]") + + def action_search(self) -> None: + """Perform the search operation""" + input_widget = self.query_one(Input) + search_term = input_widget.value.strip() + + if not search_term: + self.update_status("[red]Please enter a search term[/red]") + return + + self.search_term = search_term + self.update_status(f"[blue]Searching for: {search_term}[/blue]") + + # Trigger search - get the app and push results screen + app = self.app + if hasattr(app, "push_results_screen"): + app.push_results_screen(search_term) + else: + # Fallback if app methods aren't available + from youtube_tui.screens.results import ResultsScreen + + self.app.push_screen(ResultsScreen(search_term)) + + def action_cancel(self) -> None: + """Handle cancel action""" + self.action_go_back() + + def action_go_back(self) -> None: + """Go back to previous screen""" + self.app.pop_screen() + + def action_quit(self) -> None: + """Quit the application""" + # Only quit if we're at the root level + if len(self.app.screen_stack) <= 2: # Header + Footer + Screen + self.app.exit() + else: + self.action_go_back() + + def on_button_pressed(self, event: Button.Pressed) -> None: + """Handle button presses""" + if event.button.id == "search-button": + self.action_search() + elif event.button.id == "cancel-button": + self.action_quit() + + def on_input_submitted(self, event: Input.Submitted) -> None: + """Handle enter key in search input""" + self.action_search() diff --git a/services/__init__.py b/services/__init__.py new file mode 100644 index 0000000..c8ac248 --- /dev/null +++ b/services/__init__.py @@ -0,0 +1,7 @@ +""" +Services package for YouTube TUI +""" + +from youtube_tui.services.youtube import YouTubeService + +__all__ = ["YouTubeService"] diff --git a/services/download_manager.py b/services/download_manager.py new file mode 100644 index 0000000..1bc45ca --- /dev/null +++ b/services/download_manager.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +""" +Download Manager for YouTube TUI +Handles background downloads sequentially +""" + +import asyncio +import logging +from logging.handlers import RotatingFileHandler +from pathlib import Path +from typing import Optional + +from rich.console import Console + +from youtube_tui.models.queue_item import QueueItem, QueueStatus +from youtube_tui.models.video import Video +from youtube_tui.services.queue import DownloadQueue +from youtube_tui.services.youtube import YouTubeService + +console = Console() + +# Configure logging +LOG_DIR = Path.home() / ".config" / "youtube_cli" / "logs" +LOG_DIR.mkdir(parents=True, exist_ok=True) +LOG_FILE = LOG_DIR / "app.log" + +# Use RotatingFileHandler for log rotation (10MB, 5 backups) +file_handler = RotatingFileHandler(LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5) +file_handler.setLevel(logging.DEBUG) +file_handler.setFormatter( + logging.Formatter( + "%(asctime)s | %(name)s | %(levelname)s | %(message)s", + "%Y-%m-%d %H:%M:%S", + ) +) + +# Create console handler +console_handler = logging.StreamHandler() +console_handler.setLevel(logging.INFO) +console_handler.setFormatter(logging.Formatter("%(message)s")) + +# Configure root logger +logging.basicConfig( + level=logging.DEBUG, + handlers=[ + file_handler, + console_handler, + ], +) +logger = logging.getLogger(__name__) + + +class DownloadManager: + """Manages background downloads from the queue""" + + def __init__(self, queue: DownloadQueue, youtube_service: YouTubeService): + self._queue: DownloadQueue = queue + self._active_task: Optional[asyncio.Task] = None + self._current_item: Optional[QueueItem] = None + self._is_running = False + self._cancel_requested = False + self._youtube_service = youtube_service + + def add_to_queue( + self, + video: Video, + category: Optional[str] = None, + network_folder: Optional[str] = None, + ) -> QueueItem: + """Add a video to the download queue""" + return self._queue.add_video(video, category, network_folder) + + def remove_from_queue(self, item_id: str) -> bool: + """Remove an item from the queue by UUID string""" + return self._queue.remove_item(item_id) + + def cancel_active_download(self) -> None: + """Cancel the currently active download""" + self._cancel_requested = True + if self._active_task: + self._active_task.cancel() + + def get_queue_status(self) -> dict: + """Get queue status information""" + stats = self._queue.get_stats() + # Use _current_item to determine if there's an active download + has_active_download = self._current_item is not None + return { + "pending_count": stats["pending"], + "downloading_count": 1 if has_active_download else stats["downloading"], + "total_count": stats["total"], + "has_active_download": has_active_download, + "active_item": self._current_item.to_dict() if self._current_item else None, + } + + def get_active_item(self) -> Optional[QueueItem]: + """Get the currently downloading item""" + return self._current_item + + def start_processing(self) -> None: + """Start the background download processing task""" + if not self._is_running: + self._is_running = True + self._active_task = asyncio.create_task(self._process_queue()) + + def stop_processing(self) -> None: + """Stop the background download processing task""" + self._is_running = False + if self._active_task: + self._active_task.cancel() + + async def _process_queue(self) -> None: + """Process the download queue sequentially""" + # Loop is available via asyncio.run() in main context + + while self._is_running: + try: + # Check if we have a pending item + item = self._queue.get_next_pending() + if item is None: + await asyncio.sleep(1) # Wait for new items + continue + + # Mark item as current + self._current_item = item + self._cancel_requested = False + + # Start downloading + await self._download_item(item) + + # Clear current item after completion + self._current_item = None + + except asyncio.CancelledError: + # Task was cancelled + logger.warning("Download manager cancelled") + break + except Exception as e: + logger.warning(f"Error in download manager: {e}") + await asyncio.sleep(1) + + async def _download_item(self, item: QueueItem) -> None: + """Download a single queue item""" + # Update status to downloading + item.start_download() + if item.video: + self._queue.update_item_status(str(item.id), QueueStatus.DOWNLOADING) + + try: + # Determine if it's a playlist or video + is_playlist = item.video and ( + "/playlist" in item.video.url.lower() + or "list=" in item.video.url.lower() + ) + + # Download with progress callback + async def progress_callback(percentage: int) -> bool: + """Progress callback that checks for cancellation""" + # Update progress + if item.video: + self._queue.update_progress(str(item.id), percentage) + item.update_progress(percentage) + + # Check for cancellation + if self._cancel_requested: + raise asyncio.CancelledError("Download cancelled by user") + + return True + + if is_playlist and item.video: + success = await self._youtube_service.download_playlist( + item.video, + category=item.category, + network_folder=item.network_folder, + progress_callback=progress_callback, + ) + elif item.video: + success = await self._youtube_service.download_video( + item.video, + category=item.category, + network_folder=item.network_folder, + progress_callback=progress_callback, + ) + else: + # No video to download + if item.video is None: + item.fail(error_message="No video data available") + success = False + + # Check final status + if self._cancel_requested: + # Download was cancelled + if item.video: + self._queue.update_item_status(str(item.id), QueueStatus.CANCELLED) + item.cancel() + elif success: + # Download succeeded + if item.video: + self._queue.update_item_status(str(item.id), QueueStatus.COMPLETED) + item.complete() + else: + # Download failed + if item.video: + self._queue.update_item_status(str(item.id), QueueStatus.FAILED) + item.fail(error_message="Download failed") + + except asyncio.CancelledError: + # Task was cancelled + if item.video: + self._queue.update_item_status(str(item.id), QueueStatus.CANCELLED) + item.cancel() + except Exception as e: + logger.error(f"Download error: {e}") + if item.video: + self._queue.update_item_status(str(item.id), QueueStatus.FAILED) + item.fail(error_message=str(e)) + + def is_processing(self) -> bool: + """Check if download manager is processing queue""" + return self._is_running diff --git a/services/queue.py b/services/queue.py new file mode 100644 index 0000000..aa8bf07 --- /dev/null +++ b/services/queue.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +""" +Download Queue Service for YouTube TUI +Manages the queue of videos to download +""" + +import json +import logging +from logging.handlers import RotatingFileHandler +from pathlib import Path +from typing import List, Optional + +from rich.console import Console + +from youtube_tui.models.queue_item import QueueItem, QueueStatus +from youtube_tui.models.video import Video + +console = Console() + +# Configure logging +LOG_DIR = Path.home() / ".config" / "youtube_cli" / "logs" +LOG_DIR.mkdir(parents=True, exist_ok=True) +LOG_FILE = LOG_DIR / "app.log" + +# Use RotatingFileHandler for log rotation (10MB, 5 backups) +file_handler = RotatingFileHandler(LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5) +file_handler.setLevel(logging.DEBUG) +file_handler.setFormatter( + logging.Formatter( + "%(asctime)s | %(name)s | %(levelname)s | %(message)s", + "%Y-%m-%d %H:%M:%S", + ) +) + +# Create console handler +console_handler = logging.StreamHandler() +console_handler.setLevel(logging.INFO) +console_handler.setFormatter(logging.Formatter("%(message)s")) + +# Configure root logger +logging.basicConfig( + level=logging.DEBUG, + handlers=[ + file_handler, + console_handler, + ], +) +logger = logging.getLogger(__name__) + + +class DownloadQueue: + """Manages the download queue""" + + _archive_file = Path.home() / ".config" / "youtube_cli" / "download_queue.json" + + def __init__(self) -> None: + self._queue: List[QueueItem] = [] + self._load_queue() + + def _load_queue(self) -> None: + """Load queue from archive file""" + try: + if self._archive_file.exists(): + with open(self._archive_file, "r") as f: + data = json.load(f) + self._queue = [QueueItem.from_dict(item) for item in data] + except Exception as e: + logger.warning(f"Error loading queue: {e}") + self._queue = [] + + def _save_queue(self) -> None: + """Save queue to archive file""" + try: + self._archive_file.parent.mkdir(parents=True, exist_ok=True) + with open(self._archive_file, "w") as f: + data = [item.to_dict() for item in self._queue] + json.dump(data, f, indent=2) + except Exception as e: + logger.warning(f"Error saving queue: {e}") + + def add_video( + self, + video: Video, + category: Optional[str] = None, + network_folder: Optional[str] = None, + ) -> QueueItem: + """Add a video to the queue""" + item = QueueItem(video=video, category=category, network_folder=network_folder) + self._queue.append(item) + self._save_queue() + return item + + def remove_item(self, item_id: str) -> bool: + """Remove a queue item by its UUID string""" + try: + import uuid + + item_uuid = uuid.UUID(item_id) + except (ValueError, TypeError): + return False + + for i, item in enumerate(self._queue): + if item.id == item_uuid: + del self._queue[i] + self._save_queue() + return True + return False + + def get_next_pending(self) -> Optional[QueueItem]: + """Get the next pending video to download""" + for item in self._queue: + if item.status == QueueStatus.PENDING: + return item + return None + + def update_item_status(self, item_id: str, status: QueueStatus) -> None: + """Update the status of a queue item by UUID string""" + try: + import uuid + + item_uuid = uuid.UUID(item_id) + except (ValueError, TypeError): + return + + for item in self._queue: + if item.id == item_uuid: + item.status = status + self._save_queue() + return + + def update_progress(self, item_id: str, percentage: int) -> None: + """Update the progress of a queue item by UUID string""" + try: + import uuid + + item_uuid = uuid.UUID(item_id) + except (ValueError, TypeError): + return + + for item in self._queue: + if item.id == item_uuid: + item.update_progress(percentage) + self._save_queue() + return + + def get_all_items(self) -> List[QueueItem]: + """Get all queue items""" + return self._queue.copy() + + def get_active_count(self) -> int: + """Get count of active items (pending + downloading)""" + return sum( + 1 + for item in self._queue + if item.status in (QueueStatus.PENDING, QueueStatus.DOWNLOADING) + ) + + def get_pending_count(self) -> int: + """Get count of pending items""" + return sum(1 for item in self._queue if item.status == QueueStatus.PENDING) + + def cancel_item(self, item_id: str) -> None: + """Cancel a queue item by UUID string""" + try: + import uuid + + item_uuid = uuid.UUID(item_id) + except (ValueError, TypeError): + return + + for item in self._queue: + if item.id == item_uuid: + item.cancel() + self._save_queue() + return + + def clear_completed(self) -> int: + """Remove completed and cancelled items from queue""" + initial_count = len(self._queue) + self._queue = [ + item + for item in self._queue + if item.status not in (QueueStatus.COMPLETED, QueueStatus.CANCELLED) + ] + removed = initial_count - len(self._queue) + if removed > 0: + self._save_queue() + return removed + + def clear_failed(self) -> int: + """Remove failed items from queue""" + initial_count = len(self._queue) + self._queue = [ + item for item in self._queue if item.status != QueueStatus.FAILED + ] + removed = initial_count - len(self._queue) + if removed > 0: + self._save_queue() + return removed + + def get_downloading_item(self) -> Optional[QueueItem]: + """Get the currently downloading item""" + for item in self._queue: + if item.status == QueueStatus.DOWNLOADING: + return item + return None + + def remove_video(self, video_id: str) -> bool: + """Remove a video from the queue by video ID (alias for remove_by_video_id)""" + return self.remove_by_video_id(video_id) + + def update_status( + self, video_id: str, status: QueueStatus, progress: Optional[int] = None + ) -> None: + """Update the status of a video in the queue by video ID""" + for item in self._queue: + if item.video and item.video.video_id == video_id: + item.status = status + if progress is not None: + item.update_progress(progress) + self._save_queue() + return + + def cancel_video(self, video_id: str) -> bool: + """Cancel a video in the queue by video ID""" + for item in self._queue: + if item.video and item.video.video_id == video_id: + item.cancel() + self._save_queue() + return True + return False + + def remove_by_video_id(self, video_id: str) -> bool: + """Remove a video from the queue by video ID""" + for i, item in enumerate(self._queue): + if item.video and item.video.video_id == video_id: + del self._queue[i] + self._save_queue() + return True + return False + + def get_queue(self) -> List[QueueItem]: + """Get all queue items (alias for get_all_items)""" + return self.get_all_items() + + def get_stats(self) -> dict: + """Get queue statistics""" + total = len(self._queue) + pending = sum(1 for item in self._queue if item.status == QueueStatus.PENDING) + downloading = sum( + 1 for item in self._queue if item.status == QueueStatus.DOWNLOADING + ) + completed = sum( + 1 for item in self._queue if item.status == QueueStatus.COMPLETED + ) + cancelled = sum( + 1 for item in self._queue if item.status == QueueStatus.CANCELLED + ) + failed = sum(1 for item in self._queue if item.status == QueueStatus.FAILED) + + return { + "total": total, + "pending": pending, + "downloading": downloading, + "completed": completed, + "cancelled": cancelled, + "failed": failed, + } diff --git a/services/youtube.py b/services/youtube.py new file mode 100644 index 0000000..f31c2e1 --- /dev/null +++ b/services/youtube.py @@ -0,0 +1,344 @@ +#!/usr/bin/env python3 +""" +YouTube service wrapper around YouTubeCLI - Async implementation +""" + +import asyncio +import logging +from logging.handlers import RotatingFileHandler +from pathlib import Path +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() + +# Configure logging +LOG_DIR = Path.home() / ".config" / "youtube_cli" / "logs" +LOG_DIR.mkdir(parents=True, exist_ok=True) +LOG_FILE = LOG_DIR / "app.log" + +# Use RotatingFileHandler for log rotation (10MB, 5 backups) +file_handler = RotatingFileHandler(LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5) +file_handler.setLevel(logging.DEBUG) +file_handler.setFormatter( + logging.Formatter( + "%(asctime)s | %(name)s | %(levelname)s | %(message)s", + "%Y-%m-%d %H:%M:%S", + ) +) + +# Create console handler +console_handler = logging.StreamHandler() +console_handler.setLevel(logging.INFO) +console_handler.setFormatter(logging.Formatter("%(message)s")) + +# Configure root logger +logging.basicConfig( + level=logging.DEBUG, + handlers=[ + file_handler, + console_handler, + ], +) +logger = logging.getLogger(__name__) + + +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: + logger.error(f"Error searching videos: {e}") + 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: + logger.error(f"Error downloading video: {e}") + 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: + logger.error(f"Error downloading playlist: {e}") + 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: + logger.error(f"Error adding to archive: {e}") + 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) diff --git a/test_tui.py b/test_tui.py new file mode 100644 index 0000000..53b7ead --- /dev/null +++ b/test_tui.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +""" +Test script for YouTube TUI +""" + +from youtube_tui.app import YouTubeTUI + + +def test_imports(): + """Test that all imports work correctly""" + try: + print("✓ App imported successfully") + + return True + except Exception as e: + print(f"✗ Import failed: {e}") + return False + + +def test_app_creation(): + """Test that the app can be created""" + try: + app = YouTubeTUI() + print("✓ YouTubeTUI created successfully") + print(f" - App version: {app.VERSION}") + print(f" - yt-dlp version: {app.yt_dlp_version}") + print(f" - Search history loaded: {len(app.search_history)} items") + return True + except Exception as e: + print(f"✗ App creation failed: {e}") + return False + + +def test_video_model(): + """Test the Video model""" + try: + from youtube_tui.models.video import Video + + video = Video( + video_id="dQw4w9WgXcQ", + title="Test Video", + channel="Test Channel", + channel_id="UC123", + duration="3:45", + view_count="1000000", + upload_date="20230101", + description="Test description", + ) + + print("✓ Video model created successfully") + print(f" - Video ID: {video.video_id}") + print(f" - Title: {video.title}") + print(f" - Display title: {video.display_title}") + print(f" - Duration: {video.duration}") + print(f" - URL: {video.url}") + + return True + except Exception as e: + print(f"✗ Video model test failed: {e}") + return False + + +def test_search_history(): + """Test search history functionality""" + try: + app = YouTubeTUI() + + # Test adding to history + app.add_to_search_history("test search 1") + app.add_to_search_history("test search 2") + + print("✓ Search history operations successful") + print(f" - History items: {len(app.search_history)}") + + # Test that duplicates are removed + app.add_to_search_history("test search 1") + print(f" - After duplicate: {len(app.search_history)} items") + + return True + except Exception as e: + print(f"✗ Search history test failed: {e}") + return False + + +def test_video_from_dict(): + """Test Video model from_dict method""" + try: + from youtube_tui.models.video import Video + + data = { + "video_id": "abc123", + "title": "Test Video", + "channel": "Test Channel", + "channel_id": "UC123", + "duration": "5:30", + "view_count": "500000", + "upload_date": "20230101", + "description": "Test description", + } + + video = Video.from_dict(data) + + print("✓ Video.from_dict() works correctly") + print(f" - Video ID: {video.video_id}") + print(f" - Title: {video.title}") + + # Test to_dict + video_dict = video.to_dict() + print(f" - to_dict() keys: {list(video_dict.keys())}") + + return True + except Exception as e: + print(f"✗ Video from_dict test failed: {e}") + return False + + +if __name__ == "__main__": + print("=" * 60) + print("YouTube TUI Test Suite") + print("=" * 60) + print() + + tests = [ + ("Imports", test_imports), + ("App Creation", test_app_creation), + ("Video Model", test_video_model), + ("Search History", test_search_history), + ("Video from Dict", test_video_from_dict), + ] + + results = [] + for name, test_func in tests: + print(f"\nTesting: {name}") + print("-" * 40) + result = test_func() + results.append((name, result)) + print() + + print("=" * 60) + print("Test Results Summary") + print("=" * 60) + + passed = sum(1 for _, r in results if r) + total = len(results) + + for name, result in results: + status = "✓ PASS" if result else "✗ FAIL" + print(f"{status}: {name}") + + print() + print(f"Total: {passed}/{total} tests passed") + print("=" * 60) diff --git a/widgets/__init__.py b/widgets/__init__.py new file mode 100644 index 0000000..f5e9cd6 --- /dev/null +++ b/widgets/__init__.py @@ -0,0 +1,11 @@ +""" +Widgets package for YouTube TUI +""" + +from youtube_tui.widgets.command_palette import CommandPalette +from youtube_tui.widgets.status_bar import StatusBar + +__all__ = [ + "StatusBar", + "CommandPalette", +] diff --git a/widgets/command_palette.py b/widgets/command_palette.py new file mode 100644 index 0000000..69a1ad7 --- /dev/null +++ b/widgets/command_palette.py @@ -0,0 +1,199 @@ +""" +Command Palette Widget for YouTube TUI +""" + +from textual.app import ComposeResult +from textual.containers import Container +from textual.screen import ModalScreen +from textual.widgets import Footer, Header, Input, ListView, ListItem, Static + + +class CommandPalette(ModalScreen): + """Command palette for quick access to actions""" + + DEFAULT_COMMANDS = [ + ("Search", "Search for videos"), + ("Download", "Quick download mode"), + ("History", "Show search history"), + ("Settings", "Open settings"), + ("Help", "Show help screen"), + ("Quit", "Quit application"), + ] + + CSS = """ + CommandPalette { + align: center middle; + } + + #palette-container { + width: 60%; + height: auto; + border: solid #555555; + background: $surface; + padding: 1; + } + + #palette-title { + width: 100%; + height: 3; + dock: top; + background: $primary; + content-align: center middle; + color: $text; + } + + #palette-input { + width: 100%; + margin: 1 0; + } + + #palette-list { + width: 100%; + height: 20; + margin: 1 0; + } + + #palette-actions { + width: 100%; + height: auto; + dock: bottom; + margin-top: 1; + } + + ListItem { + height: 3; + padding: 0 1; + } + + ListItem:hover { + background: $primary-darken-2; + } + + ListItem.--highlight { + background: $primary; + } + + .command-description { + color: $text-muted; + } + """ + + BINDINGS = [ + ("escape", "close_palette", "Close"), + ("up", "cursor_up", "Cursor Up"), + ("down", "cursor_down", "Cursor Down"), + ("enter", "select_command", "Select"), + ] + + def __init__(self): + super().__init__() + self.selected_index = 0 + self.commands = self.DEFAULT_COMMANDS.copy() + + def compose(self) -> ComposeResult: + """Compose the command palette""" + yield Header() + yield Container( + Static("Command Palette", id="palette-title"), + Input(placeholder="Type to filter commands...", id="palette-input"), + ListView(id="palette-list-view"), + id="palette-container", + ) + yield Footer() + + def on_mount(self) -> None: + """Called when palette is mounted""" + self.update_list() + self.query_one(Input).focus() + + def update_list(self) -> None: + """Update the command list""" + input_widget = self.query_one("#palette-input", Input) + filter_text = input_widget.value.lower() + + # Filter commands + if filter_text: + self.commands = [ + (cmd, desc) + for cmd, desc in self.DEFAULT_COMMANDS + if filter_text in cmd.lower() or filter_text in desc.lower() + ] + else: + self.commands = self.DEFAULT_COMMANDS.copy() + + # Update the list view + list_view = self.query_one("#palette-list-view", ListView) + list_view.clear() + + for i, (command, description) in enumerate(self.commands): + item = ListItem( + Static(f"[bold]{command}[/bold]\n{description}"), + ) + list_view.append(item) + + # Update selected index if needed + if self.selected_index >= len(self.commands): + self.selected_index = max(0, len(self.commands) - 1) + + # Highlight selected item + self.highlight_selected() + + def highlight_selected(self) -> None: + """Highlight the selected item""" + list_view = self.query_one("#palette-list-view", ListView) + + for i, item in enumerate(list_view.children): + if i == self.selected_index: + item.add_class("--highlight") + else: + item.remove_class("--highlight") + + def action_cursor_up(self) -> None: + """Move selection up""" + if self.commands: + self.selected_index = max(0, self.selected_index - 1) + self.highlight_selected() + + def action_cursor_down(self) -> None: + """Move selection down""" + if self.commands: + self.selected_index = min(len(self.commands) - 1, self.selected_index + 1) + self.highlight_selected() + + def action_select_command(self) -> None: + """Execute the selected command""" + if not self.commands: + return + + command = self.commands[self.selected_index][0] + self.execute_command(command) + + def execute_command(self, command: str) -> None: + """Execute a command""" + if command == "Search": + self.app.push_screen("search") + self.app.pop_screen() + elif command == "Download": + # Go to search for quick download + self.app.push_screen("search") + self.app.pop_screen() + elif command == "History": + self.app.push_screen("history") + elif command == "Settings": + self.app.push_screen("settings") + elif command == "Help": + self.app.push_screen("help") + elif command == "Quit": + self.app.exit() + + def action_close_palette(self) -> None: + """Close the palette""" + self.app.pop_screen() + + def on_input_changed(self, event: Input.Changed) -> None: + """Handle input changes""" + self.update_list() + + def on_data_table_row_selected(self, event) -> None: + """Handle row selection""" + self.action_select_command() diff --git a/widgets/footer.py b/widgets/footer.py new file mode 100644 index 0000000..1359804 --- /dev/null +++ b/widgets/footer.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +""" +Custom Footer Widget for YouTube TUI +Includes status bar with queue information +""" + +from datetime import datetime + +from textual.widgets import Footer +from textual.app import App + + +class CustomFooter(Footer): + """Custom footer widget with status bar that includes queue info""" + + def __init__(self, app: App, *args, **kwargs): + super().__init__(*args, **kwargs) + self._app = app + self.current_screen = "Search" + self.status_message = "Ready" + self.downloading = False + self.queue_pending = 0 + self.download_progress = 0 + self.yt_dlp_version = "unknown" + self.update_version() + + def update_version(self) -> None: + """Update yt-dlp version""" + try: + import subprocess + + result = subprocess.run( + ["yt-dlp", "--version"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0: + self.yt_dlp_version = result.stdout.strip() + else: + self.yt_dlp_version = "not installed" + except Exception: + self.yt_dlp_version = "unknown" + + def set_screen(self, screen_name: str) -> None: + """Set the current screen name""" + self.current_screen = screen_name + self.refresh() + + def set_downloading(self, downloading: bool) -> None: + """Set downloading state""" + self.downloading = downloading + self.refresh() + + def set_status(self, message: str) -> None: + """Set status message""" + self.status_message = message + self.refresh() + + def set_queue_pending(self, count: int) -> None: + """Set the number of pending queue items""" + self.queue_pending = count + self.refresh() + + def set_download_progress(self, progress: int) -> None: + """Set the current download progress percentage""" + self.download_progress = max(0, min(100, progress)) + self.refresh() + + def update_time(self) -> None: + """Update the time display""" + self.refresh() + + def render(self): + """Render the footer content with queue status""" + # Get current time + current_time = datetime.now().strftime("%H:%M:%S") + + # Get theme info + theme_name = getattr(self._app, "theme", "css") + + # Build status string + status_parts = [ + f"[bold]{self.current_screen}[/bold]", + f"v{getattr(self._app, 'VERSION', '0.1.0')}", + f"yt-dlp {self.yt_dlp_version}", + ] + + # Add queue status if available + if self.queue_pending > 0: + status_parts.append(f"[cyan]Queue: {self.queue_pending} pending[/cyan]") + + # Add download progress if available + if self.downloading and self.download_progress > 0: + status_parts.append( + f"| [yellow]Downloading: {self.download_progress}%[/yellow]" + ) + + # Add status message + status_parts.append(f"[bold]{self.status_message}[/bold]") + + # Add footer elements + status_parts.append(f"[dim]{current_time}[/dim]") + status_parts.append(f"[dim]{theme_name} theme[/dim]") + + return " ".join(status_parts) diff --git a/widgets/status_bar.py b/widgets/status_bar.py new file mode 100644 index 0000000..85ca2a0 --- /dev/null +++ b/widgets/status_bar.py @@ -0,0 +1,119 @@ +""" +Status Bar Widget for YouTube TUI +Simple status bar widget (custom footer handles queue status) +""" + +from datetime import datetime + +from textual.app import App +from textual.widgets import Static + + +class StatusBar(Static): + """Simple status bar widget for YouTube TUI""" + + def __init__(self, app: App, *args, **kwargs): + super().__init__(*args, **kwargs) + # Store app as _app since app is a read-only property in Static + self._app = app + self.current_screen = "Search" + self.status_message = "Ready" + self.downloading = False + self.queue_pending = 0 + self.download_progress = 0 + self.yt_dlp_version = "unknown" + self.update_version() + # Note: set_interval requires active app context, so we skip it in tests + # The timer functionality is tested separately if needed + try: + self.set_interval(1, self.update_time) + except Exception: + # Timer not available in test context, skip + pass + + def update_version(self) -> None: + """Update yt-dlp version""" + try: + import subprocess + + result = subprocess.run( + ["yt-dlp", "--version"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0: + self.yt_dlp_version = result.stdout.strip() + else: + self.yt_dlp_version = "not installed" + except Exception: + self.yt_dlp_version = "unknown" + + def set_screen(self, screen_name: str) -> None: + """Set the current screen name""" + self.current_screen = screen_name + self.refresh() + + def set_status(self, message: str) -> None: + """Set status message""" + self.status_message = message + self.refresh() + + def set_downloading(self, downloading: bool) -> None: + """Set downloading state""" + self.downloading = downloading + self.refresh() + + def set_queue_pending(self, count: int) -> None: + """Set the number of pending queue items""" + self.queue_pending = count + self.refresh() + + def set_download_progress(self, progress: int) -> None: + """Set the current download progress percentage""" + self.download_progress = max(0, min(100, progress)) + self.refresh() + + def update_time(self) -> None: + """Update the time display""" + self.refresh() + + def render(self): + """Render the status bar content""" + # Get current time + current_time = datetime.now().strftime("%H:%M:%S") + + # Get theme info + theme_name = getattr(self._app, "theme", "css") + + # Build status string + status_parts = [ + f"[bold]{self.current_screen}[/bold]", + f"v{getattr(self._app, 'VERSION', '0.1.0')}", + f"yt-dlp {self.yt_dlp_version}", + ] + + # Add download indicator if downloading + if self.downloading: + status_parts.append("[bold green]↓[/bold green]") + + # Add queue status if there are pending items + if self.queue_pending > 0: + status_parts.append( + f"[bold cyan]Queue: {self.queue_pending} pending[/bold cyan]" + ) + + # Add download progress if downloading + if self.downloading and self.download_progress > 0: + status_parts.append( + f"[bold yellow]Downloading: {self.download_progress}%[/bold yellow]" + ) + + # Add status message + status_parts.append(f"[bold]{self.status_message}[/bold]") + + # Add footer elements + status_parts.append(f"[dim]{current_time}[/dim]") + status_parts.append(f"[dim]{theme_name} theme[/dim]") + + return " ".join(status_parts)