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