1177 lines
43 KiB
Python
1177 lines
43 KiB
Python
#!/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
|
|
LOG_FILE = Path.home() / ".config" / "youtube_cli" / "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
|
|
self.archive_file = (
|
|
Path.home() / ".config" / "youtube_cli" / "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": "/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",
|
|
"/Volumes/MediaServer/Youtube/Anime",
|
|
],
|
|
"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:
|
|
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",
|
|
"--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:
|
|
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) or url.strip() == "":
|
|
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 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:
|
|
logger.info("Starting download...")
|
|
|
|
# Show what format will be used for download (if available)
|
|
if "format" in ytdlp_args:
|
|
logger.info(f"Using custom format: {ytdlp_args['format']}")
|
|
else:
|
|
logger.info("Using 1080p quality by default")
|
|
|
|
# Run command and let yt-dlp handle progress natively
|
|
# Removed timeout to support long-running downloads in queue
|
|
# Use a pipe to capture progress and support cancellation
|
|
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.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:
|
|
# 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:
|
|
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}")
|
|
|
|
# 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
|
|
):
|
|
logger.warning(
|
|
"Note: This video requires JavaScript challenge solving."
|
|
)
|
|
logger.warning("Install required components with:")
|
|
logger.warning("yt-dlp --remote-components ejs:github")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error during download: {str(e)}")
|
|
|
|
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
|
|
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:
|
|
logger.info("Starting playlist download...")
|
|
|
|
# Show what format will be used for download (if available)
|
|
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()
|