""" YouTube utilities for channel info, playlist info, and yt-dlp command generation. """ import subprocess import json from pathlib import Path from typing import List, Dict, Any, Optional def get_channel_info(channel_url: str, yt_dlp_path: str = "downloader/yt-dlp.exe") -> tuple[str, str]: """Get channel information using yt-dlp. Returns (channel_name, channel_id).""" try: # Extract channel name from URL for now (faster than calling yt-dlp) if "/@" in channel_url: # Keep the @ symbol for cache key consistency channel_name = "@" + channel_url.split("/@")[1].split("/")[0] elif "/channel/" in channel_url: channel_name = channel_url.split("/channel/")[1].split("/")[0] else: channel_name = "Unknown" # Extract channel ID from URL (keep @ symbol for @ channels) if "/channel/" in channel_url: channel_id = channel_url.split("/channel/")[1].split("/")[0] elif "/@" in channel_url: # Keep the @ symbol for cache key consistency channel_id = "@" + channel_url.split("/@")[1].split("/")[0] else: channel_id = channel_url return channel_name, channel_id except Exception as e: print(f"❌ Failed to get channel info: {e}") return "Unknown", channel_url def get_playlist_info(playlist_url: str, yt_dlp_path: str = "downloader/yt-dlp.exe") -> List[Dict[str, Any]]: """Get playlist information using yt-dlp.""" try: cmd = [ yt_dlp_path, "--dump-json", "--flat-playlist", playlist_url ] result = subprocess.run(cmd, capture_output=True, text=True, check=True) videos = [] for line in result.stdout.strip().split('\n'): if line.strip(): videos.append(json.loads(line)) return videos except subprocess.CalledProcessError as e: print(f"❌ Failed to get playlist info: {e}") return [] def build_yt_dlp_command( yt_dlp_path: str, video_url: str, output_path: Path, config: Dict[str, Any], additional_args: Optional[List[str]] = None ) -> List[str]: """ Build a standardized yt-dlp command with consistent arguments. Args: yt_dlp_path: Path to yt-dlp executable video_url: YouTube video URL output_path: Output file path config: Configuration dictionary with download settings additional_args: Optional additional arguments to append Returns: List of command arguments for subprocess.run """ cmd = [ str(yt_dlp_path), "--no-check-certificates", "--ignore-errors", "--no-warnings", "-o", str(output_path), "-f", config.get("download_settings", {}).get("format", "best[height<=720][ext=mp4]/best[height<=720]/best[ext=mp4]/best"), video_url ] # Add any additional arguments if additional_args: cmd.extend(additional_args) return cmd def execute_yt_dlp_command(cmd: List[str], timeout: Optional[int] = None) -> subprocess.CompletedProcess: """ Execute a yt-dlp command with standardized error handling. Args: cmd: Command list to execute timeout: Optional timeout in seconds Returns: CompletedProcess object Raises: subprocess.CalledProcessError: If the command fails subprocess.TimeoutExpired: If the command times out """ return subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=timeout) def show_available_formats(video_url: str, yt_dlp_path: str = "downloader/yt-dlp.exe", timeout: int = 30) -> None: """ Show available formats for a video (debugging utility). Args: video_url: YouTube video URL yt_dlp_path: Path to yt-dlp executable timeout: Timeout in seconds """ print(f"🔍 Checking available formats for: {video_url}") format_cmd = [ str(yt_dlp_path), "--list-formats", video_url ] try: format_result = subprocess.run(format_cmd, capture_output=True, text=True, timeout=timeout) print(f"📋 Available formats:\n{format_result.stdout}") except Exception as e: print(f"⚠️ Could not check formats: {e}")