1018 lines
40 KiB
Python
1018 lines
40 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
YouTube CLI - A command-line interface for browsing and downloading YouTube videos
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from rich.console import Console
|
|
from rich.progress import Progress, SpinnerColumn, TextColumn
|
|
from rich.table import Table
|
|
|
|
console = Console()
|
|
|
|
|
|
class YouTubeCLI:
|
|
def __init__(self, config_path=None):
|
|
self.config = self.load_config(config_path)
|
|
self.original_query = None
|
|
self.current_page = 1
|
|
self.archive_file = (
|
|
Path(self.config.get("download_dir", "./")) / "downloaded_videos.json"
|
|
)
|
|
self.downloaded_videos = self.load_archive()
|
|
|
|
def get_yt_dlp_version(self):
|
|
"""Get the current version of yt-dlp installed."""
|
|
try:
|
|
import yt_dlp
|
|
return yt_dlp.version.__version__
|
|
except ImportError:
|
|
return None
|
|
|
|
def get_latest_yt_dlp_version(self):
|
|
"""Get the latest version of yt-dlp available from PyPI."""
|
|
try:
|
|
import requests
|
|
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
return data["info"]["version"]
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
def update_yt_dlp(self):
|
|
"""Update yt-dlp to the latest version."""
|
|
try:
|
|
console.print("[blue]Updating yt-dlp to the latest version...[/blue]")
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "pip", "install", "--upgrade", "yt-dlp"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
console.print("[green]yt-dlp updated successfully![/green]")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
console.print(f"[red]Failed to update yt-dlp: {e.stderr}[/red]")
|
|
return False
|
|
except Exception as e:
|
|
console.print(f"[red]Error updating yt-dlp: {e}[/red]")
|
|
return False
|
|
|
|
def check_for_updates(self):
|
|
"""Check if yt-dlp needs to be updated and update if needed."""
|
|
current_version = self.get_yt_dlp_version()
|
|
if not current_version:
|
|
console.print("[yellow]Could not determine current yt-dlp version[/yellow]")
|
|
return False
|
|
|
|
latest_version = self.get_latest_yt_dlp_version()
|
|
if not latest_version:
|
|
console.print("[yellow]Could not determine latest yt-dlp version[/yellow]")
|
|
return False
|
|
|
|
# Simple version comparison (basic implementation)
|
|
# In a real implementation, you'd want a more robust version comparison
|
|
# Check if versions are different
|
|
if current_version != latest_version:
|
|
console.print(f"[yellow]Newer version available: {latest_version} (current: {current_version})[/yellow]")
|
|
console.print("[blue]Would you like to update? (y/n): [/blue]", end="")
|
|
try:
|
|
choice = input().strip().lower()
|
|
if choice in ['y', 'yes']:
|
|
return self.update_yt_dlp()
|
|
else:
|
|
console.print("[yellow]Update skipped[/yellow]")
|
|
return False
|
|
except Exception:
|
|
console.print("[yellow]Update skipped[/yellow]")
|
|
return False
|
|
else:
|
|
console.print(f"[green]yt-dlp is up to date: {current_version}[/green]")
|
|
return True
|
|
|
|
def load_config(self, config_path=None):
|
|
"""Load configuration from file or use defaults."""
|
|
default_config = {
|
|
"download_dir": "/Volumes/MediaServer/Youtube/",
|
|
"default_locations": [
|
|
"/Volumes/MediaServer/Youtube/",
|
|
"/Volumes/MediaServer/Youtube/Tech",
|
|
"/Volumes/MediaServer/Youtube/AI",
|
|
"/Volumes/MediaServer/Youtube/Art",
|
|
"/Volumes/MediaServer/Youtube/Homes",
|
|
"/Volumes/MediaServer/Youtube/Cooking",
|
|
"/Volumes/MediaServer/Youtube/Fitness",
|
|
"/Volumes/MediaServer/Youtube/Music",
|
|
"/Volumes/MediaServer/Youtube/Gaming",
|
|
"/Volumes/MediaServer/Youtube/Education",
|
|
"/Volumes/MediaServer/Youtube/Travel",
|
|
"/Volumes/MediaServer/Youtube/Business",
|
|
"/Volumes/MediaServer/Youtube/Science",
|
|
"/Volumes/MediaServer/Youtube/History",
|
|
"/Volumes/MediaServer/Youtube/Comedy",
|
|
"/Volumes/MediaServer/Youtube/News",
|
|
"/Volumes/MediaServer/Youtube/Sports",
|
|
"/Volumes/MediaServer/Youtube/Nature",
|
|
"/Volumes/MediaServer/Youtube/Photography",
|
|
"/Volumes/MediaServer/Youtube/Language",
|
|
"/Volumes/MediaServer/Youtube/Automotive",
|
|
],
|
|
"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",
|
|
},
|
|
"network_share_path": "/Volumes/MediaServer/Youtube/",
|
|
"default_network_subfolder": "General",
|
|
}
|
|
|
|
if config_path and os.path.exists(config_path):
|
|
try:
|
|
with open(config_path, "r") as f:
|
|
config = json.load(f)
|
|
# Merge with defaults
|
|
for key, value in default_config.items():
|
|
if key not in config:
|
|
config[key] = value
|
|
return config
|
|
except Exception as e:
|
|
console.print(f"[red]Error loading config: {e}[/red]")
|
|
|
|
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 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."""
|
|
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
|
|
cmd = [
|
|
"yt-dlp",
|
|
"--flat-playlist", # Get video info without downloading
|
|
"--dump-single-json", # Output as JSON single item
|
|
f"--playlist-start={15 * (page - 1) + 1}",
|
|
f"--playlist-end={15 * page}",
|
|
"--no-warnings",
|
|
"--no-progress",
|
|
"--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
|
"--remote-components",
|
|
"ejs:github",
|
|
f"ytsearch{15 * page}:{sanitized_query}",
|
|
]
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
|
|
|
|
if result.returncode != 0:
|
|
console.print(f"[red]Error searching videos: {result.stderr}[/red]")
|
|
console.print("[yellow]Try with a simpler search query.[/yellow]")
|
|
return
|
|
|
|
# Parse JSON output
|
|
import json
|
|
|
|
try:
|
|
data = json.loads(result.stdout.strip())
|
|
except json.JSONDecodeError as e:
|
|
console.print(f"[red]Error parsing search results: {e}[/red]")
|
|
console.print("[yellow]Try with a simpler search query.[/yellow]")
|
|
return
|
|
|
|
# Process videos into our format
|
|
videos = []
|
|
|
|
if isinstance(data, list):
|
|
videos_data = data
|
|
elif "entries" in data:
|
|
videos_data = data["entries"]
|
|
else:
|
|
videos_data = [data]
|
|
|
|
for entry in videos_data:
|
|
if not entry:
|
|
continue
|
|
|
|
title = entry.get("title", "Unknown Title")
|
|
author = entry.get("uploader", "Unknown Author")
|
|
duration = entry.get("duration", 0)
|
|
url = entry.get("url", "") or entry.get("webpage_url", "")
|
|
view_count = entry.get("view_count", None)
|
|
playlist_title = entry.get("playlist_title", "")
|
|
|
|
# Format duration
|
|
length = self.format_duration(duration)
|
|
|
|
# Check if this is a short video
|
|
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
|
|
videos.append(
|
|
{
|
|
"title": title,
|
|
"author": author,
|
|
"length": length,
|
|
"url": url,
|
|
"is_short": is_short,
|
|
"is_playlist": is_playlist,
|
|
"id": entry.get("id", ""),
|
|
"thumbnail": entry.get("thumbnail", ""),
|
|
"view_count": view_count,
|
|
}
|
|
)
|
|
|
|
if not videos:
|
|
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
|
|
|
|
self.display_videos(videos, config, page=page)
|
|
|
|
except subprocess.TimeoutExpired:
|
|
console.print("[red]Search timed out. Please try again.[/red]")
|
|
except Exception as e:
|
|
console.print(f"[red]Error during search: {str(e)}[/red]")
|
|
|
|
def format_duration(self, seconds):
|
|
"""Convert seconds to MM:SS or HH:MM:SS format."""
|
|
if not seconds:
|
|
return "0:00"
|
|
|
|
hours = int(seconds // 3600)
|
|
minutes = int((seconds % 3600) // 60)
|
|
secs = int(seconds % 60)
|
|
|
|
if hours > 0:
|
|
return f"{hours}:{minutes:02d}:{secs:02d}"
|
|
else:
|
|
return f"{minutes}:{secs:02d}"
|
|
|
|
def display_videos(self, videos, config, page=1):
|
|
"""Display videos in a formatted table."""
|
|
console.print("\n" + "=" * 80)
|
|
console.print(f"[bold]YouTube Search Results - Page {page}[/bold]")
|
|
console.print("=" * 80)
|
|
|
|
table = Table(
|
|
title=f"Page {page} of search results",
|
|
show_header=True,
|
|
header_style="bold magenta",
|
|
)
|
|
table.add_column("No.", style="dim", width=3)
|
|
table.add_column("Title", width=35)
|
|
table.add_column("Author", width=20)
|
|
table.add_column("Duration", width=10)
|
|
table.add_column("Type", width=8)
|
|
|
|
for i, video in enumerate(videos, 1):
|
|
title = video["title"]
|
|
author = video["author"]
|
|
length = video["length"]
|
|
|
|
# Mark short videos
|
|
if video["is_short"]:
|
|
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(
|
|
str(i),
|
|
title,
|
|
author[:18] + "..." if len(author) > 18 else author,
|
|
length,
|
|
display_type,
|
|
)
|
|
|
|
console.print(table)
|
|
|
|
# Show pagination options
|
|
console.print("=" * 80)
|
|
console.print("[blue]Options:[/blue]")
|
|
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(
|
|
" [yellow]Number(s)[/yellow] - Select and download video(s) (e.g., 1,2,3 or 1-3)"
|
|
)
|
|
|
|
# Get user input
|
|
user_input = input("\nChoose an option: ").strip().lower()
|
|
|
|
if user_input == "q":
|
|
console.print("[green]Goodbye![/green]")
|
|
return
|
|
|
|
elif user_input == "n":
|
|
console.print(f"[blue]Loading page {page + 1}...[/blue]")
|
|
# Use the original query for pagination - this preserves the search term
|
|
if self.original_query:
|
|
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
|
|
|
|
else:
|
|
# Handle download selection (single or multiple videos)
|
|
try:
|
|
# Check if it's a range format like "1-3"
|
|
if "-" in user_input:
|
|
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(
|
|
"[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 category selection first
|
|
console.print("[blue]Select category for all downloads:[/blue]")
|
|
selected_category = self.select_category(config)
|
|
if not selected_category:
|
|
console.print("[yellow]Download cancelled.[/yellow]")
|
|
return
|
|
|
|
# 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"
|
|
)
|
|
|
|
network_input = input("Network folder name: ").strip()
|
|
if network_input:
|
|
network_folder = network_input
|
|
else:
|
|
network_folder = None
|
|
|
|
console.print(
|
|
f"[blue]Downloading {len(valid_videos)} videos in sequence to category '{Path(selected_category).name}'...[/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, selected_category
|
|
)
|
|
else:
|
|
self.download_video(
|
|
selected_video["url"],
|
|
config,
|
|
network_folder=network_folder,
|
|
category=selected_category,
|
|
)
|
|
|
|
# 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:
|
|
console.print(
|
|
"[red]Invalid input. Please enter a number or range of numbers, 'n', 's', or 'q'.[/red]"
|
|
)
|
|
|
|
def get_categories(self, config):
|
|
"""Get list of available categories from config."""
|
|
return config.get("default_locations", [])
|
|
|
|
def select_category(self, config):
|
|
"""Allow user to select a category for download."""
|
|
categories = self.get_categories(config)
|
|
|
|
if not categories:
|
|
console.print("[red]No categories found in configuration[/red]")
|
|
return None
|
|
|
|
console.print("\n[blue]Available Categories:[/blue]")
|
|
for i, category in enumerate(categories, 1):
|
|
# Extract just the folder name for display
|
|
folder_name = Path(category).name if Path(category).name else "Root"
|
|
console.print(f" [green]{i}[/green] - {folder_name}")
|
|
|
|
while True:
|
|
try:
|
|
console.print("\n[blue]Select a category (enter number):[/blue]")
|
|
choice = input().strip()
|
|
choice_num = int(choice)
|
|
if 1 <= choice_num <= len(categories):
|
|
selected_category = categories[choice_num - 1]
|
|
console.print(f"[green]Selected category: {Path(selected_category).name}[/green]")
|
|
return selected_category
|
|
else:
|
|
console.print("[red]Invalid selection. Please try again.[/red]")
|
|
except ValueError:
|
|
console.print("[red]Please enter a valid number.[/red]")
|
|
except KeyboardInterrupt:
|
|
console.print("\n[yellow]Operation cancelled.[/yellow]")
|
|
return None
|
|
|
|
def download_video(self, url, config, network_folder=None, category=None):
|
|
"""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}")
|
|
|
|
# Check if yt-dlp is available
|
|
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
|
|
|
|
# Set download directory - always use category folder
|
|
if category:
|
|
# Ensure we're not downloading directly to base path
|
|
base_dir = Path(config["download_dir"])
|
|
category_path = Path(category)
|
|
|
|
# If category is just the base path, we need to select a different category
|
|
if str(category_path) == str(base_dir):
|
|
console.print("[yellow]Cannot download directly to base path. Please select a category.[/yellow]")
|
|
selected_category = self.select_category(config)
|
|
if not selected_category:
|
|
return
|
|
download_dir = base_dir / selected_category
|
|
else:
|
|
download_dir = base_dir / category
|
|
else:
|
|
# If no category specified, ask user to select one
|
|
selected_category = self.select_category(config)
|
|
if not selected_category:
|
|
return
|
|
download_dir = Path(config["download_dir"]) / selected_category
|
|
|
|
download_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Prepare yt-dlp command with better handling for JS challenges
|
|
cmd = [
|
|
"yt-dlp",
|
|
"--no-warnings",
|
|
"-o",
|
|
str(download_dir / "%(title)s.%(ext)s"),
|
|
"--write-thumbnail",
|
|
"--remote-components",
|
|
"ejs:github",
|
|
]
|
|
|
|
# Add custom args from config if they exist
|
|
ytdlp_args = config.get("yt_dlp_args", {})
|
|
if "format" in ytdlp_args:
|
|
cmd.extend(["--format", ytdlp_args["format"]])
|
|
if ytdlp_args.get("write_thumbnail", False):
|
|
cmd.append("--write-thumbnail")
|
|
|
|
# Add extractor args
|
|
if "extractor_args" in ytdlp_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
|
|
cmd.append(url)
|
|
|
|
try:
|
|
console.print("[blue]Starting download...[/blue]")
|
|
|
|
# Show what format will be used for download (if available)
|
|
if "format" in ytdlp_args:
|
|
console.print(
|
|
f"[cyan]Using custom format: {ytdlp_args['format']}[/cyan]"
|
|
)
|
|
else:
|
|
console.print("[cyan]Using 1080p quality by default[/cyan]")
|
|
|
|
console.print("[blue]Starting 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
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
console.print("[green]Download completed successfully![/green]")
|
|
|
|
# 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:
|
|
console.print(f"[red]Error during download: {str(e)}[/red]")
|
|
|
|
def download_playlist(self, url, config, network_folder=None, category=None):
|
|
"""Download a YouTube playlist into a dedicated folder."""
|
|
console.print(f"[blue]Preparing to download playlist:[/blue] {url}")
|
|
|
|
# Check if yt-dlp is available
|
|
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:
|
|
playlist_data = json.loads(result.stdout.strip())
|
|
playlist_title = playlist_data.get("title", "Unknown Playlist")
|
|
except json.JSONDecodeError:
|
|
playlist_title = "Unknown Playlist"
|
|
else:
|
|
playlist_title = "Unknown Playlist"
|
|
except Exception:
|
|
playlist_title = "Unknown Playlist"
|
|
|
|
# Set download directory with category support
|
|
if category:
|
|
# Ensure we're not downloading directly to base path
|
|
base_dir = Path(config["download_dir"])
|
|
category_path = Path(category)
|
|
|
|
# If category is just the base path, we need to select a different category
|
|
if str(category_path) == str(base_dir):
|
|
console.print("[yellow]Cannot download directly to base path. Please select a category.[/yellow]")
|
|
selected_category = self.select_category(config)
|
|
if not selected_category:
|
|
return
|
|
download_dir = base_dir / selected_category
|
|
else:
|
|
download_dir = base_dir / category
|
|
else:
|
|
# If no category specified, ask user to select one
|
|
selected_category = self.select_category(config)
|
|
if not selected_category:
|
|
return
|
|
download_dir = Path(config["download_dir"]) / selected_category
|
|
|
|
download_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create playlist directory within the category folder
|
|
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"])
|
|
|
|
# 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]")
|
|
|
|
# Show what format will be used for download (if available)
|
|
if "format" in ytdlp_args:
|
|
console.print(
|
|
f"[cyan]Using custom format: {ytdlp_args['format']}[/cyan]"
|
|
)
|
|
else:
|
|
console.print("[cyan]Using 1080p quality by default[/cyan]")
|
|
|
|
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=1200, # 20 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")
|
|
parser.add_argument("--update", action="store_true", help="Update yt-dlp to latest version")
|
|
parser.add_argument("--check-update", action="store_true", help="Check for yt-dlp updates")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Create CLI instance
|
|
cli = YouTubeCLI(args.config)
|
|
|
|
# Check for updates if requested
|
|
if args.check_update:
|
|
cli.check_for_updates()
|
|
return
|
|
|
|
# Update if requested
|
|
if args.update:
|
|
cli.update_yt_dlp()
|
|
return
|
|
|
|
# Pre-fill archive with existing downloads
|
|
cli.prefill_archive_from_downloads()
|
|
|
|
# Check for updates before proceeding with main operations
|
|
# This is a lightweight check that doesn't interrupt user flow
|
|
try:
|
|
cli.check_for_updates()
|
|
except Exception:
|
|
# Don't let update checking break the application
|
|
pass
|
|
|
|
if args.download:
|
|
# Handle download functionality
|
|
if not args.query:
|
|
console.print(
|
|
"[red]Error: You must provide a video URL for downloading[/red]"
|
|
)
|
|
return
|
|
# For direct download, we'll ask for category selection
|
|
cli.download_video(args.query, cli.config, category=None)
|
|
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__":
|
|
main()
|