multi video support, yt-dlp fix

This commit is contained in:
Jarian Cottingham 2025-12-27 00:00:01 -06:00
parent cd4e2c56ba
commit fcbf07b141
5 changed files with 3137 additions and 335 deletions

View File

@ -1,14 +0,0 @@
{
"download_dir": "~/Downloads/youtube",
"default_locations": [
"~/Downloads/youtube",
"~/Movies/youtube",
"/tmp/youtube"
],
"max_videos_per_page": 15,
"yt_dlp_args": {
"format": "bestvideo[height=1080]+bestaudio/bestvideo[height<=1080]+bestaudio",
"write_thumbnail": true,
"extractor_args": "youtube:player-client=default,-tv_simply"
}
}

36
structure.md Normal file
View File

@ -0,0 +1,36 @@
# YouTube CLI Project Structure
## Root Directory Files
- **.gitignore** - Git ignore file to exclude unnecessary files from version control
- **README.md** - Project documentation and usage instructions
- **requirements.txt** - Python dependencies required for the project
- **run.sh** - Shell script to run the application
- **setup.py** - Python package setup configuration
- **config.json** - Configuration file for the application settings
- **prompt.md** - Prompt template used for generating markdown content
## Package Directory: youtube_cli
- **youtube_cli/__init__.py** - Package initialization file
- **youtube_cli/__main__.py** - Main entry point for running the package as a module
- **youtube_cli/main.py** - Core application logic and main functions
## Package Cache Directory: youtube_cli/__pycache__
- **youtube_cli/__pycache__/__init__.cpython-314.pyc** - Compiled Python cache file for __init__.py
- **youtube_cli/__pycache__/__main__.cpython-314.pyc** - Compiled Python cache file for __main__.py
- **youtube_cli/__pycache__/main.cpython-314.pyc** - Compiled Python cache file for main.py
## Package Info Directory: youtube_cli.egg-info
- **youtube_cli.egg-info/dependency_links.txt** - Dependency links for the package
- **youtube_cli.egg-info/entry_points.txt** - Entry points for console scripts
- **youtube_cli.egg-info/PKG-INFO** - Package metadata information
- **youtube_cli.egg-info/requires.txt** - Required dependencies for the package
- **youtube_cli.egg-info/SOURCES.txt** - List of source files in the package
- **youtube_cli.egg-info/top_level.txt** - Top-level package names
## User Data Directory
- **~/Downloads/youtube/downloaded_videos.json** - JSON file storing information about downloaded videos (user-specific path)

View File

@ -1,9 +0,0 @@
#!/usr/bin/env python3
"""
YouTube CLI - A command-line interface for browsing and downloading YouTube videos
"""
from .main import main
if __name__ == "__main__":
main()

View File

