#!/usr/bin/env python3 """Audio download and preprocessing with robust error handling. === v7.0 OPTIMIZATION: Download-Time Resample === Uses ffmpeg directly to create both 16kHz processing audio AND original quality audio in a single pass, eliminating Python resampling overhead. Previous: yt-dlp → temp.wav → torchaudio load → Python resample → save both Now: yt-dlp → ffmpeg pipe → dual output (16kHz + original) in one pass Saves ~10-15s per video by avoiding torchaudio resampling. """ import time import os import logging import hashlib import subprocess from pathlib import Path from typing import Tuple, Dict, Any, Optional, List, Iterable import torchaudio import yt_dlp logger = logging.getLogger("FastPipelineV6.Download") def get_ytdlp_proxy() -> Optional[str]: """ Get proxy URL for yt-dlp from environment variable. Set YTDLP_PROXY environment variable with your proxy URL, e.g.: export YTDLP_PROXY="http://user:pass@proxy.example.com:7777" Returns: Proxy URL string or None if not configured. """ return os.environ.get('YTDLP_PROXY') # === COMPATIBILITY FIX: torchaudio.info deprecated in newer versions === def get_audio_info(file_path: str) -> Dict[str, Any]: """ Get audio file info (sample_rate, num_frames, duration). Compatible with all torchaudio versions by using torchaudio.load and inferring metadata from the returned tensor. Args: file_path: Path to audio file Returns: Dict with 'sample_rate', 'num_frames', 'duration' keys """ try: # torchaudio.info exists in older versions if hasattr(torchaudio, 'info'): try: info = torchaudio.info(str(file_path)) return { 'sample_rate': info.sample_rate, 'num_frames': info.num_frames, 'duration': info.num_frames / info.sample_rate } except Exception: pass # Fall through to load method # Fallback: Load file to get metadata (works in all versions) # Only load first 0.1s to get sample rate, then compute frames from file size waveform, sample_rate = torchaudio.load(str(file_path)) num_frames = waveform.shape[1] return { 'sample_rate': sample_rate, 'num_frames': num_frames, 'duration': num_frames / sample_rate } except Exception as e: raise RuntimeError(f"Could not get audio info for {file_path}: {e}") # === EDGE CASE HANDLING: Video Validation === class VideoValidationError(Exception): """Raised when video fails validation checks.""" pass class DownloadError(Exception): """Raised when download fails after retries.""" pass def fetch_youtube_metadata(video_id: str, timeout: float = 10.0) -> Dict[str, Any]: """ Fetch YouTube metadata (chapters, title, duration) without downloading. === v3: Proxy support for rate limit avoidance === Set YTDLP_PROXY env var for high-volume metadata fetching. Args: video_id: YouTube video ID (11 chars) timeout: Max seconds to wait for metadata Returns: Dict with chapters, title, duration, etc. Empty dict on failure. """ youtube_url = f"https://www.youtube.com/watch?v={video_id}" ydl_opts = { 'quiet': True, 'no_warnings': True, 'extract_flat': False, 'skip_download': True, 'socket_timeout': timeout, } # Add proxy if configured proxy = get_ytdlp_proxy() if proxy: ydl_opts['proxy'] = proxy try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(youtube_url, download=False) if info: return { 'title': info.get('title', f'Video {video_id}'), 'duration': info.get('duration', 0), 'chapters': info.get('chapters') or [], 'uploader': info.get('uploader', 'Unknown'), 'upload_date': info.get('upload_date'), } except Exception as e: logger.debug(f"Could not fetch YouTube metadata for {video_id}: {e}") return {} def validate_video(video_url: str, min_duration: float = 30.0, max_duration: float = 14400.0) -> Tuple[bool, str, Dict]: """ Validate YouTube video before processing. Checks: - Video exists and is accessible - Video duration is within acceptable range - Video has audio track - Video is not age-restricted or private Args: video_url: YouTube URL to validate min_duration: Minimum video duration in seconds (default: 30s) max_duration: Maximum video duration in seconds (default: 4 hours) Returns: (is_valid, message, info_dict) """ logger.info(f"🔍 Validating video: {video_url}") ydl_opts = { 'quiet': True, 'no_warnings': True, 'extract_flat': False, 'skip_download': True, } # Add proxy if configured proxy = get_ytdlp_proxy() if proxy: ydl_opts['proxy'] = proxy try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(video_url, download=False) if info is None: return False, "Video info extraction returned None", {} # Check if video has audio if info.get('acodec') == 'none' and not info.get('formats'): return False, "Video has no audio track", info # Check duration duration = info.get('duration', 0) if duration < min_duration: return False, f"Video too short: {duration}s (min: {min_duration}s)", info if duration > max_duration: return False, f"Video too long: {duration}s (max: {max_duration}s)", info # Check availability if info.get('availability') == 'private': return False, "Video is private", info if info.get('age_limit', 0) > 0: logger.warning(f"⚠️ Video is age-restricted (age_limit={info.get('age_limit')})") # Check for live stream (not fully supported) if info.get('is_live'): return False, "Live streams are not supported", info title = info.get('title', 'Unknown') logger.info(f"✅ Video validated: '{title}' ({duration}s)") return True, "OK", info except yt_dlp.utils.DownloadError as e: error_msg = str(e) if "Video unavailable" in error_msg: return False, "Video unavailable or deleted", {} elif "Private video" in error_msg: return False, "Video is private", {} elif "Sign in" in error_msg: return False, "Video requires sign-in (age-restricted or private)", {} else: return False, f"Download error: {error_msg}", {} except Exception as e: return False, f"Validation failed: {str(e)}", {} def get_dynamic_intro_skip(total_duration: float) -> float: """ Calculate dynamic intro skip based on video duration. Strategy (when no chapter info available): - Video ≤ 30 min (1800s): Skip 180s (3 min) - ~10% intro ratio - Video > 30 min: Skip 300s (5 min) - smaller % for longer content This is ONLY used when no intro chapter is detected. Chapter-based detection always takes priority. Args: total_duration: Total video duration in seconds Returns: Intro skip duration in seconds """ THRESHOLD_30MIN = 30 * 60 # 1800s SKIP_SHORT = 180.0 # 3 min for ≤30min videos SKIP_LONG = 300.0 # 5 min for >30min videos if total_duration <= THRESHOLD_30MIN: return SKIP_SHORT else: return SKIP_LONG def detect_intro_outro_from_chapters(chapters, total_duration, config): """ Detect intro/outro sections from YouTube chapters. === v3: SIMPLIFIED INTRO DETECTION === === LOGIC === 1. If chapters available: - Always skip 1st chapter (assumed to be intro) - If 2nd chapter name matches intro keywords, skip it too 2. If no chapters: - Video ≤30 min: skip 3 min (180s) - Video >30 min: skip 5 min (300s) Returns: (intro_seconds, outro_seconds, skip_reason, matched_title): Trim amounts + metadata """ # === MULTILINGUAL INTRO KEYWORDS (for 2nd chapter check) === intro_keywords = [ # English 'intro', 'introduction', 'opening', 'advertisement', 'sponsor', 'ad', 'promo', 'sponsored', 'sponsorship', 'brought to you', 'promotional', # Hindi (हिंदी) 'परिचय', 'भूमिका', 'शुरुआत', 'प्रस्तावना', 'विज्ञापन', 'प्रायोजक', 'parichay', 'bhumika', 'shuruaat', 'prastavana', 'vigyapan', 'prayojak', # Telugu (తెలుగు) 'పరిచయం', 'ప్రారంభం', 'ప్రకటన', 'స్పాన్సర్', 'parichayam', 'prarambham', 'prakatana', # Kannada (ಕನ್ನಡ) 'ಪರಿಚಯ', 'ಆರಂಭ', 'ಜಾಹೀರಾತು', 'ಪ್ರಾಯೋಜಕ', 'parichaya', 'arambha', 'jahiratu', # Tamil (தமிழ்) 'அறிமுகம்', 'தொடக்கம்', 'விளம்பரம்', 'ஸ்பான்சர்', 'arimukam', 'thodakkam', 'vilambaram', # Malayalam (മലയാളം) 'പരിചയം', 'ആമുഖം', 'തുടക്കം', 'പരസ്യം', 'സ്പോൺസർ', 'parichayam', 'aamukham', 'thudakkam', 'parasyam', # Gujarati (ગુજરાતી) 'પરિચય', 'આરંભ', 'જાહેરાત', 'સ્પોન્સર', 'parichay', 'aarambh', 'jaherat', # Bengali (বাংলা) 'পরিচয়', 'ভূমিকা', 'শুরু', 'বিজ্ঞাপন', 'স্পন্সর', 'porichoy', 'bhumika', 'shuru', 'biggapon', # Marathi (मराठी) 'परिचय', 'प्रस्तावना', 'सुरुवात', 'जाहिरात', 'प्रायोजक', 'parichay', 'prastavana', 'suruvat', 'jahirat', # Punjabi (ਪੰਜਾਬੀ) 'ਜਾਣ-ਪਛਾਣ', 'ਸ਼ੁਰੂਆਤ', 'ਇਸ਼ਤਿਹਾਰ', 'ਸਪਾਂਸਰ', 'jaan-pehchaan', 'shuruaat', 'ishtihar', ] outro_keywords = [ 'outro', 'ending', 'credits', 'sponsor', 'ad', 'promo', 'endcard', 'conclusion', 'closing', 'subscribe', 'like and subscribe', # Hindi 'समापन', 'अंत', 'सब्सक्राइब', # Telugu 'ముగింపు', 'సబ్స్క్రైబ్', # Other Indic 'ముగింపు', 'முடிவு', 'അവസാനം', 'સમાપન', 'শেষ', ] intro_skip = 0.0 outro_skip = 0.0 skip_reason = None matched_title = None # === v3: CHAPTER-BASED DETECTION === if chapters and len(chapters) > 0: first_chapter = chapters[0] first_chapter_title = first_chapter.get('title', '') first_chapter_end = first_chapter.get('end_time', 0) # Always skip 1st chapter (assumed intro) intro_skip = first_chapter_end skip_reason = "first_chapter" matched_title = first_chapter_title logger.info(f" 📖 Skipping 1st chapter: '{first_chapter_title}' (0s - {first_chapter_end:.1f}s)") # Check if 2nd chapter also matches intro keywords if len(chapters) > 1: second_chapter = chapters[1] second_chapter_title = second_chapter.get('title', '') second_chapter_title_lower = second_chapter_title.lower() second_chapter_end = second_chapter.get('end_time', 0) # Check for keyword match in 2nd chapter matched_keyword = None for keyword in intro_keywords: if keyword in second_chapter_title_lower: matched_keyword = keyword break if matched_keyword: intro_skip = second_chapter_end skip_reason = f"first_two_chapters:{matched_keyword}" matched_title = f"{first_chapter_title} + {second_chapter_title}" logger.info(f" 📖 Also skipping 2nd chapter (keyword '{matched_keyword}'): '{second_chapter_title}' ({second_chapter_end:.1f}s)") # Check last chapter for outro if len(chapters) > 1: last_chapter = chapters[-1] chapter_title = last_chapter.get('title', '').lower() chapter_start = last_chapter.get('start_time', 0) chapter_end = last_chapter.get('end_time', total_duration) # If last chapter is short (<5 min) and has outro keywords, trim it if any(keyword in chapter_title for keyword in outro_keywords): if chapter_end - chapter_start < 300: # Less than 5 minutes outro_skip = total_duration - chapter_start logger.info(f" 📖 Chapter-based outro: '{last_chapter.get('title')}' ({outro_skip:.1f}s)") # === DYNAMIC DURATION-BASED SKIP (when NO chapters available) === else: if getattr(config, 'auto_intro_skip', True) and config.intro_skip_seconds == 0.0: intro_skip = get_dynamic_intro_skip(total_duration) duration_threshold = '≤30min' if total_duration <= 1800 else '>30min' skip_reason = f"dynamic:{duration_threshold}" matched_title = None logger.info(f" ⏱️ Dynamic intro skip (no chapters): {intro_skip:.0f}s (video {duration_threshold})") return intro_skip, outro_skip, skip_reason, matched_title def extract_video_id(video_url: str) -> str: """Extract video ID from various URL formats (YouTube, R2, etc).""" # Bare YouTube video ID (most common in batch files) # Example: kIP1SiiNzVA if len(video_url) == 11 and all((c.isalnum() or c in "-_") for c in video_url): return video_url # YouTube formats if 'watch?v=' in video_url: return video_url.split('watch?v=')[1].split('&')[0] elif 'youtu.be/' in video_url: return video_url.split('youtu.be/')[1].split('?')[0] elif '/shorts/' in video_url: return video_url.split('/shorts/')[1].split('?')[0] # R2/S3 presigned URL format: extract from path before extension # e.g., https://xxx.r2.cloudflarestorage.com/bucket/podcasts/---g_VL7ySo.webm?X-Amz-... elif 'r2.cloudflarestorage.com' in video_url or 's3.amazonaws.com' in video_url: # Extract path from URL (before query string) path = video_url.split('?')[0] # Get the filename filename = path.split('/')[-1] # Remove extension video_id = filename.rsplit('.', 1)[0] if '.' in filename else filename return video_id else: return hashlib.md5(video_url.encode()).hexdigest()[:11] # Per-process cached R2 client for source/original videos (bucket: "test"). # We keep this out of module imports to avoid paying boto3 init cost for # non-R2 workflows. _R2_SOURCE_CLIENT = None def _get_r2_source_client(config) -> "R2Client": """Lazy-init the R2 client used for reading source videos (bucket_type='source').""" global _R2_SOURCE_CLIENT if _R2_SOURCE_CLIENT is None: from src.r2_client import R2Client, get_r2_config_from_env, R2Config base = get_r2_config_from_env() # Avoid accidental prefixing from generic R2_PREFIX when accessing the source bucket. # Use R2_SOURCE_PREFIX if provided; otherwise default to empty. source_prefix = os.environ.get('R2_SOURCE_PREFIX', '') cfg = R2Config( endpoint_url=base.endpoint_url, access_key_id=base.access_key_id, secret_access_key=base.secret_access_key, bucket=base.bucket, # overridden by bucket_type='source' prefix=source_prefix, ) _R2_SOURCE_CLIENT = R2Client(config=cfg, bucket_type='source') return _R2_SOURCE_CLIENT def _iter_r2_source_prefixes(config) -> List[str]: """ Build a small set of candidate prefixes for source objects without listing the bucket. """ prefixes: List[str] = [] # Highest priority: explicit env/config. env_prefixes = os.environ.get('R2_SOURCE_PREFIXES', '').strip() if env_prefixes: prefixes.extend([p.strip() for p in env_prefixes.split(',') if p.strip()]) cfg_prefix = getattr(config, 'r2_source_prefix', '') or '' if cfg_prefix: prefixes.append(cfg_prefix) # Common defaults seen in earlier code/comments. prefixes.extend(['', 'podcasts', 'podcasts/', 'videos', 'videos/']) # Normalize to "dir/" or "" and dedupe while preserving order. normalized: List[str] = [] seen = set() for p in prefixes: p = p.strip() if not p: p_norm = '' else: p_norm = p if p.endswith('/') else (p + '/') if p_norm not in seen: seen.add(p_norm) normalized.append(p_norm) return normalized def _iter_r2_source_extensions(config) -> List[str]: """Return candidate extensions to try for source objects (small list, no listing).""" exts: List[str] = [] env_exts = os.environ.get('R2_SOURCE_EXTENSIONS', '').strip() if env_exts: exts.extend([e.strip() for e in env_exts.split(',') if e.strip()]) cfg_exts = getattr(config, 'r2_source_extensions', None) if cfg_exts: if isinstance(cfg_exts, str): exts.extend([e.strip() for e in cfg_exts.split(',') if e.strip()]) elif isinstance(cfg_exts, (list, tuple)): exts.extend([str(e).strip() for e in cfg_exts if str(e).strip()]) # Reasonable defaults for YouTube-origin files. exts.extend(['', '.webm', '.mp4', '.mkv', '.m4a', '.mp3', '.opus']) # Normalize: ensure extensions start with '.' (except empty). normalized: List[str] = [] seen = set() for e in exts: e = e.strip() if not e: e_norm = '' else: e_norm = e if e.startswith('.') else ('.' + e) if e_norm not in seen: seen.add(e_norm) normalized.append(e_norm) return normalized def _resolve_r2_source_object(video_id: str, config) -> Tuple[str, Dict[str, Any]]: """ Resolve a video_id to an object key in the R2 source bucket. We avoid bucket listing for performance/cost reasons. Instead we try a small set of likely prefixes/extensions via HEAD requests only. Returns: (remote_key, head_object_response) """ client = _get_r2_source_client(config) def _prioritize(items: List[str], preferred: List[str]) -> List[str]: """Return items with preferred values moved to the front (stable).""" out: List[str] = [] seen = set() for p in preferred: if p in items and p not in seen: out.append(p) seen.add(p) for it in items: if it not in seen: out.append(it) seen.add(it) return out prefixes = _prioritize( _iter_r2_source_prefixes(config), # Most common layout: podcasts/{id}.webm ["podcasts/", "videos/", ""], ) exts = _prioritize( _iter_r2_source_extensions(config), # Prefer common YouTube container formats first. [".webm", ".mp4", ".mkv", ".m4a", ".opus", ".mp3", ""], ) last_key = None for prefix in prefixes: for ext in exts: key = f"{prefix}{video_id}{ext}" last_key = key head = client.head_object(key) if head: return key, head raise DownloadError( f"R2 source not found: s3://{client.bucket}/{last_key or video_id}" ) def is_r2_url(url: str) -> bool: """Check if URL is an R2/S3 presigned URL.""" return 'r2.cloudflarestorage.com' in url or ('s3.' in url and 'amazonaws.com' in url) def extract_r2_path(presigned_url: str) -> str: """ Extract clean R2 bucket/key path from a presigned URL. Example: Input: https://xxx.r2.cloudflarestorage.com/test/podcasts/VIDEO.webm?X-Amz-... Output: test/podcasts/VIDEO.webm """ try: from urllib.parse import urlparse parsed = urlparse(presigned_url) # Path is like /bucket/key - remove leading slash path = parsed.path.lstrip('/') return path except: return presigned_url # Fallback to full URL def _download_from_r2( r2_url: str, video_id: str, output_dir: Path, config, max_retries: int = 3, supabase_metadata: Dict[str, Any] = None, ) -> Tuple[str, Dict[str, Any]]: """ Download audio directly from R2 presigned URL using ffmpeg. === v2: R2 DIRECT DOWNLOAD === For pre-cached videos in R2, we bypass yt-dlp entirely and use ffmpeg to download and convert in a single pass. Creates: - {video_id}_trimmed.wav (16kHz mono for processing) - {video_id}_original.flac (original quality mono for export) Returns: (audio_path, metadata): Path to processed audio and basic metadata """ logger.info(f"⚡ R2 direct download: {video_id}") start = time.time() # Output file paths trimmed_file = output_dir / f"{video_id}_trimmed.wav" original_file = output_dir / f"{video_id}_original.flac" raw_file = output_dir / f"{video_id}_raw.wav" # Check cache if trimmed_file.exists() and original_file.exists(): logger.info(f"✅ Using cached audio: {trimmed_file}") try: # Load a bit to get metadata (torchaudio.info deprecated in newer versions) proc_waveform, proc_sr = torchaudio.load(str(trimmed_file)) orig_waveform, orig_sr = torchaudio.load(str(original_file)) return str(trimmed_file), { 'video_id': video_id, 'youtube_url': f'https://www.youtube.com/watch?v={video_id}', 'title': f'R2 Source: {video_id}', 'original_duration': orig_waveform.shape[1] / orig_sr, 'intro_skipped': 0.0, 'processed_duration': proc_waveform.shape[1] / proc_sr, 'output_dir': str(output_dir), 'chapters': [], 'sample_rate': proc_sr, 'original_sample_rate': orig_sr, 'original_audio_path': str(original_file), 'original_audio_preserved': True, 'audio_format': 'flac', 'source_type': 'r2', } except Exception as e: logger.warning(f"Cache validation failed: {e}, re-downloading") # Step 1: Download raw audio from R2 using ffmpeg last_error = None for attempt in range(max_retries): try: logger.info(f" R2 download attempt {attempt + 1}/{max_retries}") # ffmpeg can download from HTTP URLs directly # Convert to mono WAV, preserve original sample rate ffmpeg_download = [ 'ffmpeg', '-y', '-i', r2_url, '-vn', # No video '-ac', '1', # Mono '-f', 'wav', str(raw_file) ] result = subprocess.run( ffmpeg_download, capture_output=True, text=True, timeout=300 # 5 min timeout ) if result.returncode != 0: raise DownloadError(f"ffmpeg failed: {result.stderr[:500]}") if raw_file.exists() and raw_file.stat().st_size > 10000: break else: raise DownloadError("Downloaded file too small or missing") except subprocess.TimeoutExpired: last_error = DownloadError("Download timed out") logger.warning(f" R2 download timeout, attempt {attempt + 1}") except Exception as e: last_error = e logger.warning(f" R2 download attempt {attempt + 1} failed: {e}") if attempt < max_retries - 1: time.sleep(2 ** attempt) continue if not raw_file.exists(): raise DownloadError(f"R2 download failed after {max_retries} attempts: {last_error}") # Get audio info - use torchaudio.load metadata or ffprobe try: # Load a small portion to get metadata waveform, original_sr = torchaudio.load(str(raw_file)) actual_duration = waveform.shape[1] / original_sr del waveform # Free memory except Exception as e: logger.warning(f"Could not get audio info via torchaudio: {e}, trying ffprobe") try: # Fallback to ffprobe probe_cmd = [ 'ffprobe', '-v', 'error', '-show_entries', 'format=duration:stream=sample_rate', '-of', 'json', str(raw_file) ] probe_result = subprocess.run(probe_cmd, capture_output=True, text=True) import json probe_data = json.loads(probe_result.stdout) actual_duration = float(probe_data.get('format', {}).get('duration', 0)) original_sr = int(probe_data.get('streams', [{}])[0].get('sample_rate', 48000)) except Exception as e2: logger.warning(f"ffprobe also failed: {e2}") original_sr = 48000 actual_duration = 0 logger.info(f" 📊 Raw audio: {original_sr}Hz, {actual_duration:.1f}s") # NOTE: Spectral analysis moved to AFTER intro skip (analyzes actual speech content) # === v3: Use Supabase metadata for chapter-based intro skip === # Chapter info now comes from Supabase (synced from CSV), NOT yt-dlp intro_skip = 0.0 outro_skip = 0.0 skip_reason = "none" matched_title = None video_title = f'R2 Source: {video_id}' # Use Supabase metadata if provided if supabase_metadata: video_title = supabase_metadata.get('title') or video_title chapters = supabase_metadata.get('chapters') or [] has_chapters = supabase_metadata.get('has_chapters', False) source_duration = supabase_metadata.get('source_duration_min', 0) * 60 if supabase_metadata.get('source_duration_min') else actual_duration if chapters and has_chapters: logger.info(f" 📖 Using Supabase chapters ({len(chapters)} chapters)") # Convert chapter format if needed (Supabase stores as JSON with 'seconds' key) normalized_chapters = [] for i, ch in enumerate(chapters): start_sec = ch.get('seconds', ch.get('start_time', 0)) # Calculate end_time from next chapter or total duration if i + 1 < len(chapters): end_sec = chapters[i + 1].get('seconds', chapters[i + 1].get('start_time', source_duration)) else: end_sec = source_duration normalized_chapters.append({ 'title': ch.get('title', ''), 'start_time': start_sec, 'end_time': end_sec, }) # Use chapter-based intro detection intro_skip, outro_skip, skip_reason, matched_title = detect_intro_outro_from_chapters( normalized_chapters, source_duration, config ) logger.info(f" 📖 Chapter-based skip: intro={intro_skip:.1f}s, reason={skip_reason}") else: # No chapters in Supabase, use dynamic skip logger.info(f" 📊 No chapters in Supabase, using dynamic skip") if actual_duration > 0: # FIX: Removed 10% cap that was overriding 3min/5min logic # The get_dynamic_intro_skip function already implements proper duration-based skip intro_skip = get_dynamic_intro_skip(actual_duration) skip_reason = "dynamic:no_supabase_chapters" logger.info(f" ⏱️ Dynamic skip: {intro_skip:.0f}s") else: # No Supabase metadata provided, use dynamic skip logger.info(f" ⚠️ No Supabase metadata, using dynamic skip") if actual_duration > 0 and getattr(config, 'auto_intro_skip', True) and getattr(config, 'intro_skip_seconds', 0.0) == 0.0: intro_skip = min(get_dynamic_intro_skip(actual_duration), actual_duration * 0.1) skip_reason = "dynamic:no_metadata" logger.info(f" ⏱️ Dynamic skip: {intro_skip:.0f}s") # === MANUAL OVERRIDE (Priority 3) === if getattr(config, 'intro_skip_seconds', 0.0) > 0 and intro_skip != config.intro_skip_seconds: logger.info(f"⚙️ Manual intro skip override: {config.intro_skip_seconds:.1f}s (was: {intro_skip:.1f}s)") intro_skip = config.intro_skip_seconds skip_reason = "manual_override" # Calculate trim with safety check start_time = intro_skip end_time = max(actual_duration - outro_skip, start_time + 10) # At least 10s of audio trim_duration = end_time - start_time # If duration unknown or very short, don't trim if actual_duration < 60 or trim_duration < 30: start_time = 0 trim_duration = actual_duration if actual_duration > 0 else 999999 # No trim intro_skip = 0 skip_reason = "none:too_short_or_unknown" # Step 2: Create dual output (16kHz processing + original FLAC) # CRITICAL FIX (v7.6): -ss and -t must be specified BEFORE EACH output! # FFmpeg multi-output does NOT apply input options to all outputs automatically. target_sr = config.sample_rate ffmpeg_dual = [ 'ffmpeg', '-y', '-i', str(raw_file), # Output 1: 16kHz mono WAV (with trim) '-ss', str(start_time), '-t', str(trim_duration), '-map', '0:a', '-ar', str(target_sr), '-ac', '1', str(trimmed_file), # Output 2: Original quality FLAC (SAME trim - repeat -ss/-t!) '-ss', str(start_time), '-t', str(trim_duration), '-map', '0:a', '-ac', '1', '-c:a', 'flac', '-compression_level', '5', str(original_file), ] logger.info(f" ⚡ FFmpeg dual output: 16kHz WAV + {original_sr}Hz FLAC (skip: {intro_skip:.0f}s)") try: result = subprocess.run(ffmpeg_dual, capture_output=True, text=True, check=True) except subprocess.CalledProcessError as e: raise DownloadError(f"FFmpeg dual output failed: {e.stderr[:500]}") # Verify outputs if not trimmed_file.exists(): raise DownloadError(f"FFmpeg did not create trimmed file") if not original_file.exists(): raise DownloadError(f"FFmpeg did not create original file") # === v7.6: CRITICAL VERIFICATION - Both files MUST have same duration === try: trimmed_info = get_audio_info(str(trimmed_file)) original_info = get_audio_info(str(original_file)) trimmed_duration = trimmed_info['duration'] original_duration_check = original_info['duration'] duration_diff = abs(trimmed_duration - original_duration_check) if duration_diff > 0.1: # Allow 0.1s tolerance logger.error(f" ❌ DURATION MISMATCH! trimmed={trimmed_duration:.2f}s, original={original_duration_check:.2f}s") raise DownloadError(f"Duration mismatch between outputs: {duration_diff:.2f}s difference") processed_duration = trimmed_duration logger.info(f" ✅ Duration verified: {trimmed_duration:.2f}s (diff={duration_diff:.3f}s)") except DownloadError: raise except Exception as e: logger.warning(f" ⚠️ Could not verify durations: {e}") processed_duration = trim_duration # Cleanup raw file try: raw_file.unlink() except: pass # === Spectral quality analysis (on TRIMMED content, not intro) === # This ensures we analyze actual speech, not intro music/silence spectral_quality = None spectral_info = {} try: from src.spectral_analysis import analyze_spectral_quality # Load original FLAC (intro-skipped) for analysis - better quality than 16kHz WAV orig_waveform, orig_sr = torchaudio.load(str(original_file)) spectral_quality = analyze_spectral_quality( orig_waveform.squeeze(0).numpy(), orig_sr ) logger.info(f" 🔬 Spectral: {spectral_quality.original_format_guess} " f"(rolloff={spectral_quality.rolloff_frequency:.0f}Hz)") spectral_info = { # Container vs actual quality 'claimed_sample_rate': orig_sr, # What the container says 'effective_sample_rate': spectral_quality.effective_sample_rate, # Estimated true quality 'detected_nyquist': round(spectral_quality.detected_nyquist, 0), # Where content ends 'rolloff_frequency': round(spectral_quality.rolloff_frequency, 0), # Energy rolloff point # Quality assessment 'is_upsampled': spectral_quality.is_upsampled, 'original_format': spectral_quality.original_format_guess, 'confidence': round(spectral_quality.confidence, 2), 'is_suitable_for_tts': spectral_quality.is_suitable_for_tts, 'quality_issues': spectral_quality.quality_issues, # Detailed metrics 'high_freq_energy_ratio': round(spectral_quality.high_freq_energy_ratio, 4), 'spectral_flatness': round(spectral_quality.spectral_flatness, 4), } del orig_waveform # Free memory except Exception as e: logger.warning(f" Spectral analysis skipped: {e}") elapsed = time.time() - start logger.info(f" ✅ R2 download complete: {elapsed:.1f}s, {processed_duration:.0f}s audio") # Build metadata - use Supabase info if available sb_chapters = (supabase_metadata.get('chapters') or []) if supabase_metadata else [] # Extract clean R2 path from presigned URL r2_path = extract_r2_path(r2_url) return str(trimmed_file), { 'video_id': video_id, 'youtube_url': f'https://www.youtube.com/watch?v={video_id}', 'title': video_title, 'original_duration': actual_duration, 'intro_skipped': intro_skip, 'intro_skip_reason': skip_reason, 'intro_skip_matched_title': matched_title, 'processed_duration': processed_duration, 'output_dir': str(output_dir), 'chapters': sb_chapters, 'sample_rate': target_sr, 'original_sample_rate': original_sr, 'original_audio_path': str(original_file), # Local path to FLAC (for cutting) 'source_url': r2_path, # Clean R2 path: bucket/key (e.g., test/podcasts/VIDEO.webm) 'original_audio_preserved': True, 'audio_format': 'flac', 'source_type': 'r2', 'spectral_quality': spectral_info, # audio_native_sample_rate = actual container sample rate (from ffprobe) # This is the REAL sample rate, not the spectral-estimated effective rate 'audio_native_sample_rate': original_sr, # is_native_quality = whether spectral analysis suggests upsampling 'is_native_quality': not spectral_info.get('is_upsampled', False), } def download_video_for_visualization(url: str, output_dir: Path, video_id: str, max_retries: int = 3) -> Optional[Path]: """Download video for visualization (720p MP4).""" logger.info(f"🎥 Downloading video for visualization: {url}") start = time.time() output_file = output_dir / f"{video_id}.mp4" if output_file.exists() and output_file.stat().st_size > 1_000_000: logger.info(f"✅ Using cached video: {output_file}") return output_file ydl_opts = { 'format': 'bestvideo[height<=720]+bestaudio/best[height<=720]', 'outtmpl': str(output_dir / f"{video_id}.%(ext)s"), 'merge_output_format': 'mp4', 'quiet': True, 'no_warnings': True, 'retries': 3, 'fragment_retries': 3, } # Add proxy if configured proxy = get_ytdlp_proxy() if proxy: ydl_opts['proxy'] = proxy for attempt in range(max_retries): try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.extract_info(url, download=True) if output_file.exists(): logger.info(f"✅ Video download: {time.time()-start:.1f}s | Size: {output_file.stat().st_size/1e6:.1f}MB") return output_file except Exception as e: logger.warning(f"Video download attempt {attempt+1} failed: {e}") if attempt < max_retries - 1: time.sleep(2 ** attempt) logger.error(f"❌ Video download failed after {max_retries} attempts") return None def download_audio( video_url: str, config, validate: bool = True, max_retries: int = 3, supabase_metadata: Dict[str, Any] = None, ) -> Tuple[str, Dict[str, Any]]: """ Download audio from YouTube URL or R2 presigned URL and prepare for processing. === v7.0 OPTIMIZATION: Download-Time Resample === Uses ffmpeg directly for dual-output in single pass: - 16kHz mono WAV for processing pipeline - Original quality mono WAV for high-quality export Previous: yt-dlp → temp.wav → torchaudio load → Python resample → save both Now: yt-dlp → single ffmpeg → dual output (NO Python resampling!) === v2: R2 URL SUPPORT === When video_url is an R2 presigned URL, we use ffmpeg directly instead of yt-dlp. === v3: Supabase Metadata === When supabase_metadata is provided (from distributed queue), use it for: - Title, chapters, duration (no yt-dlp metadata fetch needed) - Chapter-based intro/outro skip Args: video_url: YouTube URL or R2 presigned URL config: Pipeline configuration validate: Whether to validate video before download (default: True) max_retries: Number of download retries (default: 3) supabase_metadata: Pre-fetched metadata from Supabase (chapters, title, etc.) Returns: (audio_path, metadata): Path to processed audio and video metadata Raises: VideoValidationError: If video fails validation DownloadError: If download fails after retries """ logger.info(f"📥 Downloading: {video_url[:80]}...") start = time.time() # Extract video ID video_id = extract_video_id(video_url) output_dir = Path(config.output_dir) / video_id output_dir.mkdir(parents=True, exist_ok=True) # === R2 SOURCE MODE (video_id -> R2 object) === # For throughput benchmarking we want to avoid YouTube rate limits entirely. # When enabled, we resolve the video_id to an object in the R2 *source* bucket # (bucket_type='source' -> bucket name "test") using HEAD requests only. input_source = getattr(config, 'input_source', 'youtube') if input_source == 'r2' and not is_r2_url(video_url): client = _get_r2_source_client(config) remote_key, head = _resolve_r2_source_object(video_id, config) # Presigned URL lets ffmpeg stream directly (no local video download). ttl = int(getattr(config, 'r2_presign_ttl_sec', 3600)) presigned_url = client.generate_presigned_get_url(remote_key, expires_in_sec=ttl) logger.info( f"📦 R2 source resolved: s3://{client.bucket}/{remote_key} " f"({head.get('ContentLength', 0) / (1024**2):.1f} MB)" ) audio_path, metadata = _download_from_r2( r2_url=presigned_url, video_id=video_id, output_dir=output_dir, config=config, max_retries=max_retries, supabase_metadata=supabase_metadata, ) elapsed = time.time() - start metadata['download_wall_time_sec'] = elapsed metadata['r2_source_bucket'] = client.bucket metadata['r2_source_key'] = remote_key metadata['r2_source_size_bytes'] = head.get('ContentLength') logger.info(f"✅ Download (R2): {elapsed:.1f}s | Duration: {metadata.get('processed_duration', 0):.0f}s") return audio_path, metadata # === v2: R2 URL HANDLING === # R2 presigned URLs use ffmpeg directly (not yt-dlp) if is_r2_url(video_url): logger.info(f"📦 R2 source detected, using ffmpeg direct download") audio_path, metadata = _download_from_r2( r2_url=video_url, video_id=video_id, output_dir=output_dir, config=config, max_retries=max_retries, supabase_metadata=supabase_metadata, ) elapsed = time.time() - start metadata['download_wall_time_sec'] = elapsed logger.info(f"✅ Download (R2 URL): {elapsed:.1f}s | Duration: {metadata.get('processed_duration', 0):.0f}s") return audio_path, metadata # === Standard YouTube path === info = {} if validate: is_valid, message, info = validate_video(video_url) if not is_valid: raise VideoValidationError(f"Video validation failed: {message}") # Fetch metadata if not already fetched during validation if not info: logger.info("📋 Fetching video metadata...") ydl_info_opts = { 'quiet': True, 'no_warnings': True, 'extract_flat': False, } try: with yt_dlp.YoutubeDL(ydl_info_opts) as ydl: info = ydl.extract_info(video_url, download=False) except Exception as e: logger.warning(f"Could not fetch metadata: {e}") info = {'title': 'Unknown', 'chapters': []} # Output file paths # v7.2: Original audio saved as FLAC (lossless, ~50% smaller than WAV) trimmed_file = output_dir / f"{video_id}_trimmed.wav" original_file = output_dir / f"{video_id}_original.flac" # Check cache - both files must exist for full cache hit preserve_original = getattr(config, 'preserve_original_audio', False) # Also check for legacy .wav original files (backward compatibility) legacy_original_file = output_dir / f"{video_id}_original.wav" if trimmed_file.exists(): cache_valid = True # Accept either .flac or .wav for original actual_original = original_file if original_file.exists() else (legacy_original_file if legacy_original_file.exists() else None) if preserve_original and not actual_original: cache_valid = False logger.info("⚠️ Cached trimmed file exists but original missing, re-downloading") if cache_valid: logger.info(f"✅ Using cached audio: {trimmed_file}") try: waveform, sr = torchaudio.load(str(trimmed_file)) # === EDGE CASE: Validate cached file isn't corrupted === if waveform.shape[1] < sr * 10: # Less than 10 seconds logger.warning(f"⚠️ Cached file seems corrupted or too short, re-downloading") trimmed_file.unlink() if actual_original and actual_original.exists(): actual_original.unlink() else: # Get original sample rate from original file if it exists orig_sr = sr if actual_original and actual_original.exists(): try: orig_info = get_audio_info(str(actual_original)) orig_sr = orig_info['sample_rate'] except: pass return str(trimmed_file), { 'video_id': video_id, 'youtube_url': f'https://www.youtube.com/watch?v={video_id}', 'title': info.get('title', 'Cached'), 'original_duration': waveform.shape[1] / sr, 'intro_skipped': 0.0, 'processed_duration': waveform.shape[1] / sr, 'output_dir': str(output_dir), 'chapters': info.get('chapters', []), 'sample_rate': sr, 'original_sample_rate': orig_sr, 'original_audio_path': str(actual_original) if actual_original else None, 'original_audio_preserved': actual_original is not None, 'audio_format': 'flac' if actual_original and str(actual_original).endswith('.flac') else 'wav', } except Exception as e: logger.warning(f"⚠️ Failed to load cached file: {e}, re-downloading") try: trimmed_file.unlink() if actual_original and actual_original.exists(): actual_original.unlink() except: pass # Get video duration for intro/outro detection original_duration = info.get('duration', 0) chapters = info.get('chapters', []) if info else [] intro_skip, outro_skip, skip_reason, matched_title = detect_intro_outro_from_chapters(chapters, original_duration, config) # === MANUAL OVERRIDE (Priority 3) === if config.intro_skip_seconds > 0 and intro_skip != config.intro_skip_seconds: logger.info(f"⚙️ Manual intro skip override: {config.intro_skip_seconds:.1f}s (was: {intro_skip:.1f}s)") intro_skip = config.intro_skip_seconds # === v7.0 OPTIMIZATION: Single-pass ffmpeg dual output === # Download audio stream and process with ffmpeg in one command if preserve_original: result = _download_with_ffmpeg_dual_output( video_url=video_url, video_id=video_id, output_dir=output_dir, trimmed_file=trimmed_file, original_file=original_file, target_sr=config.sample_rate, intro_skip=intro_skip, outro_skip=outro_skip, original_duration=original_duration, max_retries=max_retries, info=info, skip_reason=skip_reason, matched_title=matched_title ) else: # Standard download without original preservation result = _download_standard( video_url=video_url, video_id=video_id, output_dir=output_dir, trimmed_file=trimmed_file, config=config, intro_skip=intro_skip, outro_skip=outro_skip, original_duration=original_duration, max_retries=max_retries, info=info ) elapsed = time.time() - start logger.info(f"✅ Download: {elapsed:.1f}s | Duration: {result['processed_duration']:.0f}s") result['download_wall_time_sec'] = elapsed return str(trimmed_file), result def _download_with_ffmpeg_dual_output( video_url: str, video_id: str, output_dir: Path, trimmed_file: Path, original_file: Path, target_sr: int, intro_skip: float, outro_skip: float, original_duration: float, max_retries: int, info: dict, skip_reason: str = None, matched_title: str = None ) -> Dict[str, Any]: """ === v7.2 OPTIMIZATION: Download-Time Dual Output with FLAC === Single ffmpeg command creates both outputs: 1. 16kHz mono WAV (for processing) - with intro/outro trim 2. Original quality mono FLAC (for export) - LOSSLESS, ~50% smaller than WAV FLAC Benefits: - 100% lossless (bit-perfect audio quality) - ~50% smaller than WAV (saves storage costs) - Fast decode (CPU-efficient for training) - Compression level 5 (balanced speed vs size) Key insight: ffmpeg can output to multiple files from single input stream, encoding each to different formats in one pass. """ logger.info("⚡ Using optimized dual-output download with FLAC (v7.2)") # Temp file for raw download raw_file = output_dir / f"{video_id}_raw.wav" # Step 1: Download audio at original quality using yt-dlp # NO resampling in yt-dlp - we'll do it in ffmpeg ydl_opts = { 'format': 'bestaudio/best', 'outtmpl': str(output_dir / f"{video_id}_raw.%(ext)s"), 'quiet': True, 'no_warnings': True, 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'wav', 'preferredquality': '0', # Best quality }], # Only convert to mono, preserve original sample rate 'postprocessor_args': ['-ac', '1'], 'retries': 3, 'fragment_retries': 3, } # Add proxy if configured proxy = get_ytdlp_proxy() if proxy: ydl_opts['proxy'] = proxy last_error = None for attempt in range(max_retries): try: logger.info(f" Download attempt {attempt + 1}/{max_retries}") with yt_dlp.YoutubeDL(ydl_opts) as ydl: dl_info = ydl.extract_info(video_url, download=True) # Find the downloaded file if not raw_file.exists(): for ext_file in output_dir.glob(f"{video_id}_raw.*"): if ext_file.suffix in ['.wav', '.mp3', '.m4a', '.opus', '.webm']: if ext_file.suffix != '.wav': # Convert to wav first subprocess.run([ 'ffmpeg', '-i', str(ext_file), '-ac', '1', str(raw_file), '-y' ], capture_output=True, check=True) ext_file.unlink() else: ext_file.rename(raw_file) break if raw_file.exists(): break else: raise DownloadError(f"Downloaded file not found") except Exception as e: last_error = e logger.warning(f" Download attempt {attempt + 1} failed: {e}") if attempt < max_retries - 1: time.sleep(2 ** attempt) continue if not raw_file.exists(): raise DownloadError(f"Download failed after {max_retries} attempts: {last_error}") # Get actual audio info try: audio_info = get_audio_info(str(raw_file)) original_sr = audio_info['sample_rate'] actual_duration = audio_info['duration'] except Exception as e: logger.warning(f"Could not get audio info: {e}, using metadata duration") original_sr = 48000 # Common default actual_duration = original_duration logger.info(f" 📊 Raw audio: {original_sr}Hz, {actual_duration:.1f}s") # === v7.3: SPECTRAL QUALITY ANALYSIS === # Detect if audio is upsampled (e.g., 44.1kHz source at 48kHz container) # This info is crucial for quality filtering and proper training data curation spectral_quality = None try: from src.spectral_analysis import analyze_spectral_quality import torchaudio raw_waveform, _ = torchaudio.load(str(raw_file)) spectral_quality = analyze_spectral_quality( raw_waveform.squeeze(0).numpy(), original_sr ) logger.info(f" 🔬 Spectral: {spectral_quality.original_format_guess} " f"(effective={spectral_quality.effective_sample_rate}Hz, " f"rolloff={spectral_quality.rolloff_frequency:.0f}Hz)") if spectral_quality.quality_issues: logger.warning(f" ⚠️ Quality: {', '.join(spectral_quality.quality_issues)}") except Exception as e: logger.warning(f" Spectral analysis skipped: {e}") # Calculate trim parameters start_time = intro_skip if intro_skip > 0 else 0 end_time = actual_duration - outro_skip if outro_skip > 0 else actual_duration trim_duration = end_time - start_time # Step 2: Single ffmpeg command for dual output with trim # CRITICAL FIX (v7.6): -ss and -t must be specified BEFORE EACH output! # FFmpeg multi-output does NOT apply input options to all outputs automatically. # Without this fix, only the first output gets trimmed, causing timestamp misalignment. ffmpeg_cmd = [ 'ffmpeg', '-i', str(raw_file), # Output 1: 16kHz mono WAV for processing pipeline '-ss', str(start_time), # Trim BEFORE this output '-t', str(trim_duration), '-map', '0:a', '-ar', str(target_sr), '-ac', '1', '-y', str(trimmed_file), # Output 2: Original quality mono FLAC for export/training # Must repeat -ss/-t for this output too! '-ss', str(start_time), '-t', str(trim_duration), '-map', '0:a', '-ac', '1', '-c:a', 'flac', '-compression_level', '5', '-y', str(original_file) ] logger.info(f" ⚡ FFmpeg dual output: 16kHz WAV + {original_sr}Hz FLAC (trim: {start_time:.1f}s-{end_time:.1f}s)") try: result = subprocess.run(ffmpeg_cmd, capture_output=True, text=True, check=True) except subprocess.CalledProcessError as e: logger.error(f"FFmpeg failed: {e.stderr}") raise DownloadError(f"FFmpeg dual output failed: {e}") # Verify outputs if not trimmed_file.exists(): raise DownloadError(f"FFmpeg did not create trimmed file: {trimmed_file}") if not original_file.exists(): raise DownloadError(f"FFmpeg did not create original file: {original_file}") # === v7.6: CRITICAL VERIFICATION - Both files MUST have same duration === # This catches FFmpeg multi-output bugs where trim only applies to first output try: trimmed_info = get_audio_info(str(trimmed_file)) original_info = get_audio_info(str(original_file)) trimmed_duration = trimmed_info['duration'] original_duration_check = original_info['duration'] duration_diff = abs(trimmed_duration - original_duration_check) # Allow up to 0.1s difference (codec/rounding tolerance) if duration_diff > 0.1: logger.error(f" ❌ DURATION MISMATCH! trimmed={trimmed_duration:.2f}s, original={original_duration_check:.2f}s (diff={duration_diff:.2f}s)") logger.error(f" This indicates FFmpeg trim was not applied to both outputs!") raise DownloadError(f"Duration mismatch between outputs: {duration_diff:.2f}s difference") processed_duration = trimmed_duration logger.info(f" ✅ Duration verified: trimmed={trimmed_duration:.2f}s, original={original_duration_check:.2f}s (diff={duration_diff:.3f}s)") except DownloadError: raise except Exception as e: logger.warning(f" ⚠️ Could not verify durations: {e}") processed_duration = trim_duration # Cleanup raw file try: raw_file.unlink() except: pass # Cleanup any other temp files for temp in output_dir.glob(f"{video_id}_raw.*"): try: temp.unlink() except: pass logger.info(f" ✅ Dual output complete: {trimmed_file.name} ({target_sr}Hz WAV), {original_file.name} ({original_sr}Hz FLAC)") # Build spectral quality dict for metadata spectral_info = {} if spectral_quality is not None: spectral_info = { # Container vs actual quality 'claimed_sample_rate': original_sr, # What the container says 'effective_sample_rate': spectral_quality.effective_sample_rate, # Estimated true quality 'detected_nyquist': round(spectral_quality.detected_nyquist, 0), # Where content ends 'rolloff_frequency': round(spectral_quality.rolloff_frequency, 0), # Energy rolloff point # Quality assessment 'is_upsampled': spectral_quality.is_upsampled, 'original_format': spectral_quality.original_format_guess, 'confidence': round(spectral_quality.confidence, 2), 'is_suitable_for_tts': spectral_quality.is_suitable_for_tts, 'quality_issues': spectral_quality.quality_issues, # Detailed metrics 'high_freq_energy_ratio': round(spectral_quality.high_freq_energy_ratio, 4), 'spectral_flatness': round(spectral_quality.spectral_flatness, 4), } return { 'video_id': video_id, 'youtube_url': f'https://www.youtube.com/watch?v={video_id}', 'title': info.get('title', 'Unknown') if info else 'Unknown', 'original_duration': actual_duration, 'intro_skipped': intro_skip, 'processed_duration': processed_duration, 'output_dir': str(output_dir), 'chapters': info.get('chapters', []) if info else [], 'sample_rate': target_sr, 'original_sample_rate': original_sr, 'original_audio_path': str(original_file), # Local path to FLAC 'source_url': video_url, # YouTube URL (for reference) 'original_audio_preserved': True, 'audio_format': 'flac', 'audio_compression': 'lossless', 'source_type': 'youtube', # v7.3: Spectral quality info 'spectral_quality': spectral_info, # audio_native_sample_rate = ACTUAL container sample rate from ffprobe (reliable) # NOT spectral effective_sample_rate (which is an estimate that can be wrong) 'audio_native_sample_rate': original_sr, 'is_native_quality': not spectral_info.get('is_upsampled', False), # === v2: Intro skip metadata === 'intro_skip_reason': skip_reason, 'intro_skip_matched_title': matched_title, } def _download_standard( video_url: str, video_id: str, output_dir: Path, trimmed_file: Path, config, intro_skip: float, outro_skip: float, original_duration: float, max_retries: int, info: dict ) -> Dict[str, Any]: """Standard download without original preservation (16kHz only).""" temp_file = output_dir / f"{video_id}_temp.%(ext)s" downloaded_file = output_dir / f"{video_id}_temp.wav" original_file = output_dir / f"{video_id}_original.wav" # v6.8: High-quality preservation # === v6.8: HIGH-QUALITY AUDIO PRESERVATION === # If preserve_original_audio is True, we download at original quality first, # then create a 16kHz copy for processing. This allows cutting from original later. preserve_original = getattr(config, 'preserve_original_audio', False) original_sr_target = getattr(config, 'original_audio_sample_rate', 0) # 0 = keep original if preserve_original: # Download WITHOUT resampling to preserve original quality ydl_opts = { 'format': 'bestaudio/best', 'outtmpl': str(temp_file), 'quiet': True, 'no_warnings': True, 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'wav', 'preferredquality': '0', }], # Only mono conversion, NO sample rate change 'postprocessor_args': ['-ac', '1'] if original_sr_target == 0 else ['-ar', str(original_sr_target), '-ac', '1'], 'retries': 3, 'fragment_retries': 3, } else: # Standard: Download at 16kHz for processing ydl_opts = { 'format': 'bestaudio/best', 'outtmpl': str(temp_file), 'quiet': True, 'no_warnings': True, 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'wav', 'preferredquality': '0', }], 'postprocessor_args': ['-ar', str(config.sample_rate), '-ac', '1'], 'retries': 3, 'fragment_retries': 3, } last_error = None for attempt in range(max_retries): try: logger.info(f" Download attempt {attempt + 1}/{max_retries}") with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.extract_info(video_url, download=True) if not downloaded_file.exists(): for ext_file in output_dir.glob(f"{video_id}_temp.*"): if ext_file.suffix in ['.wav', '.mp3', '.m4a', '.opus', '.webm']: if ext_file.suffix != '.wav': subprocess.run([ 'ffmpeg', '-i', str(ext_file), '-ar', str(config.sample_rate), '-ac', '1', str(downloaded_file), '-y' ], capture_output=True, check=True) ext_file.unlink() else: ext_file.rename(downloaded_file) break if downloaded_file.exists(): break except Exception as e: last_error = e logger.warning(f" Download attempt {attempt + 1} failed: {e}") if attempt < max_retries - 1: time.sleep(2 ** attempt) continue if not downloaded_file.exists(): raise DownloadError(f"Download failed after {max_retries} attempts: {last_error}") # Load and apply trim try: waveform, sr = torchaudio.load(str(downloaded_file)) except Exception as e: if downloaded_file.exists(): downloaded_file.unlink() raise DownloadError(f"Audio file corrupted: {e}") original_sr = sr # Keep track of original sample rate original_duration = waveform.shape[1] / sr actual_duration = original_duration # For backward compatibility # === v6.8: HIGH-QUALITY AUDIO PRESERVATION === # If preserving original audio, we need to: # 1. Save the original quality audio # 2. Create a 16kHz resampled version for processing original_audio_path = None if preserve_original and sr != config.sample_rate: logger.info(f" 🎵 Preserving original audio: {sr}Hz → {original_file.name}") original_audio_path = str(original_file) # Save original quality audio (will be trimmed with same intro/outro later) # Note: We save after trim to keep timestamps consistent # Resample to 16kHz for processing pipeline import torchaudio.transforms as T resampler = T.Resample(orig_freq=sr, new_freq=config.sample_rate) waveform = resampler(waveform) sr = config.sample_rate logger.info(f" 📊 Created processing copy: {sr}Hz") # === EDGE CASE: Very short audio after download === if original_duration < 30: logger.warning(f"⚠️ Audio very short: {original_duration:.1f}s") # Detect intro/outro using YouTube chapters (smart detection!) # Priority: 1) Chapter-based, 2) Dynamic (duration-based), 3) Manual config chapters = info.get('chapters', []) if info else [] intro_skip, outro_skip, _, _ = detect_intro_outro_from_chapters(chapters, original_duration, config) # === MANUAL OVERRIDE (Priority 3) === # If user explicitly set intro_skip_seconds > 0, it overrides everything if config.intro_skip_seconds > 0 and intro_skip != config.intro_skip_seconds: logger.info(f"⚙️ Manual intro skip override: {config.intro_skip_seconds:.1f}s (was: {intro_skip:.1f}s)") intro_skip = config.intro_skip_seconds # Apply intro/outro trimming intro_skipped = intro_skip # Track actual skip for original audio trim intro_samples = int(intro_skip * sr) outro_samples = int(outro_skip * sr) if outro_skip > 0 else 0 end_sample = waveform.shape[1] - outro_samples if outro_skip > 0 else waveform.shape[1] if intro_samples > 0 or outro_samples > 0: waveform = waveform[:, intro_samples:end_sample] if waveform.shape[0] > 1: waveform = waveform.mean(dim=0, keepdim=True) # === EDGE CASE: Check for silent/empty audio === rms = (waveform ** 2).mean().sqrt().item() if rms < 0.0001: logger.warning(f"⚠️ Audio appears to be silent (RMS={rms:.6f})") # Save processed audio (16kHz for pipeline processing) torchaudio.save(str(trimmed_file), waveform, sr) # === v6.8: Save original quality audio with same trim === original_audio_saved = None if preserve_original and downloaded_file.exists(): try: # Load original again and apply same trim orig_waveform, orig_sr = torchaudio.load(str(downloaded_file)) # Ensure mono if orig_waveform.shape[0] > 1: orig_waveform = orig_waveform.mean(dim=0, keepdim=True) # Apply same intro/outro trim (in original sample rate) orig_intro_samples = int(intro_skipped * orig_sr) orig_outro_samples = int(outro_skip * orig_sr) if outro_skip > 0 else 0 orig_end_sample = orig_waveform.shape[1] - orig_outro_samples if outro_skip > 0 else orig_waveform.shape[1] orig_waveform = orig_waveform[:, orig_intro_samples:orig_end_sample] # Save original quality torchaudio.save(str(original_file), orig_waveform, orig_sr) original_audio_saved = str(original_file) logger.info(f" ✅ Saved original quality: {original_file.name} ({orig_sr}Hz, {orig_waveform.shape[1]/orig_sr:.1f}s)") except Exception as e: logger.warning(f" ⚠️ Failed to save original audio: {e}") # Cleanup temp file if downloaded_file.exists(): downloaded_file.unlink() for temp in output_dir.glob(f"{video_id}_temp.*"): try: temp.unlink() except: pass return { 'video_id': video_id, 'youtube_url': f'https://www.youtube.com/watch?v={video_id}', 'title': info.get('title', 'Unknown') if info else 'Unknown', 'original_duration': actual_duration, 'intro_skipped': intro_skip, 'processed_duration': waveform.shape[1] / sr, 'output_dir': str(output_dir), 'chapters': chapters, # === v6.8: High-quality audio info === 'sample_rate': sr, # Processing sample rate (16kHz) 'original_sample_rate': original_sr if preserve_original else sr, # Original audio sample rate 'original_audio_path': original_audio_saved, # Path to original quality audio (if preserved) 'original_audio_preserved': original_audio_saved is not None, # === v7.5: Additional audio metadata for database === 'acodec': info.get('acodec', 'unknown') if info else 'unknown', 'abr': info.get('abr') if info else None, 'channels': info.get('audio_channels', 2) if info else 2, 'format': info.get('format', 'unknown') if info else 'unknown', }