360 lines
11 KiB
Python
360 lines
11 KiB
Python
#!/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"""
|
|
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),
|
|
)
|