@ -8,6 +8,8 @@ import json
import os import os
import subprocess import subprocess
import sys import sys
import time
from datetime import datetime
from pathlib import Path from pathlib import Path
from rich.console import Console from rich.console import Console
@ -17,43 +19,17 @@ from rich.table import Table
console = Console() console = Console()
def main(): class YouTubeCLI:
"""Main entry point for the YouTube CLI application.""" def __init__(self, config_path=None):
parser = argparse.ArgumentParser( self.config = self.load_config(config_path)
description="YouTube CLI - Browse and download YouTube videos" self.original_query = None
self.current_page = 1
self.archive_file = (
Path(self.config.get("download_dir", "./")) / "downloaded_videos.json"
) )
parser.add_argument("query", nargs="?", help="Search query for YouTube") self.downloaded_videos = self.load_archive()
parser.add_argument(
"--download", action="store_true", help="Download a specific video"
)
parser.add_argument("--config", help="Configuration file path")
parser.add_argument(
"--list", action="store_true", help="List videos (default behavior)"
)
parser.add_argument("--page", type=int, default=1, help="Page number for results")
args = parser.parse_args() def load_config(self, config_path=None):
# Initialize configuration
config = load_config(args.config)
if args.download:
# Handle download functionality
if not args.query:
print("Error: You must provide a video URL for downloading")
return
download_video(args.query, config)
return
# Default behavior - show search results
if args.query:
search_videos(args.query, config, page=args.page)
else:
# Show help if no arguments given
parser.print_help()
def load_config(config_path=None):
"""Load configuration from file or use defaults.""" """Load configuration from file or use defaults."""
default_config = { default_config = {
"download_dir": str(Path.home() / "Downloads" / "youtube"), "download_dir": str(Path.home() / "Downloads" / "youtube"),
@ -64,10 +40,12 @@ def load_config(config_path=None):
], ],
"max_videos_per_page": 15, "max_videos_per_page": 15,
"yt_dlp_args": { "yt_dlp_args": {
"format": "bestvideo[height=1080]+bestaudio/bestvideo[height<=1080]+bestaudio", "format": "best",
"write_thumbnail": True, "write_thumbnail": True,
"extractor_args": "youtube:player-client=default,-tv_simply", "extractor_args": "youtube:player-client=default,-tv_simply",
}, },
"network_share_path": "/Volumes/MediaServer/Youtube/",
"default_network_subfolder": "General",
} }
if config_path and os.path.exists(config_path): if config_path and os.path.exists(config_path):
@ -80,26 +58,144 @@ def load_config(config_path=None):
config[key] = value config[key] = value
return config return config
except Exception as e: except Exception as e:
print(f"Error loading config: {e}") console.print(f"[red]Error loading config: {e}[/red]")
return default_config return default_config
def load_archive(self):
"""Load the archive of already downloaded videos."""
try:
if self.archive_file.exists():
with open(self.archive_file, "r") as f:
return json.load(f)
else:
# Initialize empty archive
self.save_archive({})
return {}
except Exception as e:
console.print(f"[red]Error loading archive: {e}[/red]")
return {}
def search_videos(query, config, page=1): def save_archive(self, videos_dict):
"""Save the archive of downloaded videos."""
try:
with open(self.archive_file, "w") as f:
json.dump(videos_dict, f, indent=2)
except Exception as e:
console.print(f"[red]Error saving archive: {e}[/red]")
def is_video_downloaded(self, video_id):
"""Check if a video has already been downloaded."""
return video_id in self.downloaded_videos
def add_to_archive(self, video_info):
"""Add a video to the archive of downloaded videos."""
try:
video_id = video_info.get("id")
if video_id:
self.downloaded_videos[video_id] = {
"title": video_info.get("title", "Unknown Title"),
"url": video_info.get("url", ""),
"downloaded_at": datetime.now().isoformat(),
}
self.save_archive(self.downloaded_videos)
except Exception as e:
console.print(f"[red]Error adding to archive: {e}[/red]")
console.print(f"[red]Video info being added: {video_info}[/red]")
def prefill_archive_from_downloads(self):
"""Pre-fill the archive with videos that already exist in download directory."""
try:
download_dir = Path(self.config.get("download_dir", "./"))
if not download_dir.exists():
return
# Find existing video files
video_extensions = [".mp4", ".mkv", ".webm", ".flv"]
for file_path in download_dir.iterdir():
if file_path.is_file() and file_path.suffix.lower() in video_extensions:
# Extract filename-based identifier
base_name = file_path.stem
# Add this as a pre-downloaded item in archive
self.downloaded_videos[base_name] = {
"title": base_name,
"url": "",
"downloaded_at": datetime.now().isoformat(),
}
self.save_archive(self.downloaded_videos)
except Exception as e:
console.print(f"[red]Error pre-filling archive: {e}[/red]")
def copy_to_network_share(self, url, config, network_folder_name):
"""Copy downloaded video to network share after download."""
try:
# Get the download directory
download_dir = Path(config["download_dir"])
# Get network share path from config
network_path = Path(config["network_share_path"])
# Create the destination folder on network share
if network_folder_name:
dest_dir = network_path / network_folder_name
else:
dest_dir = network_path / config["default_network_subfolder"]
# Ensure the directory exists on the network share
dest_dir.mkdir(parents=True, exist_ok=True)
# Find the most recently downloaded video in the download directory
video_files = list(download_dir.glob("*.*"))
if not video_files:
console.print("[yellow]No video files found to copy[/yellow]")
return
# Sort by modification time to get newest file first
latest_file = max(video_files, key=lambda f: f.stat().st_mtime)
# Check that it's a video file
if latest_file.suffix.lower() in [".mp4", ".mkv", ".webm", ".flv"]:
dest_path = dest_dir / latest_file.name
import shutil
shutil.copy2(latest_file, dest_path)
console.print(
f"[green]Copied {latest_file.name} to network share[/green]"
)
else:
console.print("[yellow]No valid video file found for copying[/yellow]")
except Exception as e:
console.print(f"[red]Error copying to network share: {e}[/red]")
def search_videos(self, query, config, page=1):
"""Search YouTube videos based on the query using yt-dlp.""" """Search YouTube videos based on the query using yt-dlp."""
console.print(f"[blue]Searching YouTube for:[/blue] {query}") console.print(f"[blue]Searching YouTube for:[/blue] {query}")
# Store original query for pagination
self.original_query = query
self.current_page = page
# Sanitize search query to prevent issues with special characters
import re
sanitized_query = re.sub(r'[^\w\s\-\'"\.]+', "", query)
# Construct yt-dlp command for search # Construct yt-dlp command for search
cmd = [ cmd = [
"yt-dlp", "yt-dlp",
"--flat-playlist", # Get video info without downloading "--flat-playlist", # Get video info without downloading
"--dump-single-json", # Output as JSON for single video "--dump-single-json", # Output as JSON single item
f"--playlist-start={15 * (page - 1) + 1}", f"--playlist-start={15 * (page - 1) + 1}",
f"--playlist-end={15 * page}", f"--playlist-end={15 * page}",
"--no-warnings", "--no-warnings",
"--no-progress", "--no-progress",
"--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
f"ytsearch{15 * page}:{query}", "--remote-components",
"ejs:github",
f"ytsearch{15 * page}:{sanitized_query}",
] ]
try: try:
@ -107,6 +203,7 @@ def search_videos(query, config, page=1):
if result.returncode != 0: if result.returncode != 0:
console.print(f"[red]Error searching videos: {result.stderr}[/red]") console.print(f"[red]Error searching videos: {result.stderr}[/red]")
console.print("[yellow]Try with a simpler search query.[/yellow]")
return return
# Parse JSON output # Parse JSON output
@ -114,10 +211,9 @@ def search_videos(query, config, page=1):
try: try:
data = json.loads(result.stdout.strip()) data = json.loads(result.stdout.strip())
except json.JSONDecodeError: except json.JSONDecodeError as e:
console.print( console.print(f"[red]Error parsing search results: {e}[/red]")
"[red]Error parsing search results. Try another search.[/red]" console.print("[yellow]Try with a simpler search query.[/yellow]")
)
return return
# Process videos into our format # Process videos into our format
@ -138,12 +234,21 @@ def search_videos(query, config, page=1):
author = entry.get("uploader", "Unknown Author") author = entry.get("uploader", "Unknown Author")
duration = entry.get("duration", 0) duration = entry.get("duration", 0)
url = entry.get("url", "") or entry.get("webpage_url", "") url = entry.get("url", "") or entry.get("webpage_url", "")
view_count = entry.get("view_count", None)
playlist_title = entry.get("playlist_title", "")
# Format duration # Format duration
length = format_duration(duration) length = self.format_duration(duration)
# Check if this is a short video # Check if this is a short video
is_short = "/shorts/" in url is_short = "/shorts/" in url or "/shorts" in url
# Check if this is a playlist (look for playlist-specific attributes)
is_playlist = "playlist" in url.lower() or "list=" in url
# Validate URL before adding to videos list
if not url or url.strip() == "":
continue # Skip videos with invalid/missing URLs
# Create video object # Create video object
videos.append( videos.append(
@ -153,24 +258,58 @@ def search_videos(query, config, page=1):
"length": length, "length": length,
"url": url, "url": url,
"is_short": is_short, "is_short": is_short,
"is_playlist": is_playlist,
"id": entry.get("id", ""), "id": entry.get("id", ""),
"thumbnail": entry.get("thumbnail", ""), "thumbnail": entry.get("thumbnail", ""),
"view_count": view_count,
} }
) )
if not videos: if not videos:
console.print("[yellow]No videos found for your search.[/yellow]") console.print("[yellow]No videos found for your search.[/yellow]")
# Ask user what they'd like to do next
console.print("[blue]Options:[/blue]")
console.print(" [green]s[/green] - Search for a new term")
console.print(" [red]q[/red] - Quit")
user_choice = input("\nChoose an option: ").strip().lower()
if user_choice == "q":
console.print("[green]Goodbye![/green]")
return
elif user_choice == "s":
search_term = input("Enter search term: ").strip()
if search_term:
console.print(f"[blue]Searching for: {search_term}[/blue]")
self.search_videos(search_term, config, page=1)
else:
console.print("[red]No search term provided.[/red]")
# Return to previous search
if self.original_query:
self.search_videos(
self.original_query, config, page=self.current_page
)
else:
self.search_videos("placeholder", config, page=1)
else:
console.print(
"[yellow]Invalid option. Returning to search results...[/yellow]"
)
if self.original_query:
self.search_videos(
self.original_query, config, page=self.current_page
)
else:
self.search_videos("placeholder", config, page=1)
return return
display_videos(videos, config, page=page) self.display_videos(videos, config, page=page)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
console.print("[red]Search timed out. Please try again.[/red]") console.print("[red]Search timed out. Please try again.[/red]")
except Exception as e: except Exception as e:
console.print(f"[red]Error during search: {str(e)}[/red]") console.print(f"[red]Error during search: {str(e)}[/red]")
def format_duration(self, seconds):
def format_duration(seconds):
"""Convert seconds to MM:SS or HH:MM:SS format.""" """Convert seconds to MM:SS or HH:MM:SS format."""
if not seconds: if not seconds:
return "0:00" return "0:00"
@ -184,8 +323,7 @@ def format_duration(seconds):
else: else:
return f"{minutes}:{secs:02d}" return f"{minutes}:{secs:02d}"
def display_videos(self, videos, config, page=1):
def display_videos(videos, config, page=1):
"""Display videos in a formatted table.""" """Display videos in a formatted table."""
console.print("\n" + "=" * 80) console.print("\n" + "=" * 80)
console.print(f"[bold]YouTube Search Results - Page {page}[/bold]") console.print(f"[bold]YouTube Search Results - Page {page}[/bold]")
@ -211,12 +349,20 @@ def display_videos(videos, config, page=1):
if video["is_short"]: if video["is_short"]:
title = f"(short) {title}" title = f"(short) {title}"
# Determine type for display
if video["is_playlist"]:
display_type = "Playlist"
elif video["is_short"]:
display_type = "Short"
else:
display_type = "Video"
table.add_row( table.add_row(
str(i), str(i),
title, title,
author[:18] + "..." if len(author) > 18 else author, author[:18] + "..." if len(author) > 18 else author,
length, length,
"Short" if video["is_short"] else "Video", display_type,
) )
console.print(table) console.print(table)
@ -225,8 +371,11 @@ def display_videos(videos, config, page=1):
console.print("=" * 80) console.print("=" * 80)
console.print("[blue]Options:[/blue]") console.print("[blue]Options:[/blue]")
console.print(" [green]n[/green] - Next page") console.print(" [green]n[/green] - Next page")
console.print(" [green]s[/green] - Search for new term (e.g. 's red bananas')")
console.print(" [red]q[/red] - Quit") console.print(" [red]q[/red] - Quit")
console.print(" [yellow]Number[/yellow] - Select and download video (e.g., 1)") console.print(
" [yellow]Number(s)[/yellow] - Select and download video(s) (e.g., 1,2,3 or 1-3)"
)
# Get user input # Get user input
user_input = input("\nChoose an option: ").strip().lower() user_input = input("\nChoose an option: ").strip().lower()
@ -237,30 +386,140 @@ def display_videos(videos, config, page=1):
elif user_input == "n": elif user_input == "n":
console.print(f"[blue]Loading page {page + 1}...[/blue]") console.print(f"[blue]Loading page {page + 1}...[/blue]")
search_videos( # Use the original query for pagination - this preserves the search term
videos[0]["title"].split(" ")[0], config, page=page + 1 if self.original_query:
) # Simple approach self.search_videos(self.original_query, config, page=page + 1)
else:
# Fallback in case original query isn't available (shouldn't happen)
self.search_videos("placeholder", config, page=page + 1)
return
elif user_input.startswith("s "):
# Search for a new term after 's'
search_term = user_input[2:].strip() # Remove 's ' prefix
if search_term:
console.print(f"[blue]Searching for: {search_term}[/blue]")
self.search_videos(search_term, config, page=1)
else:
console.print("[red]Please provide a search term after 's'.[/red]")
return
elif user_input == "s":
# Simple search command - prompt for search term
search_term = input("Enter search term: ").strip()
if search_term:
console.print(f"[blue]Searching for: {search_term}[/blue]")
self.search_videos(search_term, config, page=1)
else:
console.print("[red]No search term provided.[/red]")
return return
else: else:
# Handle download selection (single or multiple videos)
try: try:
choice = int(user_input) # Check if it's a range format like "1-3"
if 1 <= choice <= len(videos): if "-" in user_input:
selected_video = videos[choice - 1] try:
start, end = map(int, user_input.split("-"))
video_indices = list(range(start, end + 1))
except ValueError:
console.print(f"[red]Invalid range format: {user_input}[/red]")
console.print( console.print(
f"\n[blue]Selected video:[/blue] {selected_video['title']}" "[red]Please use format like '1-7' or '1,2,3'[/red]"
)
return
else:
# Handle comma-separated numbers
try:
video_indices = [
int(x.strip()) for x in user_input.split(",") if x.strip()
]
except ValueError:
console.print(f"[red]Invalid format: {user_input}[/red]")
console.print(
"[red]Please use format like '1-7' or '1,2,3'[/red]"
)
return
# Validate indices and download videos in sequence
valid_videos = []
for idx in video_indices:
if 1 <= idx <= len(videos):
valid_videos.append(videos[idx - 1])
else:
console.print(f"[red]Invalid video number: {idx}[/red]")
# Debug information for empty selection
if not valid_videos:
console.print(
"[red]Could not find any valid videos to download.[/red]"
)
return
if valid_videos:
# Ask user for network folder name (optional)
network_folder = None
console.print("[blue]Choose download destination:[/blue]")
console.print(
" [green]Enter folder name[/green] - Copy to network share"
)
console.print(
" [yellow]Press [enter] for default[/yellow] - Download only locally"
) )
# Download the video network_input = input("Network folder name: ").strip()
download_video(selected_video["url"], config) if network_input:
network_folder = network_input
else: else:
console.print("[red]Invalid selection. Please try again.[/red]") network_folder = None
console.print(
f"[blue]Downloading {len(valid_videos)} videos in sequence...[/blue]"
)
for i, selected_video in enumerate(valid_videos):
console.print(
f"\n[blue]Downloading video {i + 1}/{len(valid_videos)}:[/blue] {selected_video['title']}"
)
# Check if this is a playlist and download accordingly
if selected_video.get("is_playlist", False):
self.download_playlist(
selected_video["url"], config, network_folder
)
else:
self.download_video(
selected_video["url"],
config,
network_folder=network_folder,
)
# Give a small pause between downloads to avoid rate limiting
if (
i < len(valid_videos) - 1
): # Don't pause after the last download
time.sleep(2)
# Return to search results after all downloads complete
if self.original_query:
console.print("[blue]Returning to search results...[/blue]")
self.search_videos(
self.original_query, config, page=self.current_page
)
else:
console.print("[red]No valid videos selected for download.[/red]")
except ValueError: except ValueError:
console.print("[red]Invalid input. Please enter a number or 'q'/'n'.[/red]") console.print(
"[red]Invalid input. Please enter a number or range of numbers, 'n', 's', or 'q'.[/red]"
)
def download_video(self, url, config, network_folder=None):
def download_video(url, config):
"""Download a video using yt-dlp with progress bar.""" """Download a video using yt-dlp with progress bar."""
# Validate URL before proceeding
if not url or not isinstance(url, str) or url.strip() == "":
console.print("[red]Error: Invalid or empty video URL provided.[/red]")
return
console.print(f"[blue]Preparing to download:[/blue] {url}") console.print(f"[blue]Preparing to download:[/blue] {url}")
# Check if yt-dlp is available # Check if yt-dlp is available
@ -276,13 +535,15 @@ def download_video(url, config):
download_dir = Path(config["download_dir"]) download_dir = Path(config["download_dir"])
download_dir.mkdir(parents=True, exist_ok=True) download_dir.mkdir(parents=True, exist_ok=True)
# Prepare yt-dlp command # Prepare yt-dlp command with better handling for JS challenges
cmd = [ cmd = [
"yt-dlp", "yt-dlp",
"--no-warnings", "--no-warnings",
"-o", "-o",
str(download_dir / "%(title)s.%(ext)s"), str(download_dir / "%(title)s.%(ext)s"),
"--write-thumbnail", "--write-thumbnail",
"--remote-components",
"ejs:github",
] ]
# Add custom args from config if they exist # Add custom args from config if they exist
@ -296,63 +557,244 @@ def download_video(url, config):
if "extractor_args" in ytdlp_args: if "extractor_args" in ytdlp_args:
cmd.extend(["--extractor-args", ytdlp_args["extractor_args"]]) cmd.extend(["--extractor-args", ytdlp_args["extractor_args"]])
# For problematic videos, also add retries and better error handling
cmd.extend(
["--no-check-certificates", "--retries", "3", "--fragment-retries", "3"]
)
# Add URL # Add URL
cmd.append(url) cmd.append(url)
try: try:
console.print("[blue]Starting download...[/blue]") console.print("[blue]Starting download...[/blue]")
# Run command with progress tracking # Run command and let yt-dlp handle progress natively
process = subprocess.Popen( result = subprocess.run(
cmd, cmd,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
universal_newlines=True, universal_newlines=True,
bufsize=1, timeout=300, # 5 minute timeout
) )
# Create a simple status indicator if result.returncode == 0:
progress_console = Console()
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
console=progress_console,
) as progress:
task = progress.add_task("[blue]Downloading...", total=100)
while True:
output = process.stdout.readline()
if output == "" and process.poll() is not None:
break
if output:
# Simple progress tracking (yt-dlp doesn't support progress events directly)
progress.update(task, advance=1)
# Check if download completed successfully
return_code = process.poll()
if return_code == 0:
console.print("[green]Download completed successfully![/green]") console.print("[green]Download completed successfully![/green]")
else:
console.print("[red]Download failed.[/red]")
# Copy to network share if specified
if network_folder:
self.copy_to_network_share(url, config, network_folder)
# Add video to archive for tracking
try:
# Extract video ID from URL for archive
import re
video_id = None
id_match = re.search(r"(?:v=|\/)([0-9A-Za-z_-]{11})", url)
if id_match:
video_id = id_match.group(1)
if video_id:
self.add_to_archive(
{"url": url, "id": video_id, "title": "Unknown Title"}
)
except Exception as e:
console.print(
f"[yellow]Could not track video in archive: {e}[/yellow]"
)
else:
console.print(
f"[red]Download failed with return code {result.returncode}[/red]"
)
if result.stdout:
console.print("[red]Error details:[/red]")
console.print(result.stdout)
# Try to check if we have a different problem
# Check for specific JavaScript challenge errors and recommend solutions
if (
"Solving JS challenges" in result.stdout
or "challenge solving failed" in result.stdout
):
console.print(
"[yellow]Note: This video requires JavaScript challenge solving.[/yellow]"
)
console.print("[yellow]Install required components with:[/yellow]")
console.print(
"[yellow]yt-dlp --remote-components ejs:github[/yellow]"
)
except subprocess.TimeoutExpired:
console.print("[red]Download timed out. Please try again.[/red]")
except Exception as e: except Exception as e:
console.print(f"[red]Error during download: {str(e)}[/red]") console.print(f"[red]Error during download: {str(e)}[/red]")
# Optionally, ask to select a different download location def download_playlist(self, url, config, network_folder=None):
locations = config.get("default_locations", [config["download_dir"]]) """Download a YouTube playlist into a dedicated folder."""
if len(locations) > 1: console.print(f"[blue]Preparing to download playlist:[/blue] {url}")
console.print("\n[blue]Select download location:[/blue]")
for i, loc in enumerate(locations, 1): # Check if yt-dlp is available
console.print(f" {i}. {loc}") try:
subprocess.run(["yt-dlp", "--version"], capture_output=True, check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
console.print(
"[red]Error: yt-dlp not found. Please install it with 'pip install yt-dlp'[/red]"
)
return
# Get playlist title for folder naming
try:
cmd_get_title = [
"yt-dlp",
"--flat-playlist",
"--dump-single-json",
"--no-warnings",
url,
]
result = subprocess.run(
cmd_get_title, capture_output=True, text=True, timeout=30
)
if result.returncode == 0:
import json
try: try:
choice = int(input("Enter selection (or press Enter to use default): ")) playlist_data = json.loads(result.stdout.strip())
if 1 <= choice <= len(locations): playlist_title = playlist_data.get("title", "Unknown Playlist")
new_dir = Path(locations[choice - 1]) except json.JSONDecodeError:
config["download_dir"] = str(new_dir) playlist_title = "Unknown Playlist"
console.print(f"[green]Download location set to: {new_dir}[/green]") else:
except ValueError: playlist_title = "Unknown Playlist"
pass except Exception:
playlist_title = "Unknown Playlist"
# Set download directory
download_dir = Path(config["download_dir"])
playlist_dir = download_dir / playlist_title
playlist_dir.mkdir(parents=True, exist_ok=True)
# Prepare yt-dlp command for playlist
cmd = [
"yt-dlp",
"-o",
str(playlist_dir / "%(title)s.%(ext)s"),
"--write-thumbnail",
"--remote-components",
"ejs:github",
]
# Use best available format that includes both video and audio
# cmd.extend(["--format", "best[ext=mp4]"])
# Add extractor args from config if they exist
ytdlp_args = config.get("yt_dlp_args", {})
if "extractor_args" in ytdlp_args:
cmd.extend(["--extractor-args", ytdlp_args["extractor_args"]])
# Add retries for playlist downloads
cmd.extend(["--retries", "3", "--fragment-retries", "3"])
# Add URL
cmd.append(url)
try:
console.print("[blue]Starting playlist download...[/blue]")
# Run command and let yt-dlp handle progress natively
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
timeout=600, # 10 minute timeout for playlist downloads
)
if result.returncode == 0:
console.print(
"[green]Playlist download completed successfully![/green]"
)
# Copy to network share if specified
if network_folder:
self.copy_to_network_share(url, config, network_folder)
# Note playlist in archive - we'll track each playlist download
try:
# Extract playlist ID from URL
import re
playlist_id = None
id_match = re.search(r"(?:list=|\/)([0-9A-Za-z_-]{30,})", url)
if id_match:
playlist_id = id_match.group(1)
# For playlist tracking, we need to get the actual title differently since
# it's not available from within search results
playlist_title_for_archive = "Unknown Playlist"
if playlist_id:
self.add_to_archive(
{
"url": url,
"id": f"playlist_{playlist_id}",
"title": f"Playlist: {playlist_title_for_archive}",
}
)
except Exception as e:
console.print(
f"[yellow]Could not track playlist in archive: {e}[/yellow]"
)
else:
console.print(
f"[red]Playlist download failed with return code {result.returncode}[/red]"
)
if result.stdout:
console.print("[red]Error details:[/red]")
console.print(result.stdout)
except subprocess.TimeoutExpired:
console.print("[red]Playlist download timed out. Please try again.[/red]")
except Exception as e:
console.print(f"[red]Error during playlist download: {str(e)}[/red]")
def main():
"""Main entry point for the YouTube CLI application."""
parser = argparse.ArgumentParser(
description="YouTube CLI - Browse and download videos from YouTube"
)
parser.add_argument("query", nargs="?", help="Search query for YouTube")
parser.add_argument(
"--download", action="store_true", help="Download a video by URL"
)
parser.add_argument("--config", help="Configuration file path")
parser.add_argument("--page", type=int, default=1, help="Page number for results")
args = parser.parse_args()
# Create CLI instance
cli = YouTubeCLI(args.config)
# Pre-fill archive with existing downloads
cli.prefill_archive_from_downloads()
if args.download:
# Handle download functionality
if not args.query:
console.print(
"[red]Error: You must provide a video URL for downloading[/red]"
)
return
cli.download_video(args.query, cli.config)
return
# Default behavior - show search results
if args.query:
cli.search_videos(args.query, cli.config, page=args.page)
else:
# Show help if no arguments given
parser.print_help()
if __name__ == "__main__": if __name__ == "__main__":

File diff suppressed because it is too large Load Diff