#!/usr/bin/env python3 """ YouTube CLI - A command-line interface for browsing and downloading YouTube videos """ import argparse import json import logging import os import signal import subprocess import sys import time from datetime import datetime from logging.handlers import RotatingFileHandler from pathlib import Path from rich.console import Console from rich.table import Table console = Console() # Configure logging _config_dir = os.environ.get("CONFIG_DIR", str(Path.home() / ".config" / "youtube_cli")) LOG_FILE = Path(_config_dir) / "logs" / "app.log" LOG_FILE.parent.mkdir(parents=True, exist_ok=True) # Use RotatingFileHandler for log rotation (10MB, 5 backups) file_handler = RotatingFileHandler( LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5 ) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter( logging.Formatter( "%(asctime)s | %(name)s | %(levelname)s | %(message)s", "%Y-%m-%d %H:%M:%S", ) ) # Create console handler console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.INFO) console_handler.setFormatter(logging.Formatter("%(message)s")) # Configure root logger logging.basicConfig( level=logging.DEBUG, handlers=[ file_handler, console_handler, ], ) logger = logging.getLogger(__name__) class YouTubeCLI: def __init__(self, config_path=None): self.config = self.load_config(config_path) self.original_query = None self.current_page = 1 # Use a proper user directory for the archive file config_dir = os.environ.get("CONFIG_DIR", str(Path.home() / ".config" / "youtube_cli")) self.archive_file = Path(config_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: logger.info("Updating yt-dlp to the latest version...") subprocess.run( [sys.executable, "-m", "pip", "install", "--upgrade", "yt-dlp"], capture_output=True, text=True, check=True, ) logger.info("yt-dlp updated successfully!") return True except subprocess.CalledProcessError as e: logger.error(f"Failed to update yt-dlp: {e.stderr}") return False except Exception as e: logger.error(f"Error updating yt-dlp: {e}") 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: logger.warning("Could not determine current yt-dlp version") return False latest_version = self.get_latest_yt_dlp_version() if not latest_version: logger.warning("Could not determine latest yt-dlp version") return False # Simple version comparison (basic implementation) # In a real implementation, you'd want a more robust version comparison # Check if versions are different using proper version comparison if self._compare_versions(current_version, latest_version) < 0: logger.warning( f"Newer version available: {latest_version} (current: {current_version})" ) logger.info("Would you like to update? (y/n): ") try: choice = input().strip().lower() if choice in ["y", "yes"]: return self.update_yt_dlp() else: logger.info("Update skipped") return False except Exception: logger.info("Update skipped") return False else: logger.info(f"yt-dlp is up to date: {current_version}") return True def _compare_versions(self, version1, version2): """Compare two version strings. Returns -1 if version1 < version2, 0 if equal, 1 if version1 > version2 """ # Split versions into components v1_parts = [int(x) for x in version1.split(".")] v2_parts = [int(x) for x in version2.split(".")] # Compare each part for i in range(min(len(v1_parts), len(v2_parts))): if v1_parts[i] < v2_parts[i]: return -1 elif v1_parts[i] > v2_parts[i]: return 1 # If all compared parts are equal, the longer version is newer if len(v1_parts) < len(v2_parts): return -1 elif len(v1_parts) > len(v2_parts): return 1 else: return 0 def load_config(self, config_path=None): """Load configuration from file or use defaults.""" default_config = { "download_dir": os.environ.get("DOWNLOAD_DIR", str(Path.home() / "Downloads" / "YouTube")), "default_locations": [ "General", "Music", "Music Videos", "Podcasts", "Educational", "Tutorials", "Gaming", "Shorts", "Vlogs", "Documentaries", "Comedy", "News", "Sports", "Cooking", "Fitness", "Tech Reviews", ], "max_videos_per_page": 15, "yt_dlp_args": { "format": "bestvideo[height<=1080]+bestaudio/best", "write_thumbnail": True, }, "network_share_path": "", "default_network_subfolder": "General", } if config_path is None: config_dir = os.environ.get("CONFIG_DIR", str(Path.home() / ".config" / "youtube_cli")) config_path = str(Path(config_dir) / "config.json") 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: logger.error(f"Error loading config: {e}") 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: logger.error(f"Error loading archive: {e}") # Create the directory if it doesn't exist try: self.archive_file.parent.mkdir(parents=True, exist_ok=True) self.save_archive({}) return {} except Exception as e2: logger.error(f"Error creating archive directory: {e2}") return {} def save_archive(self, videos_dict): """Save the archive of downloaded videos.""" try: # Ensure the directory exists self.archive_file.parent.mkdir(parents=True, exist_ok=True) with open(self.archive_file, "w") as f: json.dump(videos_dict, f, indent=2) except Exception as e: logger.error(f"Error saving archive: {e}") 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: logger.error(f"Error adding to archive: {e}") logger.error(f"Video info being added: {video_info}") 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(): logger.info( "Download directory does not exist, skipping prefill" ) 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) logger.info( f"Prefilled archive with {len(self.downloaded_videos)} videos" ) except Exception as e: logger.error(f"Error pre-filling archive: {e}") 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: logger.warning("No video files found to copy") 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) logger.info(f"Copied {latest_file.name} to network share") else: logger.warning("No valid video file found for copying") except Exception as e: logger.error(f"Error copying to network share: {e}") def search_videos( self, query, config, page=1, return_results: bool = False ): """Search YouTube videos based on the query using yt-dlp.""" logger.info(f"Searching YouTube for: {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", f"ytsearch{15 * page}:{sanitized_query}", ] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=60 ) if result.returncode != 0: logger.error(f"Error searching videos: {result.stderr}") logger.warning("Try with a simpler search query.") if return_results: return [] return # Parse JSON output import json try: data = json.loads(result.stdout.strip()) except json.JSONDecodeError as e: logger.error(f"Error parsing search results: {e}") logger.warning("Try with a simpler search query.") if return_results: return [] 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) # 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: logger.warning("No videos found for your search.") if return_results: return [] # Ask user what they'd like to do next logger.info("Options: s - Search for a new term, q - Quit") user_choice = input("\nChoose an option: ").strip().lower() if user_choice == "q": logger.info("Goodbye!") return elif user_choice == "s": search_term = input("Enter search term: ").strip() if search_term: logger.info(f"Searching for: {search_term}") self.search_videos(search_term, config, page=1) else: logger.error("No search term provided.") # 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: logger.warning( "Invalid option. Returning to search results..." ) if self.original_query: self.search_videos( self.original_query, config, page=self.current_page ) else: self.search_videos("placeholder", config, page=1) return if return_results: return videos self.display_videos(videos, config, page=page) except subprocess.TimeoutExpired: logger.error("Search timed out. Please try again.") if return_results: return [] except Exception as e: logger.error(f"Error during search: {str(e)}") if return_results: return [] 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.""" logger.info(f"Displaying {len(videos)} videos on page {page}") 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, ) # Show pagination options logger.info( f"Page {page} - Options: n - Next page, s - Search, q - Quit, or numbers to download" ) # Get user input user_input = input("\nChoose an option: ").strip().lower() if user_input == "q": logger.info("User quit") return elif user_input == "n": logger.info(f"Loading page {page + 1}...") # 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: logger.info(f"Searching for: {search_term}") self.search_videos(search_term, config, page=1) else: logger.error("Please provide a search term after 's'.") return elif user_input == "s": # Simple search command - prompt for search term search_term = input("Enter search term: ").strip() if search_term: logger.info(f"Searching for: {search_term}") self.search_videos(search_term, config, page=1) else: logger.error("No search term provided.") 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: logger.info(f"Invalid range format: {user_input}") logger.info("Please use format like '1-7' or '1,2,3'") return else: # Handle comma-separated numbers try: video_indices = [ int(x.strip()) for x in user_input.split(",") if x.strip() ] except ValueError: logger.error(f"Invalid format: {user_input}") logger.error("Please use format like '1-7' or '1,2,3'") 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: logger.error(f"Invalid video number: {idx}") # Debug information for empty selection if not valid_videos: logger.error("Could not find any valid videos to download.") return if valid_videos: # Ask user for category selection first logger.info("Select category for all downloads") selected_category = self.select_category(config) if not selected_category: logger.warning("Download cancelled.") return # Ask user for network folder name (optional) network_folder = None logger.info( "Choose download destination: Enter folder name for network share, or press Enter for default" ) network_input = input("Network folder name: ").strip() if network_input: network_folder = network_input else: network_folder = None logger.info( f"Downloading {len(valid_videos)} videos in sequence to category '{Path(selected_category).name}'..." ) for i, selected_video in enumerate(valid_videos): logger.info( f"Downloading video {i + 1}/{len(valid_videos)}: {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: logger.info("Returning to search results...") self.search_videos( self.original_query, config, page=self.current_page ) else: logger.error("No valid videos selected for download.") except ValueError: logger.error( "Invalid input. Please enter a number or range of numbers, 'n', 's', or 'q'." ) 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: logger.error("No categories found in configuration") return None logger.info(f"Available categories: {', '.join(categories)}") while True: try: logger.info( "Select a category (enter number) or type a custom folder name:" ) choice = input().strip() # Check if input is a number if choice.isdigit(): choice_num = int(choice) if 1 <= choice_num <= len(categories): selected_category = categories[choice_num - 1] logger.info( f"Selected category: {Path(selected_category).name}" ) return selected_category else: logger.error("Invalid selection. Please try again.") else: # Validate custom folder name if not choice: logger.error( "Folder name cannot be empty. Please try again." ) continue # Check for invalid characters (only allow a-z, 0-9, and hyphens/underscores) import re if not re.match(r"^[a-zA-Z0-9_-]+$", choice): logger.error( "Invalid characters. Only a-z, 0-9, hyphens, and underscores are allowed." ) continue # If valid, use the custom folder name logger.info(f"Using custom folder: {choice}") # Return the custom folder name (will be appended to base path) return choice except KeyboardInterrupt: logger.info("Operation cancelled.") return None def download_video( self, url, config, network_folder=None, category=None, progress_callback=None, ): """Download a video using yt-dlp with progress bar.""" # Validate URL before proceeding if not url or not isinstance(url, str): logger.error("Invalid or empty video URL provided.") return False logger.info(f"Preparing to download: {url}") # Check if yt-dlp is available try: subprocess.run( ["yt-dlp", "--version"], capture_output=True, check=True ) except (subprocess.CalledProcessError, FileNotFoundError): logger.error( "yt-dlp not found. Please install it with 'pip install yt-dlp'" ) 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): logger.warning( "Cannot download directly to base path. Please select a category." ) 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) logger.debug(f"Download directory: {download_dir}") # Prepare yt-dlp command (minimal - nightly handles JS/challenges) cmd = [ "yt-dlp", "-o", str(download_dir / "%(title)s.%(ext)s"), "--write-thumbnail", "--download-archive", str(download_dir / ".yt-dlp-archive.txt"), "--js-runtimes", "deno", "--remote-components", "ejs:github", "--extractor-args", "youtube:pot_provider=deno,player_client=web,ios,android", ] # Add retries cmd.extend(["--retries", "3", "--fragment-retries", "3"]) # Add URL cmd.append(url) try: logger.info("Starting download...") # Show what format will be used for download (if available) ytdlp_args = self.config.get("yt_dlp_args", {}) if "format" in ytdlp_args: logger.info(f"Using custom format: {ytdlp_args['format']}") else: logger.info("Using 1080p quality by default") # Run command with progress bar process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, text=True, ) # Rich progress bar for downloads from rich.progress import ( BarColumn, DownloadColumn, Progress, TextColumn, TimeRemainingColumn, TransferSpeedColumn, ) progress = Progress( TextColumn("[bold blue]{task.description}"), BarColumn(bar_width=40), "[progress.percentage]{task.percentage:>3.1f}%", "•", DownloadColumn(), "•", TransferSpeedColumn(), "•", TimeRemainingColumn(), ) downloaded_bytes = 0 total_bytes = 0 download_task = None output_lines = [] try: with progress: for line in process.stdout: output_lines.append(line) # Parse progress line: "[download] 5.0% of ~ 10.00MiB at 1.23MiB/s ETA 00:08" import re as re_mod progress_match = re_mod.search( r"\[download\]\s+(\d+\.?\d*)%\s+of\s+[~]?\s*([\d.]+)\s*(B|KiB|MiB|GiB)", line, ) if progress_match: pct = float(progress_match.group(1)) size_str = progress_match.group(2) size_unit = progress_match.group(3) total_bytes = self._parse_size(size_str, size_unit) downloaded_bytes = int(total_bytes * pct / 100) if download_task is None: download_task = progress.add_task( "Downloading...", total=total_bytes if total_bytes > 0 else None, ) progress.update( download_task, completed=downloaded_bytes, total=total_bytes if total_bytes > 0 else None, ) elif "[download]" in line and "[" not in line.split("[download]")[1].split()[0]: pass elif "[download] Deprecated commandline" in line: pass result = subprocess.CompletedProcess( cmd, process.returncode, "".join(output_lines), "" ) except KeyboardInterrupt: process.kill() logger.warning("Download cancelled by user") return False if result.returncode == 0: logger.info("Download completed successfully!") # 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: 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: logger.warning(f"Could not track video in archive: {e}") else: logger.error( f"Download failed with return code {result.returncode}" ) if result.stdout: logger.error(f"Error details: {result.stdout}") if ( "Solving JS challenges" in result.stdout or "challenge solving failed" in result.stdout ): logger.warning( "Note: This video requires JavaScript challenge solving." ) logger.warning("Update yt-dlp with:") logger.warning("pip install --upgrade --pre yt-dlp") except Exception as e: logger.error(f"Error during download: {str(e)}") def _parse_size(self, size_str: str, unit: str) -> int: """Parse size string with unit to bytes.""" multipliers = {"B": 1, "KiB": 1024, "MiB": 1024**2, "GiB": 1024**3} return int(float(size_str) * multipliers.get(unit, 1)) def download_playlist( self, url, config, network_folder=None, category=None, progress_callback=None, ): """Download a YouTube playlist into a dedicated folder.""" logger.info(f"Preparing to download playlist: {url}") # Check if yt-dlp is available try: subprocess.run( ["yt-dlp", "--version"], capture_output=True, check=True ) except (subprocess.CalledProcessError, FileNotFoundError): logger.info( "Error: yt-dlp not found. Please install it with 'pip install yt-dlp'" ) 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): logger.info( "Cannot download directly to base path. Please select a category." ) 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 (minimal) cmd = [ "yt-dlp", "-o", str(playlist_dir / "%(title)s.%(ext)s"), "--write-thumbnail", "--download-archive", str(playlist_dir / ".yt-dlp-archive.txt"), "--js-runtimes", "deno", "--remote-components", "ejs:github", "--extractor-args", "youtube:pot_provider=deno,player_client=web,ios,android", "--retries", "3", "--fragment-retries", "3", ] # Add URL cmd.append(url) try: logger.info("Starting playlist download...") # Show what format will be used for download (if available) ytdlp_args = self.config.get("yt_dlp_args", {}) if "format" in ytdlp_args: logger.info(f"Using custom format: {ytdlp_args['format']}") else: logger.info("Using 1080p quality by default") logger.info("Starting playlist download...") # Run command and let yt-dlp handle progress natively # Removed timeout to support long-running downloads in queue # Use subprocess.Popen for cancellation support process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, text=True, ) try: stdout, _ = process.communicate(timeout=None) # No timeout result = subprocess.CompletedProcess( cmd, process.returncode, stdout, "" ) except subprocess.TimeoutExpired: process.kill() logger.info("Download cancelled by user") return False if result.returncode == 0: logger.info("Playlist download completed successfully!") # 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: logger.info(f"Could not track playlist in archive: {e}") else: logger.info( f"Playlist download failed with return code {result.returncode}" ) if result.stdout: logger.info("Error details:") logger.info(result.stdout) except Exception as e: logger.info(f"[red]Error during playlist download: {str(e)}[/red]") def signal_handler(sig, frame): """Handle Ctrl+C gracefully.""" logger.info("\nOperation cancelled by user.") sys.exit(0) def main(): """Main entry point for the YouTube CLI application.""" # Register signal handler for Ctrl+C signal.signal(signal.SIGINT, signal_handler) 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: logger.info("Error: You must provide a video URL for downloading") 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()