jarianc/youtube-tui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: youtube-tui (extracted from youtube-cli)

Browse Source
This commit is contained in:
Jarian Cottingham 2026-08-21 17:08:28 +00:00
commit 17f1413d5e
28 changed files with 4073 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored Normal file
View File

@ -0,0 +1,4 @@
__pycache__/
*.pyc
.pytest_cache/
*.egg-info/

21
LICENSE Normal file
View File

@ -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.

35
README.md Normal file
View File

@ -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

5
__init__.py Normal file
View File

@ -0,0 +1,5 @@
"""
YouTube TUI - A Text-based User Interface for browsing and downloading YouTube videos
"""
__version__ = "0.1.0"

9
__main__.py Normal file
View File

@ -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()

296
app.py Normal file
View File

@ -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()

7
models/__init__.py Normal file
View File

@ -0,0 +1,7 @@
"""
Models package for YouTube TUI
"""
from youtube_tui.models.video import Video
__all__ = ["Video"]

144
models/queue_item.py Normal file
View File

@ -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

77
models/video.py Normal file
View File

@ -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", ""),
)

59
pyproject.toml Normal file
View File

@ -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"

19
screens/__init__.py Normal file
View File

@ -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",
]

204
screens/download.py Normal file
View File

@ -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,
)

183
screens/help.py Normal file
View File

@ -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()

186
screens/history.py Normal file
View File

@ -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)

192
screens/home.py Normal file
View File

@ -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()

253
screens/modal.py Normal file
View File

@ -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()

433
screens/queue.py Normal file
View File

@ -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}")

364
screens/results.py Normal file
View File

@ -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),
)

156
screens/search.py Normal file
View File

@ -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()

7
services/__init__.py Normal file
View File

@ -0,0 +1,7 @@
"""
Services package for YouTube TUI
"""
from youtube_tui.services.youtube import YouTubeService
__all__ = ["YouTubeService"]

220
services/download_manager.py Normal file
View File

@ -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

268
services/queue.py Normal file
View File

@ -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,
}

344
services/youtube.py Normal file
View File

@ -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)

152
test_tui.py Normal file
View File

@ -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)

11
widgets/__init__.py Normal file
View File

@ -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",
]

199
widgets/command_palette.py Normal file
View File

@ -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()

106
widgets/footer.py Normal file
View File

@ -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)

119
widgets/status_bar.py Normal file
View File

@ -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)