#!/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}")