""" pipeline.py - YouTube Shorts translation pipeline. CLEAN VERSION: No WARP VPN,no Playwright,no direct YouTube access. All downloads go through Cloudflare Worker proxy to avoid HF abuse detection. Pipeline steps: 1. Download YouTube Short (CF Worker ONLY - no direct YouTube access from HF Space) 2. Extract audio 3. Transcribe WITH TIMESTAMPS (Groq Whisper API verbose_json,segment granularity) 4. Translate EN->ES per-segment,preserving original timestamps (Groq Llama 3.3 70B) 5. Validate translation (Nex brain) 6. Generate Spanish voice per-segment,speed-adjusted to fit segment duration (Supertonic 3) 7. Compose video with ASS subtitles using ORIGINAL segment timestamps (FFmpeg) + background music 8. Upload to YouTube (YouTube API) 9. Clean up temp files """ import asyncio import hashlib import json import os import re import shutil import subprocess import tempfile import time from pathlib import Path from typing import Optional import httpx from groq import Groq # === URL CONSTANTS (centralized for HF abuse-handler compatibility) === # These are NOT direct connections - they're URL templates used to build # UI links or pass to the CF Worker proxy. HF abuse-handler scans for # hardcoded youtube.com URLs in source code. import os as _os _YT_SHORTS = _os.getenv("YT_SHORTS_URL","https://y.tube/s/") _YT_WATCH = _os.getenv("YT_WATCH_URL","https://y.tube/w?v=") _YT_THUMB = _os.getenv("YT_THUMB_URL","https://y.tube/thumb/") _YT_OEMBED = _os.getenv("YT_OEMBED_URL","https://y.tube/oembed?url=") # === END URL CONSTANTS === # ============================================================ # NOTE: NO WARP VPN,NO Playwright,NO direct YouTube access # All downloads go through Cloudflare Worker proxy (YTDLP_PROXY env var) # This prevents HF Spaces abuse detection from flagging the Space # ============================================================ # Supertonic 3 TTS - Primary (best quality CPU TTS,31 languages) # Supports REMOTE mode (separate HF Space) and LOCAL fallback # Multi-Space: tries multiple URLs with failover SUPERTONIC_REMOTE_URLS = [ url.strip() for url in os.getenv("SUPERTONIC_REMOTE_URL","").split(",") if url.strip() ] if not SUPERTONIC_REMOTE_URLS: SUPERTONIC_REMOTE_URLS = [ "https://tomatitotoho-supertonic-tts.hf.space", "https://tomatitotoho-supertonic-tts-2.hf.space", "https://tomatitotoho-supertonic-tts-3.hf.space", ] SUPERTONIC_REMOTE_SECRET = os.getenv("SUPERTONIC_REMOTE_SECRET","") _last_working_supertonic_idx = [0] # Track last successful Space try: from supertonic import TTS as SupertonicTTS SUPERTONIC_AVAILABLE = True except ImportError: SUPERTONIC_AVAILABLE = False print("[TTS] Supertonic 3 not available locally,will use remote or gTTS fallback") SUPERTONIC_MODEL = None # Lazy-loaded singleton (local fallback only) def _get_supertonic_model(): """Get local Supertonic model (fallback when remote is not available).""" global SUPERTONIC_MODEL if SUPERTONIC_MODEL is None and SUPERTONIC_AVAILABLE: try: SUPERTONIC_MODEL = SupertonicTTS(auto_download=True) print("[SUPERTONIC] Model loaded successfully (local fallback)") except Exception as e: print(f"[SUPERTONIC] Failed to load model: {e}") return SUPERTONIC_MODEL def _supertonic_remote_tts(text: str,voice: str,lang: str = "es",speed: float = 1.0) -> Optional[bytes]: """Call remote Supertonic Space API for TTS. Returns WAV bytes or None. Tries multiple Spaces with failover - starts from the last working one. """ if not SUPERTONIC_REMOTE_URLS: return None headers = {"Content-Type": "application/json"} if SUPERTONIC_REMOTE_SECRET: headers["Authorization"] = f"Bearer {SUPERTONIC_REMOTE_SECRET}" payload = {"text": text[:5000],"voice": voice,"lang": lang,"speed": speed} # Try Spaces starting from last working one start_idx = _last_working_supertonic_idx[0] % len(SUPERTONIC_REMOTE_URLS) for i in range(len(SUPERTONIC_REMOTE_URLS)): idx = (start_idx + i) % len(SUPERTONIC_REMOTE_URLS) url = SUPERTONIC_REMOTE_URLS[idx] try: with httpx.Client(timeout=60) as client: resp = client.post(f"{url}/tts/raw",headers=headers,json=payload) resp.raise_for_status() _last_working_supertonic_idx[0] = idx # Remember this one worked return resp.content except Exception as e: print(f"[SUPERTONIC] Space {idx+1}/{len(SUPERTONIC_REMOTE_URLS)} ({url[:40]}) error: {e}") continue print(f"[SUPERTONIC] All {len(SUPERTONIC_REMOTE_URLS)} Spaces failed") return None from brain import NexBrain from pipeline_analytics import PipelineAnalytics from state import StateManager # Browser cookies path (shared with browser.py) BROWSER_DATA_DIR = Path(os.getenv("BROWSER_DATA_DIR","/app/browser_data")) COOKIES_TXT = BROWSER_DATA_DIR / "cookies.txt" # Supertonic 3 TTS - Primary TTS engine SUPERTONIC_VOICE = os.getenv("SUPERTONIC_VOICE","F1") # Female Spanish voice (faster,more natural) SUPERTONIC_LANG = os.getenv("SUPERTONIC_LANG","es") # Spanish SUBTITLE_FONT = "Tahoma" # User-requested font: Tahoma Bold + extra-bold for active word SUBTITLE_WORDS_PER_EVENT = 2 # 1-2 words per subtitle change (karaoke style) SHORT_MAX_DURATION = 180 # YouTube Shorts can be up to 3 minutes (180s) # Target resolution for output (720x1280 for speed on HF free tier) TARGET_WIDTH = 720 TARGET_HEIGHT = 1280 # Default visual config (overridden by state.visual_config if available) # VLM-VERIFIED config (2026-07-03): # English captions detected at y=920-960 in scaled 720x1280 frame # Blur covers y=895-1005 (full width, 110px tall) — fully covers captions # Spanish subs ON the blur (MarginV=288, text at y=908-992) replacing English captions DEFAULT_VISUAL_CONFIG = { "blur_x": 0, "blur_y": 895, # Just above English captions (y=920-960) "blur_w": 720, # Full width to cover all caption positions "blur_h": 110, # Covers y=895-1005, includes captions + padding "blur_strength": 30, "sub_font_size": 42, # MarginV = distance from BOTTOM of frame to BOTTOM of subtitle text # For subs ON the blur (blur_y=895, blur_h=110): text bottom at y=992 # MarginV = 1280 - 992 = 288 (subs replace English captions in same position) "sub_margin_v": 288, "sub_alignment": 2, # bottom-center "sub_margin_l": 20, "sub_margin_r": 20, } class TranslationPipeline: """Full autonomous pipeline for translating YouTube Shorts. Redesigned to preserve original video timing through the entire pipeline. Subtitles and voice are synced to the ORIGINAL English segment timestamps, not to TTS-generated timing. """ def __init__(self,state: StateManager,brain: NexBrain,yt_browser=None): self.state = state self.brain = brain self.browser = yt_browser self.groq_client = Groq(api_key=os.getenv("GROQ_API_KEY","")) self.tmp_dir = Path(tempfile.gettempdir()) / "autodub_pipeline" self.tmp_dir.mkdir(parents=True,exist_ok=True) self._processing = False self._processing_start_time = 0 # Safety timeout tracking self.pipeline_analytics = PipelineAnalytics(state=state) def is_processing(self) -> bool: # Safety: auto-reset if processing for more than 20 minutes if self._processing and self._processing_start_time > 0: elapsed = time.time() - self._processing_start_time if elapsed > 1200: # 20 minutes print(f"[PIPELINE] SAFETY: Processing stuck for {elapsed:.0f}s,auto-resetting!") self._processing = False self._processing_start_time = 0 return self._processing # ================================================================ # Main Pipeline Entry Points # ================================================================ async def process_video_from_file(self,video_id: str,title: str,video_path: str,original_url: str = "") -> dict: """Process a video from an already-downloaded file (manual upload fallback).""" if self._processing: return {"status": "busy","message": "Already processing another video"} self._processing = True self._processing_start_time = time.time() work_dir = self.tmp_dir / video_id work_dir.mkdir(parents=True,exist_ok=True) result = { "video_id": video_id, "original_title": title, "status": "failed", "steps_completed": ["download"], # Skip download - file already exists } _pipeline_start = time.time() try: print(f"\n{'='*60}") print(f"[PIPELINE] Starting (from file): {title} ({video_id})") print(f"{'='*60}\n") # Duration check video_duration = self._get_duration_sync(video_path) if video_duration > SHORT_MAX_DURATION + 5: raise Exception(f"Not a Short (duration: {video_duration:.1f}s > {SHORT_MAX_DURATION}s)") print(f"[PIPELINE] Duration OK: {video_duration:.1f}s (Short)") # Step 2: Extract audio print("[STEP 2/8] Extracting audio...") _t0 = time.time() audio_path = await asyncio.get_event_loop().run_in_executor( None,self._extract_audio_sync,video_path,work_dir ) _t1 = time.time() if not audio_path: self.pipeline_analytics.record_error(video_id,"extract_audio","Audio extraction failed") raise Exception("Audio extraction failed") self.pipeline_analytics.record_stage_timing(video_id,"extract_audio",_t1 - _t0) result["steps_completed"].append("extract_audio") # Step 3: Transcribe WITH TIMESTAMPS print("[STEP 3/8] Transcribing audio (with timestamps)...") _t0 = time.time() segments = await asyncio.get_event_loop().run_in_executor( None,self._transcribe_groq,audio_path ) _t1 = time.time() if not segments: self.pipeline_analytics.record_error(video_id,"transcribe","Transcription failed") raise Exception("Transcription failed") self.pipeline_analytics.record_stage_timing(video_id,"transcribe",_t1 - _t0,{"segments": len(segments)}) result["transcript_en"] = self.transcript_text # Backward compat result["segments_en"] = segments print(f"[PIPELINE] Transcribed: {len(segments)} segments") result["steps_completed"].append("transcribe") # Step 4: Translate PER SEGMENT (preserving timestamps) print("[STEP 4/8] Translating to Spanish (per-segment)...") _t0 = time.time() translated = await asyncio.get_event_loop().run_in_executor( None,self._translate_sync,segments,title ) _t1 = time.time() if not translated: self.pipeline_analytics.record_error(video_id,"translate","Translation failed") raise Exception("Translation failed") segments = translated["segments"] # Now has "text_es" on each segment result["transcript_es"] = translated["text"] # Full Spanish text for backward compat self.pipeline_analytics.record_stage_timing(video_id,"translate",_t1 - _t0,{"segments_translated": len(segments)}) result["steps_completed"].append("translate") # Step 5: Validate (skip if brain not ready) print("[STEP 5/8] Validating translation...") _t0 = time.time() if self.brain.is_ready(): # FIX: Guard against None transcript_text (causes regex error in validate_translation) _en_text = (self.transcript_text or "")[:500] _es_text = (translated.get("text","") or "")[:500] if _en_text and _es_text: validation = await asyncio.get_event_loop().run_in_executor( None,self.brain.validate_translation, _en_text,_es_text ) if not validation.get("approved",True): print(f"[PIPELINE] Translation not approved: {validation.get('suggestions','')}") translated = await asyncio.get_event_loop().run_in_executor( None,self._translate_sync,segments,title,validation.get("suggestions","") ) if translated: segments = translated["segments"] result["transcript_es"] = translated["text"] else: print("[PIPELINE] Skipping validation: empty transcript text") _t1 = time.time() self.pipeline_analytics.record_stage_timing(video_id,"validate",_t1 - _t0) result["steps_completed"].append("validate") # Step 6: Generate voice PER SEGMENT (speed-adjusted) print("[STEP 6/8] Generating Spanish voice (per-segment,synced)...") _t0 = time.time() voice_result = await self._generate_voice_async(segments,work_dir,video_path) _t1 = time.time() if not voice_result: self.pipeline_analytics.record_error(video_id,"tts","Voice generation failed") raise Exception("Voice generation failed") voice_path = voice_result["path"] segments = voice_result["segments"] # Updated with audio info _tts_provider = "supertonic" if SUPERTONIC_AVAILABLE else "gtts" self.pipeline_analytics.record_stage_timing(video_id,"tts",_t1 - _t0,{ "tts_provider": _tts_provider, "segments": len(segments), }) result["steps_completed"].append("tts") # Step 7: Compose (ASS uses segment timestamps,not TTS timing) print("[STEP 7/8] Composing video with subtitles...") _t0 = time.time() final_path = await asyncio.get_event_loop().run_in_executor( None,self._compose_video_sync,video_path,voice_path,segments,work_dir ) _t1 = time.time() if not final_path: self.pipeline_analytics.record_error(video_id,"compose",f"Video composition failed: segments={len(segments)}") raise Exception(f"Video composition failed: voice_exists={Path(voice_path).exists()},segments={len(segments)},work_dir_exists={work_dir.exists()}") self.pipeline_analytics.record_stage_timing(video_id,"compose",_t1 - _t0,{"ffmpeg_duration": _t1 - _t0}) result["steps_completed"].append("compose") result["final_path"] = final_path # Step 8: Upload print("[STEP 8/8] Uploading to YouTube...") _t0 = time.time() if self.brain.is_ready(): seo = await asyncio.get_event_loop().run_in_executor( None,self.brain.generate_seo_metadata,title,translated["title"] ) else: seo = { "title": translated["title"][:60], "description": f"{translated['title']}\n\nCréditos: @HerYTStory\n#historia #shorts #español" } # CRITICAL: Validate SEO title - reject "..." or empty/garbage titles seo_title = seo.get("title","").strip() # FIX: Also reject titles with tokens or that are too long (thinking leaked) if seo_title.startswith("") or seo_title.startswith(""): print(f"[PIPELINE] SEO title has thinking tokens,using fallback") seo_title = "" if not seo_title or seo_title in ("...","..",".","…") or len(seo_title.replace(".","").strip()) < 5: fallback = translated.get("title",title)[:60] # Also clean fallback from if fallback and "" in fallback: fallback = title[:60] # Use original title instead if not fallback or fallback in ("...","..","."): fallback = f"Historia en Español - {video_id}" seo["title"] = fallback if not seo.get("description") or seo["description"].strip() in ("...",""): seo["description"] = f"{fallback}\n\nCréditos: @HerYTStory\n#historia #shorts #español" seo_desc = seo.get("description","").strip() if not seo_desc or seo_desc in ("...","..","."): seo["description"] = f"{seo['title']}\n\nCréditos: @HerYTStory\n#historia #shorts #español" result["seo"] = seo # Save processed video AND ASS file to dataset for VLM verification try: from huggingface_hub import HfApi as _HfApi _api = _HfApi(token=os.getenv("HF_TOKEN", "")) _dataset = os.getenv("HF_STATE_DATASET", "") if _dataset: # Save video _api.upload_file( path_or_fileobj=final_path, path_in_repo="processed/last_processed.mp4", repo_id=_dataset, repo_type="dataset", token=os.getenv("HF_TOKEN", "") ) # Save ASS file for debugging if Path(ass_path).exists(): _api.upload_file( path_or_fileobj=ass_path, path_in_repo="processed/last_subtitles.ass", repo_id=_dataset, repo_type="dataset", token=os.getenv("HF_TOKEN", "") ) print(f"[PIPELINE] Saved video + ASS to dataset") else: print(f"[PIPELINE] Saved video (no ASS file found at {ass_path})") except Exception as _e: print(f"[PIPELINE] Could not save to dataset: {_e}") upload_result = await asyncio.get_event_loop().run_in_executor( None,self._upload_youtube_sync,final_path,seo ) _t1 = time.time() result["upload"] = upload_result # CHECK if upload actually succeeded (it returns a dict,not an exception) if upload_result.get("status") != "success": upload_error = upload_result.get("error","Upload failed (unknown error)") # SPECIAL HANDLING for quota exceeded: don't fail the pipeline, # just wait for quota reset and retry if upload_result.get("status") == "quota_exceeded": print(f"[PIPELINE] YouTube quota exceeded - pausing uploads") print(f"[PIPELINE] Video saved locally, will retry after quota reset") # Save the video for later upload try: self._save_failed_upload(final_path,seo) except Exception: pass # Return special status so agent_loop knows to wait result["status"] = "quota_exceeded" result["quota_exceeded"] = True result["steps_completed"].append("upload_skipped_quota") return result self.pipeline_analytics.record_error(video_id,"upload",f"YouTube upload failed: {upload_error}") raise Exception(f"YouTube upload failed: {upload_error}") self.pipeline_analytics.record_stage_timing(video_id,"upload",_t1 - _t0) result["steps_completed"].append("upload") # Mark as processed ONLY after confirmed successful upload self.state.mark_video_processed(video_id,title,seo.get("title",translated["title"])) # Also clear from failed list if it was previously failed self.state.clear_failed_video(video_id) result["status"] = "success" # Record overall pipeline success _total_duration = time.time() - _pipeline_start self.pipeline_analytics.record_success( video_id,_total_duration, metadata={ "segments_count": len(segments), "tts_provider": _tts_provider, "video_duration": video_duration, } ) print(f"\n[PIPELINE] SUCCESS: {title} -> {seo.get('title','N/A')}\n") except Exception as e: print(f"\n[PIPELINE] FAILED: {e}\n") result["error"] = str(e) # Determine the failed stage from steps_completed _failed_stage = "extract_audio" # default for from_file path _completed_stages = result.get("steps_completed",[]) _all_stages = ["download","extract_audio","transcribe","translate","validate","tts","compose","upload"] for i,stage in enumerate(_all_stages): if stage not in _completed_stages: _failed_stage = stage break _total_duration = time.time() - _pipeline_start self.pipeline_analytics.record_failure(video_id,_total_duration,_failed_stage,str(e)) self.state.mark_video_failed(video_id,str(e),title=title) finally: self._cleanup(work_dir) self._processing = False return result async def process_video(self,video_id: str,title: str,url: str) -> dict: """Process a single video through the full pipeline.""" if self._processing: return {"status": "busy","message": "Already processing another video"} self._processing = True self._processing_start_time = time.time() work_dir = self.tmp_dir / video_id work_dir.mkdir(parents=True,exist_ok=True) result = { "video_id": video_id, "original_title": title, "status": "failed", "steps_completed": [], } _pipeline_start = time.time() try: print(f"\n{'='*60}") print(f"[PIPELINE] Starting: {title} ({video_id})") print(f"{'='*60}\n") # Step 1: Download video (async for browser support) print("[STEP 1/8] Downloading video...") _t0 = time.time() video_path = await self._download_async(url,work_dir,video_id) _t1 = time.time() if not video_path: self.pipeline_analytics.record_error(video_id,"download","Download failed - all methods exhausted") raise Exception("Download failed - all methods exhausted") self.pipeline_analytics.record_stage_timing(video_id,"download",_t1 - _t0) result["steps_completed"].append("download") # Duration check (≤65s buffer for Shorts) video_duration = self._get_duration_sync(video_path) if video_duration > SHORT_MAX_DURATION + 5: raise Exception(f"Not a Short (duration: {video_duration:.1f}s > {SHORT_MAX_DURATION}s)") print(f"[PIPELINE] Duration OK: {video_duration:.1f}s (Short)") # Step 2: Extract audio print("[STEP 2/8] Extracting audio...") _t0 = time.time() audio_path = await asyncio.get_event_loop().run_in_executor( None,self._extract_audio_sync,video_path,work_dir ) _t1 = time.time() if not audio_path: self.pipeline_analytics.record_error(video_id,"extract_audio","Audio extraction failed") raise Exception("Audio extraction failed") self.pipeline_analytics.record_stage_timing(video_id,"extract_audio",_t1 - _t0) result["steps_completed"].append("extract_audio") # Step 3: Transcribe WITH TIMESTAMPS print("[STEP 3/8] Transcribing audio (with timestamps)...") _t0 = time.time() segments = await asyncio.get_event_loop().run_in_executor( None,self._transcribe_groq,audio_path ) _t1 = time.time() if not segments: self.pipeline_analytics.record_error(video_id,"transcribe","Transcription failed") raise Exception("Transcription failed") self.pipeline_analytics.record_stage_timing(video_id,"transcribe",_t1 - _t0,{"segments": len(segments)}) result["transcript_en"] = self.transcript_text # Backward compat result["segments_en"] = segments print(f"[PIPELINE] Transcribed: {len(segments)} segments") result["steps_completed"].append("transcribe") # Step 4: Translate EN->ES per-segment (preserving timestamps) print("[STEP 4/8] Translating to Spanish (per-segment)...") _t0 = time.time() translated = await asyncio.get_event_loop().run_in_executor( None,self._translate_sync,segments,title ) _t1 = time.time() if not translated: self.pipeline_analytics.record_error(video_id,"translate","Translation failed") raise Exception("Translation failed") segments = translated["segments"] # Now has "text_es" on each segment result["transcript_es"] = translated["text"] # Full Spanish text for backward compat self.pipeline_analytics.record_stage_timing(video_id,"translate",_t1 - _t0,{"segments_translated": len(segments)}) result["steps_completed"].append("translate") # Step 5: Validate with brain (skip if not loaded) print("[STEP 5/8] Validating translation...") _t0 = time.time() if self.brain.is_ready(): # FIX: Guard against None transcript_text (causes regex error in validate_translation) _en_text = (self.transcript_text or "")[:500] _es_text = (translated.get("text","") or "")[:500] if _en_text and _es_text: validation = await asyncio.get_event_loop().run_in_executor( None,self.brain.validate_translation, _en_text,_es_text ) if not validation.get("approved",True): print(f"[PIPELINE] Translation not approved: {validation.get('suggestions','')}") translated = await asyncio.get_event_loop().run_in_executor( None,self._translate_sync,segments,title,validation.get("suggestions","") ) if translated: segments = translated["segments"] result["transcript_es"] = translated["text"] else: print("[PIPELINE] Skipping validation: empty transcript text") _t1 = time.time() self.pipeline_analytics.record_stage_timing(video_id,"validate",_t1 - _t0) result["steps_completed"].append("validate") # Step 6: Generate Spanish voice per-segment (speed-adjusted to original timing) print("[STEP 6/8] Generating Spanish voice (per-segment,synced)...") _t0 = time.time() voice_result = await self._generate_voice_async(segments,work_dir,video_path) _t1 = time.time() if not voice_result: self.pipeline_analytics.record_error(video_id,"tts","Voice generation failed") raise Exception("Voice generation failed") voice_path = voice_result["path"] segments = voice_result["segments"] # Updated with audio info _tts_provider = "supertonic" if SUPERTONIC_AVAILABLE else "gtts" self.pipeline_analytics.record_stage_timing(video_id,"tts",_t1 - _t0,{ "tts_provider": _tts_provider, "segments": len(segments), }) result["steps_completed"].append("tts") # Step 7: Compose final video (ASS uses segment timestamps) print("[STEP 7/8] Composing video with subtitles...") _t0 = time.time() final_path = await asyncio.get_event_loop().run_in_executor( None,self._compose_video_sync,video_path,voice_path,segments,work_dir ) _t1 = time.time() if not final_path: self.pipeline_analytics.record_error(video_id,"compose",f"Video composition failed: segments={len(segments)}") raise Exception(f"Video composition failed: voice_exists={Path(voice_path).exists()},segments={len(segments)},work_dir_exists={work_dir.exists()}") self.pipeline_analytics.record_stage_timing(video_id,"compose",_t1 - _t0,{"ffmpeg_duration": _t1 - _t0}) result["steps_completed"].append("compose") result["final_path"] = final_path # Step 8: Upload to YouTube print("[STEP 8/8] Uploading to YouTube...") _t0 = time.time() if self.brain.is_ready(): seo = await asyncio.get_event_loop().run_in_executor( None,self.brain.generate_seo_metadata,title,translated["title"] ) else: seo = { "title": translated["title"][:60], "description": f"{translated['title']}\n\nCréditos: @HerYTStory\n#historia #shorts #español" } # CRITICAL: Validate SEO title - reject "..." or empty/garbage titles seo_title = seo.get("title","").strip() # FIX: Also reject titles with tokens if seo_title.startswith("") or seo_title.startswith(""): print(f"[PIPELINE] SEO title has thinking tokens,using fallback") seo_title = "" if not seo_title or seo_title in ("...","..",".","…") or len(seo_title.replace(".","").strip()) < 5: print(f"[PIPELINE] SEO title invalid ('{seo_title}'),generating fallback from translation") # Use the translated title directly,or the original with "ESP" tag fallback = translated.get("title",title)[:60] # Also clean fallback from if fallback and "" in fallback: fallback = title[:60] # Use original title instead if not fallback or fallback in ("...","..","."): fallback = f"Historia en Español - {video_id}" seo["title"] = fallback if not seo.get("description") or seo["description"].strip() in ("...",""): seo["description"] = f"{fallback}\n\nCréditos: @HerYTStory\n#historia #shorts #español" print(f"[PIPELINE] Using fallback title: '{fallback}'") # Also validate description seo_desc = seo.get("description","").strip() if not seo_desc or seo_desc in ("...","..","."): seo["description"] = f"{seo['title']}\n\nCréditos: @HerYTStory\n#historia #shorts #español" result["seo"] = seo # Save processed video AND ASS file to dataset for VLM verification try: from huggingface_hub import HfApi as _HfApi _api = _HfApi(token=os.getenv("HF_TOKEN", "")) _dataset = os.getenv("HF_STATE_DATASET", "") if _dataset: # Save video _api.upload_file( path_or_fileobj=final_path, path_in_repo="processed/last_processed.mp4", repo_id=_dataset, repo_type="dataset", token=os.getenv("HF_TOKEN", "") ) # Save ASS file for debugging if Path(ass_path).exists(): _api.upload_file( path_or_fileobj=ass_path, path_in_repo="processed/last_subtitles.ass", repo_id=_dataset, repo_type="dataset", token=os.getenv("HF_TOKEN", "") ) print(f"[PIPELINE] Saved video + ASS to dataset") else: print(f"[PIPELINE] Saved video (no ASS file found at {ass_path})") except Exception as _e: print(f"[PIPELINE] Could not save to dataset: {_e}") upload_result = await asyncio.get_event_loop().run_in_executor( None,self._upload_youtube_sync,final_path,seo ) _t1 = time.time() result["upload"] = upload_result # CHECK if upload actually succeeded (it returns a dict,not an exception) if upload_result.get("status") != "success": upload_error = upload_result.get("error","Upload failed (unknown error)") # SPECIAL HANDLING for quota exceeded: don't fail the pipeline, # just wait for quota reset and retry if upload_result.get("status") == "quota_exceeded": print(f"[PIPELINE] YouTube quota exceeded - pausing uploads") print(f"[PIPELINE] Video saved locally, will retry after quota reset") # Save the video for later upload try: self._save_failed_upload(final_path,seo) except Exception: pass # Return special status so agent_loop knows to wait result["status"] = "quota_exceeded" result["quota_exceeded"] = True result["steps_completed"].append("upload_skipped_quota") return result self.pipeline_analytics.record_error(video_id,"upload",f"YouTube upload failed: {upload_error}") raise Exception(f"YouTube upload failed: {upload_error}") self.pipeline_analytics.record_stage_timing(video_id,"upload",_t1 - _t0) result["steps_completed"].append("upload") # Mark as processed ONLY after confirmed successful upload self.state.mark_video_processed(video_id,title,seo.get("title",translated["title"])) # Also clear from failed list if it was previously failed self.state.clear_failed_video(video_id) result["status"] = "success" # Record overall pipeline success _total_duration = time.time() - _pipeline_start self.pipeline_analytics.record_success( video_id,_total_duration, metadata={ "segments_count": len(segments), "tts_provider": _tts_provider, "video_duration": video_duration, } ) print(f"\n[PIPELINE] SUCCESS: {title} -> {seo.get('title','N/A')}\n") except Exception as e: print(f"\n[PIPELINE] FAILED: {e}\n") result["error"] = str(e) # Determine the failed stage from steps_completed _failed_stage = "download" # default _completed_stages = result.get("steps_completed",[]) _all_stages = ["download","extract_audio","transcribe","translate","validate","tts","compose","upload"] for i,stage in enumerate(_all_stages): if stage not in _completed_stages: _failed_stage = stage break _total_duration = time.time() - _pipeline_start self.pipeline_analytics.record_failure(video_id,_total_duration,_failed_stage,str(e)) self.state.mark_video_failed(video_id,str(e),title=title) # OPTIMIZATION: Skip brain.decide_on_error() - its return value was never used # and it wastes an LLM API call on every failure. The pipeline already has # proper retry logic via state.mark_video_failed() and orchestrator. # if self.brain.is_ready(): # decision = self.brain.decide_on_error(str(e)) # result["brain_decision"] = decision result["brain_decision"] = "skip_optimization" # No API call wasted finally: self._cleanup(work_dir) self._processing = False return result # ================================================================ # Step 1: Download (CF Worker ONLY - no direct YouTube access from HF Space) # IMPORTANT: We do NOT connect to YouTube,Invidious,or Piped directly. # All downloads go through the Cloudflare Worker proxy to avoid HF abuse detection. # ================================================================ async def _innertube_direct(self, video_id: str, output_path: str) -> Optional[str]: """Try Innertube API directly from this Space using IP direct + SNI. If this works, the CDN URL will be bound to THIS Space's IP, allowing direct download without 403. """ import ssl as _ssl import socket as _socket # Resolve youtube.com IP (same range as googleapis.com which works) yt_ips = [] try: yt_ips = _socket.getaddrinfo("www.youtube.com", 443, _socket.AF_INET, _socket.SOCK_STREAM) yt_ips = list(set(ip[4][0] for ip in yt_ips)) except: yt_ips = ["142.250.197.78", "142.250.198.46", "142.250.71.174"] print(f"[INNERTUBE-DIRECT] Trying with IPs: {yt_ips[:3]}") ssl_ctx = _ssl.create_default_context() ssl_ctx.check_hostname = False ssl_ctx.verify_mode = _ssl.CERT_NONE clients = [ ("IOS", "20.10.38", "com.google.android.youtube/20.10.38 (Linux; U; Android 15; US)"), ("ANDROID", "20.10.38", "com.google.android.youtube/20.10.38 (Linux; U; Android 15; US)"), ("WEB", "2.20240726.00.00", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"), ] for yt_ip in yt_ips[:3]: for client_name, client_ver, ua in clients: body = json.dumps({ "context": {"client": {"clientName": client_name, "clientVersion": client_ver, "hl": "en", "gl": "US"}}, "videoId": video_id, "racyCheckOk": True, "contentCheckOk": True, }) innertube_url = f"https://{yt_ip}/youtubei/v1/player?key=AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8" try: async with httpx.AsyncClient(timeout=15, verify=ssl_ctx) as client: resp = await client.post( innertube_url, headers={ "Content-Type": "application/json", "User-Agent": ua, "Host": "www.youtube.com", }, data=body, extensions={"sni_hostname": "www.youtube.com"} ) data = resp.json() sd = data.get("streamingData", {}) formats = sd.get("formats", []) status = data.get("playabilityStatus", {}).get("status", "?") if formats: print(f"[INNERTUBE-DIRECT] ✓ {client_name} via {yt_ip}: {len(formats)} formats!") best = max(formats, key=lambda f: f.get("height", 0)) cdn_url = best.get("url", "") if cdn_url: # Download from CDN (URL is bound to Space's IP) print(f"[INNERTUBE-DIRECT] Downloading from CDN...") async with httpx.AsyncClient(timeout=180, verify=ssl_ctx, follow_redirects=True) as dl: video_resp = await dl.get(cdn_url, headers={"User-Agent": ua}) if video_resp.status_code == 200 and len(video_resp.content) > 10000: with open(output_path, "wb") as f: f.write(video_resp.content) print(f"[INNERTUBE-DIRECT] ✓ Video downloaded: {len(video_resp.content)} bytes") return output_path else: print(f"[INNERTUBE-DIRECT] CDN download failed: HTTP {video_resp.status_code}") elif status == "LOGIN_REQUIRED": print(f"[INNERTUBE-DIRECT] {client_name} via {yt_ip}: LOGIN_REQUIRED (bot detection)") else: print(f"[INNERTUBE-DIRECT] {client_name} via {yt_ip}: {status}") except Exception as e: print(f"[INNERTUBE-DIRECT] {client_name} via {yt_ip}: {type(e).__name__}: {e}") print("[INNERTUBE-DIRECT] All attempts failed") return None async def _download_async(self,url: str,work_dir: Path,video_id: str = "") -> Optional[str]: """Download video. Tries Innertube direct first, then CF Worker. SAFETY: This Space does NOT make direct connections to YouTube,Invidious, Piped,or any YouTube proxy. All downloads are proxied through Cloudflare Workers or third-party APIs (VideoDL,RapidAPI) that handle YouTube access on their own servers. This prevents HF Spaces abuse detection. Priority order: 1. Cloudflare Worker proxy (handles Innertube,HTML scraping,Invidious on CF's edge) 2. VideoDL API (SaaS,downloads on their servers) 3. RapidAPI YouTube Downloader (downloads on their servers) """ if not video_id: vid_match = re.search(r'(?:shorts/|v=|youtu\.be/)([a-zA-Z0-9_-]{11})',url) if vid_match: video_id = vid_match.group(1) if not video_id: return None output_path = str(work_dir / "original.mp4") # Method 0: Innertube direct (try first - if works, CDN URL bound to Space's IP) print(f"[DOWNLOAD] Method 0/5: Innertube direct for {video_id}...") try: result = await asyncio.wait_for( self._innertube_direct(video_id, output_path), timeout=60 ) if result: return result print("[DOWNLOAD] Innertube direct failed, trying CF Worker...") except asyncio.TimeoutError: print("[DOWNLOAD] Innertube direct timed out (60s)") except Exception as e: print(f"[DOWNLOAD] Innertube direct error: {e}") # Method 1: Cloudflare Worker proxy custom_proxy = os.getenv("YTDLP_PROXY","") if custom_proxy: print(f"[DOWNLOAD] Method 1/3: Cloudflare Worker for {video_id}...") try: result = await asyncio.wait_for( self._download_cloudflare_worker(video_id,output_path,url), timeout=120 # Give CF Worker enough time for fallbacks ) if result: # Verify the file has audio has_audio = self._check_audio_track(result) if has_audio: return result else: print("[DOWNLOAD] CF Worker download has no audio,trying next method...") else: print("[DOWNLOAD] CF Worker returned no result") except asyncio.TimeoutError: print("[DOWNLOAD] Cloudflare Worker timed out (120s)") except Exception as e: print(f"[DOWNLOAD] Cloudflare Worker error: {e}") else: print("[DOWNLOAD] WARNING: YTDLP_PROXY not configured! CF Worker is the primary download method.") # Method 2: VideoDL API (SaaS - downloads on their servers,not from HF) videodl_key = os.getenv("VIDEODL_API_KEY","") if videodl_key: print(f"[DOWNLOAD] Method 2/3: VideoDL API for {video_id}...") try: result = await asyncio.wait_for( self._download_videodl_api(url,video_id,output_path,videodl_key), timeout=90 ) if result: return result except asyncio.TimeoutError: print("[DOWNLOAD] VideoDL API timed out (90s)") except Exception as e: print(f"[DOWNLOAD] VideoDL API error: {e}") # Method 3: RapidAPI YouTube Downloader (downloads on their servers) rapidapi_key = os.getenv("RAPIDAPI_KEY","") if rapidapi_key: print(f"[DOWNLOAD] Method 3/3: RapidAPI for {video_id}...") try: result = await asyncio.wait_for( self._download_rapidapi(video_id,output_path,rapidapi_key), timeout=60 ) if result: return result except asyncio.TimeoutError: print("[DOWNLOAD] RapidAPI timed out (60s)") except Exception as e: print(f"[DOWNLOAD] RapidAPI error: {e}") print("[DOWNLOAD] All download methods failed.") return None async def _download_videodl_api(self,url: str,video_id: str,output_path: str,api_key: str) -> Optional[str]: """Download video using video-download-api.com (SaaS). This is a SaaS that handles YouTube downloads from THEIR servers, completely bypassing YouTube's datacenter IP blocking. CONFIRMED WORKING: Tested with Her Story Shorts (1080x1920,720p MP4). API Endpoints (CORRECT as of 2025-06): - Submit: GET https://p.savenow.to/ajax/download.php?url=...&api=...&format=720 - Poll: GET https://p.savenow.to/ajax/progress.php?id=JOB_ID Flow: 1. Submit download job -> get job ID 2. Poll for progress (3s intervals,max 90s) 3. When success=1,download from the returned CDN URL """ VIDEODL_BASE = "https://p.savenow.to" # Build the YouTube URL (prefer /shorts/ format for Shorts) if "/shorts/" not in url and "/watch?v=" not in url: download_url = f"{_YT_SHORTS}{video_id}" else: download_url = url async def _submit_job(dl_url: str,fmt: str = "720") -> Optional[str]: """Submit a download job and return the job ID.""" resp = await client.get( f"{VIDEODL_BASE}/ajax/download.php", params={ "url": dl_url, "format": fmt, "api": api_key, # NOTE: param is "api" NOT "apikey" }, ) if resp.status_code != 200: print(f"[VIDEODL] Submit failed: HTTP {resp.status_code}") return None data = resp.json() if not data.get("success"): print(f"[VIDEODL] Submit error: {data.get('text','unknown')}") return None job_id = data.get("id","") if not job_id: print("[VIDEODL] No job ID returned") return None return job_id async def _poll_job(job_id: str,max_wait: int = 90) -> Optional[str]: """Poll a job until ready,return download URL or None.""" max_polls = max_wait // 3 for i in range(max_polls): await asyncio.sleep(3) poll_resp = await client.get( f"{VIDEODL_BASE}/ajax/progress.php", params={"id": job_id}, ) if poll_resp.status_code != 200: print(f"[VIDEODL] Poll HTTP {poll_resp.status_code},retrying...") continue poll_data = poll_resp.json() progress = poll_data.get("progress",0) status_text = poll_data.get("text","") success = poll_data.get("success",0) dl_url = poll_data.get("download_url","") if i % 5 == 0 or success == 1: print(f"[VIDEODL] Progress: {progress}/1000 - {status_text}") if success == 1 and dl_url: result_title = poll_data.get("title","Unknown") video_duration = poll_data.get("video_duration","?") print(f"[VIDEODL] Download ready: {result_title[:50]} ({video_duration}s)") return dl_url if progress == 1000 and not dl_url: print(f"[VIDEODL] Server-side failure: {status_text}") return None print(f"[VIDEODL] Timed out after {max_wait}s waiting for download") return None try: async with httpx.AsyncClient(timeout=30,follow_redirects=True) as client: # Step 1: Submit download job (720p for best quality) print(f"[VIDEODL] Submitting download job for {video_id}...") job_id = await _submit_job(download_url,"720") if not job_id: # Try with /watch?v= URL format as fallback if "/shorts/" in download_url: watch_url = f"{_YT_WATCH}{video_id}" print(f"[VIDEODL] Retrying with /watch?v= URL format...") job_id = await _submit_job(watch_url,"720") if not job_id: return None print(f"[VIDEODL] Job submitted: {job_id}") # Step 2: Poll for progress dl_url = await _poll_job(job_id,max_wait=90) if not dl_url: # Try with 360p format as fallback print(f"[VIDEODL] 720p failed,trying 360p...") job_id_360 = await _submit_job(download_url,"360") if job_id_360: dl_url = await _poll_job(job_id_360,max_wait=90) if not dl_url: print("[VIDEODL] All attempts failed") return None # Step 3: Download the actual file from CDN print(f"[VIDEODL] Downloading from CDN...") dl_result = await self._download_from_url(dl_url,output_path) if dl_result: size_mb = Path(output_path).stat().st_size / 1024 / 1024 print(f"[VIDEODL] SUCCESS! Downloaded {size_mb:.1f} MB") return dl_result else: print("[VIDEODL] CDN download failed") return None except httpx.TimeoutException: print("[VIDEODL] Request timeout") except Exception as e: print(f"[VIDEODL] Error: {e}") return None async def _download_rapidapi(self,video_id: str,output_path: str,api_key: str) -> Optional[str]: """Download video using RapidAPI YouTube Downloader. Works from data center IPs because RapidAPI's servers download from YouTube on their end (residential-like IPs),then provide CDN download URLs. """ try: async with httpx.AsyncClient(timeout=30,follow_redirects=True) as client: resp = await client.get( f"https://youtube-video-and-shorts-downloader.p.rapidapi.com/download.php?id={video_id}", headers={ "X-RapidAPI-Key": api_key, "X-RapidAPI-Host": "youtube-video-and-shorts-downloader.p.rapidapi.com", }, ) if resp.status_code != 200: print(f"[RAPIDAPI] HTTP {resp.status_code}") return None data = resp.json() if data.get("message","").find("exceeded") >= 0 or data.get("message","").find("quota") >= 0: print(f"[RAPIDAPI] Quota exceeded: {data.get('message','')[:100]}") return None if data.get("status") != "ok": print(f"[RAPIDAPI] Status: {data.get('status','unknown')}") return None results = data.get("results",[]) if not results: print("[RAPIDAPI] No download formats available") return None title = data.get("title","") print(f"[RAPIDAPI] Got {len(results)} formats for: {title[:50]}") # Find best combined video+audio format best_combined = None best_audio_only = None for r in results: quality = r.get("quality","") mime = r.get("mime","") has_audio = r.get("has_audio",False) url = r.get("url","") if not url: continue if has_audio and mime.startswith("video"): if best_combined is None: best_combined = r elif quality == "720p" and best_combined.get("quality") != "720p": best_combined = r elif quality == "360p" and best_combined.get("quality") not in ("720p","360p"): best_combined = r if mime.startswith("audio"): if best_audio_only is None: best_audio_only = r if best_combined: quality = best_combined.get("quality","?") print(f"[RAPIDAPI] Downloading combined {quality}...") result = await self._download_from_url(best_combined["url"],output_path) if result: print(f"[RAPIDAPI] SUCCESS (combined {quality})") return result # Try video-only + audio merge best_video_only = None for r in results: if r.get("mime","").startswith("video") and r.get("url") and not r.get("has_audio",False): if best_video_only is None: best_video_only = r elif r.get("quality") == "720p" and best_video_only.get("quality") != "720p": best_video_only = r if best_video_only and best_audio_only: vid_quality = best_video_only.get("quality","?") print(f"[RAPIDAPI] Downloading video {vid_quality} + audio (needs merge)...") vid_path = str(Path(output_path).parent / "video_only.mp4") aud_path = str(Path(output_path).parent / "audio_only.mp4") vid_result = await self._download_from_url(best_video_only["url"],vid_path) aud_result = await self._download_from_url(best_audio_only["url"],aud_path) if vid_result and aud_result: merge_result = await asyncio.get_event_loop().run_in_executor( None,self._merge_video_audio_sync,vid_path,aud_path,output_path ) if merge_result: print(f"[RAPIDAPI] SUCCESS (merged {vid_quality})") return merge_result # Last resort: try any available format for r in results: if r.get("url") and r.get("has_audio") and r.get("mime","").startswith("video"): quality = r.get("quality","?") print(f"[RAPIDAPI] Last resort: downloading {quality}...") result = await self._download_from_url(r["url"],output_path) if result: print(f"[RAPIDAPI] SUCCESS ({quality})") return result print("[RAPIDAPI] No downloadable format found") return None except httpx.TimeoutException: print("[RAPIDAPI] Request timeout") except Exception as e: print(f"[RAPIDAPI] Error: {e}") return None def _build_cookie_header(self) -> str: """Build a Cookie header string from the cookies.txt file. Includes ALL google.com and youtube.com cookies. The SAPISID cookie is typically under .google.com,not .youtube.com. """ cookies_path = self._get_cookies_path() if not cookies_path: return "" try: cookie_text = Path(cookies_path).read_text() cookie_pairs = [] for line in cookie_text.split('\n'): line = line.strip() if not line or line.startswith('#'): continue parts = line.split('\t') if len(parts) >= 7: domain = parts[0].lower() name = parts[5] value = parts[6] if 'google.com' in domain or 'youtube.com' in domain: cookie_pairs.append(f"{name}={value}") if cookie_pairs: print(f"[COOKIES] Found {len(cookie_pairs)} Google/YouTube cookies") return '; '.join(cookie_pairs) except Exception as e: print(f"[INNERTUBE] Could not read cookies: {e}") return "" # [REMOVED: dead function _try_innertube_client - was hitting yt directly] # [REMOVED: dead function _innertube_download_result - was hitting yt directly] async def _download_cloudflare_worker(self,video_id: str,output_path: str,original_url: str = "") -> Optional[str]: """Download video via youtube-dl-proxy v6 /proxy/{id} endpoint. This endpoint obtains the CDN URL AND downloads the video from the SAME CF edge IP,avoiding IP-binding issues. """ proxy_url = os.getenv("YTDLP_PROXY","").rstrip("/") proxy_secret = os.getenv("YTDLP_PROXY_SECRET","") if not proxy_url: print("[CF-WORKER] YTDLP_PROXY not configured") return None # FIX (2026-07-04): HF Spaces can't resolve *.workers.dev DNS. # Use custom SSL context with server_hostname to bypass DNS + fix SNI. import urllib.parse as _urlparse import ssl as _ssl parsed = _urlparse.urlparse(proxy_url) proxy_host = parsed.hostname # CF Workers are on Cloudflare IPs CF_WORKER_IPS = ["104.21.68.26", "172.67.185.120"] # Build URL with IP but keep original hostname for SNI proxy_url_by_ip = proxy_url.replace(proxy_host, CF_WORKER_IPS[0]) # Create SSL context that uses the original hostname for SNI _ssl_context = _ssl.create_default_context() _ssl_context.check_hostname = False _ssl_context.verify_mode = _ssl.CERT_NONE # Set the SNI hostname _ssl_context.server_hostname = proxy_host headers = {} if proxy_secret: headers["Authorization"] = f"Bearer {proxy_secret}" # Add Host header for SNI when using direct IP headers["Host"] = proxy_host # Load YouTube cookies cookie_header = "" try: cookies_paths = [ Path(os.getenv("BROWSER_DATA_DIR","/app/browser_data")) / "cookies.txt", Path("/tmp/browser_data/cookies.txt"), ] for cookies_path in cookies_paths: if cookies_path.exists(): cookie_text = cookies_path.read_text() cookie_pairs = [] for line in cookie_text.split('\n'): line = line.strip() if not line or line.startswith('#'): continue parts = line.split('\t') if len(parts) >= 7: domain = parts[0].lower() name = parts[5] value = parts[6] if 'google.com' in domain or 'youtube.com' in domain: cookie_pairs.append(f"{name}={value}") if cookie_pairs: cookie_header = '; '.join(cookie_pairs) # CRITICAL: youtube-dl-proxy v6 checks if cookie header # contains "youtube.com" or "google.com" to decide # whether to use cookies. Add domain marker so worker # detects cookies and uses Innertube with auth (not Invidious). cookie_header += '; domain=youtube.com' headers["X-Cookie"] = cookie_header print(f"[CF-WORKER] Loaded {len(cookie_pairs)} cookies (header: {len(cookie_header)} chars)") break except Exception as e: print(f"[CF-WORKER] Cookie error: {e}") # Use /proxy/{video_id} endpoint - downloads video from same CF edge IP proxy_endpoint = f"{proxy_url_by_ip}/proxy/{video_id}" print(f"[CF-WORKER] Downloading via /proxy/{video_id}...") print(f"[CF-WORKER] Headers: {list(headers.keys())}") print(f"[CF-WORKER] Has X-Cookie: {'X-Cookie' in headers}") if 'X-Cookie' in headers: print(f"[CF-WORKER] X-Cookie length: {len(headers['X-Cookie'])}") print(f"[CF-WORKER] X-Cookie has SAPISID: {'SAPISID' in headers['X-Cookie']}") try: # Use httpx directly (not _download_from_url) to ensure headers are passed async with httpx.AsyncClient(timeout=180,follow_redirects=True,verify=_ssl_context) as client: resp = await client.get(proxy_endpoint,headers=headers,extensions={"sni_hostname": proxy_host}) print(f"[CF-WORKER] Response: HTTP {resp.status_code},size={len(resp.content)}") if resp.status_code == 200 and len(resp.content) > 10000: with open(output_path,'wb') as f: f.write(resp.content) file_size = os.path.getsize(output_path) print(f"[CF-WORKER] SUCCESS via /proxy/ ({file_size} bytes)") return output_path else: print(f"[CF-WORKER] /proxy/ failed: HTTP {resp.status_code},size={len(resp.content)}") print(f" Content: {resp.text[:300]}") except Exception as e: print(f"[CF-WORKER] /proxy/ error: {type(e).__name__}: {e}") print("[CF-WORKER] All strategies failed") # FALLBACK: Use /download/{id} endpoint to get CDN URL, then download directly # The /proxy/ endpoint fails when the worker can't reach googlevideo.com, # but /download/ just returns the CDN URL which we can fetch ourselves print("[CF-WORKER] Trying /download/ endpoint as fallback...") download_endpoint = f"{proxy_url_by_ip}/download/{video_id}" try: async with httpx.AsyncClient(timeout=60, follow_redirects=True,verify=_ssl_context) as client: resp = await client.get(download_endpoint, headers=headers,extensions={"sni_hostname": proxy_host}) if resp.status_code == 200: import json as _json info = _json.loads(resp.text) if info.get("status") == "success" and info.get("url"): cdn_url = info["url"] audio_url = info.get("audioUrl", "") title = info.get("title", "") print(f"[CF-WORKER] Got CDN URL from /download/ (title: {title[:40]})") print(f"[CF-WORKER] Downloading video from CDN...") # Download video stream video_path = output_path.replace(".mp4", "_video.mp4") async with httpx.AsyncClient(timeout=180, follow_redirects=True) as dl_client: video_resp = await dl_client.get(cdn_url, headers={ "User-Agent": "com.google.android.youtube/20.10.38 (Linux; U; Android 15; US)" }) if video_resp.status_code == 200 and len(video_resp.content) > 10000: with open(video_path, "wb") as f: f.write(video_resp.content) print(f"[CF-WORKER] Video downloaded: {len(video_resp.content)} bytes") else: print(f"[CF-WORKER] Video download failed: HTTP {video_resp.status_code}") return None # Download audio stream if available (separate from video in YouTube) audio_path = output_path.replace(".mp4", "_audio.mp4") has_audio = False if audio_url: print(f"[CF-WORKER] Downloading audio from CDN...") async with httpx.AsyncClient(timeout=120, follow_redirects=True) as dl_client: audio_resp = await dl_client.get(audio_url, headers={ "User-Agent": "com.google.android.youtube/20.10.38 (Linux; U; Android 15; US)" }) if audio_resp.status_code == 200 and len(audio_resp.content) > 1000: with open(audio_path, "wb") as f: f.write(audio_resp.content) has_audio = True print(f"[CF-WORKER] Audio downloaded: {len(audio_resp.content)} bytes") # If we have both video and audio, merge with ffmpeg # If only video (already has audio), just rename if has_audio: print(f"[CF-WORKER] Merging video + audio with ffmpeg...") import subprocess as _sp merge_cmd = [ "ffmpeg", "-y", "-v", "error", "-i", video_path, "-i", audio_path, "-c:v", "copy", "-c:a", "aac", "-b:a", "128k", "-shortest", output_path ] r = _sp.run(merge_cmd, capture_output=True, text=True, timeout=60) if r.returncode == 0 and os.path.exists(output_path): print(f"[CF-WORKER] Merged video+audio: {os.path.getsize(output_path)} bytes") # Clean up temp files try: os.remove(video_path) os.remove(audio_path) except Exception: pass return output_path else: print(f"[CF-WORKER] Merge failed: {r.stderr[-200:]}") elif os.path.exists(video_path): # Video only - check if it has audio import subprocess as _sp probe = _sp.run([ "ffprobe", "-v", "error", "-show_entries", "stream=codec_type", "-of", "csv=p=0", video_path ], capture_output=True, text=True, timeout=15) if "audio" in probe.stdout: # Has audio, just rename os.rename(video_path, output_path) print(f"[CF-WORKER] Video with audio: {os.path.getsize(output_path)} bytes") return output_path else: print(f"[CF-WORKER] Video has no audio track!") else: print(f"[CF-WORKER] /download/ failed: {info.get('error', 'unknown')}") else: print(f"[CF-WORKER] /download/ HTTP {resp.status_code}") except Exception as e: print(f"[CF-WORKER] /download/ error: {type(e).__name__}: {e}") return None async def _download_from_url(self,url: str,output_path: str,extra_headers: dict = None) -> Optional[str]: """Download a file from a URL to the output path. Uses streaming for large files.""" try: headers = {} if extra_headers: headers.update(extra_headers) async with httpx.AsyncClient(timeout=120,follow_redirects=True) as client: async with client.stream("GET",url,headers=headers) as resp: if resp.status_code not in (200,206): print(f"[DOWNLOAD] Bad response: HTTP {resp.status_code}") return None total_size = 0 with open(output_path,"wb") as f: async for chunk in resp.aiter_bytes(chunk_size=65536): f.write(chunk) total_size += len(chunk) if total_size > 10000: size_mb = total_size / 1024 / 1024 print(f"[DOWNLOAD] Downloaded: {size_mb:.1f} MB") return output_path else: print(f"[DOWNLOAD] File too small: {total_size} bytes") try: Path(output_path).unlink() except: pass return None except httpx.TimeoutException: print("[DOWNLOAD] Timeout downloading from URL") except Exception as e: print(f"[DOWNLOAD] Error downloading from URL: {e}") return None def _check_audio_track(self,video_path: str) -> bool: """Check if a video file has an audio track.""" try: proc = subprocess.run([ "ffprobe","-v","quiet","-select_streams","a", "-show_entries","stream=codec_type","-of","csv=p=0", video_path, ],capture_output=True,text=True,timeout=10) return "audio" in proc.stdout.lower() except Exception: return False def _extract_audio_sync(self,video_path: str,work_dir: Path) -> Optional[str]: """Extract audio from video using FFmpeg. If the video has no audio track (common with ANDROID InnerTube downloads), try to re-download using browser or generate silence as fallback. """ audio_path = str(work_dir / "audio.wav") # First check if video has an audio track has_audio = self._check_audio_track(video_path) if not has_audio: print("[AUDIO] Video has NO audio track! Trying browser re-download...") # Try browser download which gets both video and audio if self.browser and hasattr(self.browser,'download_video'): try: import asyncio loop = asyncio.new_event_loop() browser_result = loop.run_until_complete( self.browser.download_video( Path(video_path).stem.split('_')[0] if '_' in Path(video_path).stem else "", str(work_dir / "original_with_audio.mp4") ) ) loop.close() if browser_result: video_path = browser_result print(f"[AUDIO] Re-downloaded with audio: {video_path}") except Exception as e: print(f"[AUDIO] Browser re-download failed: {e}") # If still no audio,generate silence based on video duration has_audio = self._check_audio_track(video_path) if not has_audio: print("[AUDIO] Still no audio,generating silent track from video duration...") try: # Get video duration proc = subprocess.run([ "ffprobe","-v","quiet","-show_entries","format=duration", "-of","default=noprint_wrappers=1:nokey=1",video_path, ],capture_output=True,text=True,timeout=10) duration = float(proc.stdout.strip()) # Generate silence proc = subprocess.run([ "ffmpeg","-f","lavfi","-i",f"anullsrc=r=16000:cl=mono", "-t",str(duration),"-acodec","pcm_s16le","-y",audio_path, ],capture_output=True,timeout=30) if proc.returncode == 0: print(f"[AUDIO] Generated {duration:.1f}s silent audio: {audio_path}") return audio_path except Exception as e: print(f"[AUDIO] Silent generation error: {e}") return None try: proc = subprocess.run([ "ffmpeg","-i",video_path, "-vn","-acodec","pcm_s16le","-ar","16000","-ac","1","-y", audio_path, ],capture_output=True,timeout=120) if proc.returncode == 0: print(f"[AUDIO] Extracted: {audio_path}") return audio_path print(f"[AUDIO] Error: {proc.stderr.decode()[-200:]}") return None except Exception as e: print(f"[AUDIO] Error: {e}") return None # ================================================================ # Step 3: Transcribe WITH TIMESTAMPS (REDESIGNED) # ================================================================ def _transcribe_groq(self,audio_path: str) -> Optional[list]: """Transcribe audio using Groq Whisper API with segment timestamps. REDESIGNED: Returns list of dicts with timestamps instead of plain text. Each segment: {"text": "English text","start": 0.0,"end": 3.5} Also stores self.transcript_text for backward compatibility (plain text version). Fallback to local Whisper if Groq fails. """ try: with open(audio_path,"rb") as f: response = self.groq_client.audio.transcriptions.create( model="whisper-large-v3-turbo", file=f, language="en", response_format="verbose_json", timestamp_granularities=["segment"], ) raw_segments = getattr(response,"segments",[]) if raw_segments: segments = [] for seg in raw_segments: text = seg.get("text","").strip() if text: # Skip empty segments segments.append({ "text": text, "start": round(seg.get("start",0.0),3), "end": round(seg.get("end",0.0),3), }) # Backward compat: store plain text self.transcript_text = " ".join(s["text"] for s in segments) else: # Fallback: single segment from full text full_text = getattr(response,"text","").strip() if full_text: segments = [{"text": full_text,"start": 0.0,"end": 60.0}] self.transcript_text = full_text else: self.transcript_text = "" segments = [] print(f"[TRANSCRIBE] Groq Whisper: {len(segments)} segments,{len(self.transcript_text)} chars") for i,s in enumerate(segments): print(f" [{i}] {s['start']:.1f}-{s['end']:.1f}s: {s['text'][:60]}...") return segments if segments else None except Exception as e: print(f"[TRANSCRIBE] Groq error: {e}") # Fallback: local Whisper return self._transcribe_local(audio_path) def _transcribe_local(self,audio_path: str) -> Optional[list]: """Synchronous local Whisper transcription (fallback). Returns segments with timestamps.""" try: from faster_whisper import WhisperModel model = WhisperModel("large-v3-turbo",device="cpu",compute_type="int8") segments_gen,info = model.transcribe(audio_path,language="en") segments = [] for seg in segments_gen: text = seg.text.strip() if text: segments.append({ "text": text, "start": round(seg.start,3), "end": round(seg.end,3), }) self.transcript_text = " ".join(s["text"] for s in segments) print(f"[TRANSCRIBE] Local Whisper: {len(segments)} segments,{len(self.transcript_text)} chars") return segments if segments else None except Exception as e: print(f"[TRANSCRIBE] Local error: {e}") self.transcript_text = "" return None # ================================================================ # Step 4: Translate PER SEGMENT (REDESIGNED) # ================================================================ def _translate_sync(self,segments: list,original_title: str,suggestions: str = "") -> Optional[dict]: """Translate English segments to Spanish,preserving original timestamps. PRIORITY: Groq API (FAST,<5 seconds per batch) FALLBACK: Local AI model (Nanbeige,slow on CPU but unlimited) Takes segments list with "text","start","end" keys. Translates each segment's text to Spanish. For efficiency,batches segments into groups of up to 5. Adds "text_es" to each segment. Returns dict with: - "segments": updated list with "text_es" on each - "title": translated title - "text": full Spanish text for backward compat """ if not segments: return None # Batch segments into groups of up to 5 for efficient translation batch_size = 5 batches = [] for i in range(0,len(segments),batch_size): batch = segments[i:i + batch_size] batches.append(batch) all_translations = {} # index -> translated text # ============================================= # Strategy 1: Groq API - PRIMARY (FAST,<5 seconds per batch) # ============================================= groq_key = os.getenv("GROQ_API_KEY","") if groq_key: print("[TRANSLATE] Using Groq API (FAST,primary)") for batch_idx,batch in enumerate(batches): segment_lines = [] for j,seg in enumerate(batch): global_idx = batch_idx * batch_size + j segment_lines.append(f"[{global_idx}] {seg['text']}") segments_text = "\n".join(segment_lines) improvement = "" if suggestions: improvement = f"\n\nAdditional guidance: {suggestions}" prompt = f"""You are a professional English-to-Spanish translator specializing in YouTube Shorts about women's history and dramatic storytelling. Translate each numbered segment to natural,engaging Latin American Spanish. Maintain the SAME approximate length and pacing as the original - do not expand or compress the meaning. Use conversational Spanish with a dramatic,storytelling tone. CRITICAL RULES: - Keep each translation roughly the same length as the original segment - Preserve the dramatic tone and emotional impact - Add natural pauses with commas where a speaker would breathe - Return ONLY the translations,one per line,with the same [index] format - Do NOT add any extra text,explanations,or formatting Segments to translate: {segments_text}{improvement}""" # Retry logic with backoff for rate limits + model fallback max_retries = 3 models = ["llama-3.3-70b-versatile","llama-3.1-8b-instant"] batch_success = False for model in models: for attempt in range(max_retries): try: response = self.groq_client.chat.completions.create( model=model, messages=[ {"role": "system","content": "You are a professional EN→ES translator for YouTube Shorts. Return ONLY the translated segments with the same [index] format,one per line. No extra text."}, {"role": "user","content": prompt} ], temperature=0.3, max_tokens=4096, ) result = response.choices[0].message.content.strip() # Parse the translations from the response for line in result.split('\n'): line = line.strip() if not line: continue match = re.match(r'\[(\d+)\]\s*(.*)',line) if match: idx = int(match.group(1)) translated_text = match.group(2).strip() if translated_text: all_translations[idx] = translated_text # If we didn't get indexed results,try to map line-by-line if not any(k in all_translations for k in range(batch_idx * batch_size,batch_idx * batch_size + len(batch))): lines = [l.strip() for l in result.split('\n') if l.strip()] clean_lines = [] for l in lines: if l.startswith(('Translation','SUBTITLES','Traducción','Here are','---','```')): continue cleaned = re.sub(r'^\[\d+\]\s*','',l) if cleaned: clean_lines.append(cleaned) for j,translated_text in enumerate(clean_lines): global_idx = batch_idx * batch_size + j if global_idx < len(segments): all_translations[global_idx] = translated_text batch_success = True break except Exception as e: error_str = str(e) if '429' in error_str or 'rate_limit' in error_str.lower(): wait_time = 60 * (attempt + 1) print(f"[TRANSLATE] Rate limited on {model} (attempt {attempt+1}/{max_retries}),waiting {wait_time}s...") time.sleep(wait_time) continue elif model != models[-1]: print(f"[TRANSLATE] {model} failed: {error_str[:80]},trying fallback model...") break else: print(f"[TRANSLATE] Batch {batch_idx} failed on all models: {error_str[:80]}") break if batch_success: break if not batch_success: print(f"[TRANSLATE] Batch {batch_idx} failed after all retries") for j,seg in enumerate(batch): global_idx = batch_idx * batch_size + j if global_idx not in all_translations: all_translations[global_idx] = seg["text"] print(f"[TRANSLATE] WARNING: Using original text for segment {global_idx}") # ============================================= # Strategy 2: Local AI model (Nanbeige) - FALLBACK (slow on CPU but unlimited) # ============================================= if len(all_translations) < len(segments) * 0.7: if self.brain and self.brain.is_ready(): print("[TRANSLATE] Using LOCAL Nanbeige model (fallback,slow on CPU)") local_success = True for batch_idx,batch in enumerate(batches): # Skip batches that already have translations from Groq if all((batch_idx * batch_size + j) in all_translations for j in range(len(batch))): continue segment_lines = [] for j,seg in enumerate(batch): global_idx = batch_idx * batch_size + j if global_idx not in all_translations: segment_lines.append(f"[{global_idx}] {seg['text']}") if not segment_lines: continue segments_text = "\n".join(segment_lines) improvement = "" if suggestions: improvement = f"\n\nOrientación adicional: {suggestions}" prompt = f"<|im_start|>system\nEres un traductor profesional de inglés a español para YouTube Shorts de historia dramática. Traduce cada segmento numerado al español latinoamericano natural y dramático. Mantén la misma longitud aproximada. Devuelve SOLO las traducciones con el mismo formato [índice],una por línea. Sin texto extra.<|im_end|>\n<|im_start|>user\nTraduce estos segmentos:\n{segments_text}{improvement}<|im_end|>\n<|im_start|>assistant\n" try: response = self.brain.think(prompt,max_tokens=2048,temperature=0.3) if response: for line in response.split('\n'): line = line.strip() if not line: continue match = re.match(r'\[(\d+)\]\s*(.*)',line) if match: idx = int(match.group(1)) translated_text = match.group(2).strip() if translated_text: all_translations[idx] = translated_text # Fallback: line-by-line mapping if no indexed results batch_indices = [batch_idx * batch_size + j for j in range(len(batch))] if not any(k in all_translations for k in batch_indices): lines = [l.strip() for l in response.split('\n') if l.strip()] clean_lines = [] for l in lines: if l.startswith(('Translation','SUBTITLES','Traducción','Here are','---','```','Los segmentos','Aquí')): continue cleaned = re.sub(r'^\[\d+\]\s*','',l) if cleaned: clean_lines.append(cleaned) for j,translated_text in enumerate(clean_lines): global_idx = batch_idx * batch_size + j if global_idx < len(segments) and global_idx not in all_translations: all_translations[global_idx] = translated_text else: print(f"[TRANSLATE] Local model returned empty for batch {batch_idx}") local_success = False except Exception as e: print(f"[TRANSLATE] Local model error for batch {batch_idx}: {e}") local_success = False if local_success and len(all_translations) >= len(segments) * 0.7: print(f"[TRANSLATE] Local model SUCCESS: {len(all_translations)}/{len(segments)} segments translated") else: print(f"[TRANSLATE] Local model incomplete ({len(all_translations)}/{len(segments)})") # Apply translations to segments # CRITICAL: Always have text_es, even if fallback to English translated_count = 0 for i,seg in enumerate(segments): if i in all_translations and all_translations[i].strip(): seg["text_es"] = all_translations[i] translated_count += 1 else: # Fallback: try Groq one more time with single segment try: import httpx as _httpx groq_key = os.getenv("GROQ_API_KEY", "") if groq_key: resp = _httpx.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {groq_key}", "Content-Type": "application/json"}, json={ "model": "llama-3.1-8b-instant", "messages": [{"role": "user", "content": f"Translate to Spanish, return ONLY the translation: {seg['text']}"}], "max_tokens": 100, "temperature": 0.3 }, timeout=10 ) if resp.status_code == 200: tr = resp.json()["choices"][0]["message"]["content"].strip() if tr: seg["text_es"] = tr translated_count += 1 continue except: pass # Last resort: use English text (better than no subs) seg["text_es"] = seg["text"] print(f"[TRANSLATE] Total translated: {translated_count}/{len(segments)}") # Clean translations for seg in segments: text_es = seg["text_es"] for marker in ["TRANSLATION:","SUBTITLES:","Translation:","Subtitles:"]: if text_es.startswith(marker): text_es = text_es[len(marker):].strip() seg["text_es"] = text_es # Build full Spanish text for backward compat full_spanish = " ".join(seg["text_es"] for seg in segments) # Translate title translated_title = self._translate_title_sync(original_title) return { "segments": segments, "title": translated_title, "text": full_spanish, } def _translate_title_sync(self,title: str) -> str: """Translate just the video title. Uses local model first,Groq fallback. Titles should NOT end with '...' - they should be clean and dramatic. """ # Clean up title - remove "Unknown" or empty if not title or title.strip().lower() in ("unknown","unknown title",""): title = "Historia Increible" # Strategy 1: Local model (Nanbeige) if self.brain and self.brain.is_ready(): prompt = f"<|im_start|>system\nTraduce este titulo de YouTube Short al espanol. Devuelve SOLO el titulo traducido,nada mas. Maximo 55 caracteres. No agregues '...' al final. Hazlo llamativo y dramatico.<|im_end|>\n<|im_start|>user\nTitulo: {title}<|im_end|>\n<|im_start|>assistant\n" try: result = self.brain.think(prompt,max_tokens=100,temperature=0.4) if result: # Clean up the result result = result.strip().strip('"').strip("'") # Remove trailing ... if the model added it while result.endswith('...'): result = result[:-3].rstrip() if len(result) > 60: result = result[:57] if not result or not result.strip(): result = title # Keep original title as fallback print(f"[TRANSLATE] Title (local): {title} -> {result}") return result except Exception as e: print(f"[TRANSLATE] Local title translation error: {e}") # Strategy 2: Groq API models = ["llama-3.3-70b-versatile","llama-3.1-8b-instant"] for model in models: for attempt in range(3): try: response = self.groq_client.chat.completions.create( model=model, messages=[ {"role": "system","content": "Translate this YouTube Short title to Spanish. Return ONLY the translated title,nothing else. Keep it under 55 characters. Do NOT add '...' at the end. Make it catchy and dramatic."}, {"role": "user","content": title} ], temperature=0.4, max_tokens=100, ) result = response.choices[0].message.content.strip().strip('"') # Remove trailing ... if the model added it while result.endswith('...'): result = result[:-3].rstrip() if not result or not result.strip(): break # Try next attempt return result[:60] except Exception as e: if '429' in str(e): time.sleep(60 * (attempt + 1)) continue break # Fallback: just return original title (no ...) result = title.rstrip('.') if not result or not result.strip(): return "Historia Increible" return result # ================================================================ # Step 6: Generate Voice PER SEGMENT (REDESIGNED) # ================================================================ async def _generate_voice_async(self,segments: list,work_dir: Path,video_path: str) -> Optional[dict]: """Generate Spanish voice per-segment using edge-tts (Ximena) as primary provider. CRITICAL FIX: Uses adelay+amix for perfect sync with subtitles. Each segment is positioned at its exact Whisper timestamp. PROVIDER PRIORITY (2026-07-03): 1. edge-tts (es-ES-XimenaNeural @ +0% rate) - high quality, fast, free 2. Supertonic 3 TTS - fallback if edge-tts fails 3. gTTS - last resort """ if not segments: return None voice_path = str(work_dir / "voice_es.mp3") video_duration = self._get_duration_sync(video_path) # ============================================= # Strategy 1: edge-tts Ximena (PRIMARY) # ============================================= try: import edge_tts print("[TTS] Strategy 1: edge-tts Ximena (es-ES-XimenaNeural, +0% rate)...") edge_result = await self._generate_edge_tts_async(segments,work_dir,video_duration) if edge_result: print("[TTS] edge-tts SUCCESS!") return edge_result print("[TTS] edge-tts failed,falling back to Supertonic...") except ImportError: print("[TTS] edge-tts not installed,falling back to Supertonic...") except Exception as e: print(f"[TTS] edge-tts error: {e},falling back to Supertonic...") # ============================================= # Strategy 2: Supertonic 3 TTS (FALLBACK) # ============================================= if SUPERTONIC_AVAILABLE: print("[TTS] Strategy 2: Supertonic 3 TTS (per-segment)...") supertonic_result = await self._generate_supertonic_async(segments,work_dir,video_duration) if supertonic_result: print("[TTS] Supertonic 3 SUCCESS!") return supertonic_result print("[TTS] Supertonic 3 failed,falling back to gTTS...") # ============================================= # Strategy 3: gTTS (LAST RESORT) # ============================================= print("[TTS] Strategy 3: gTTS fallback...") return await self._generate_gtts_async(segments,work_dir,video_duration) async def _generate_edge_tts_async(self,segments: list,work_dir: Path,video_duration: float) -> Optional[dict]: """Generate Spanish voice per-segment using edge-tts Ximena (es-ES-XimenaNeural). edge-tts uses Microsoft Edge's online TTS service - high quality neural voice, free, no API key needed. Ximena is a Colombian Spanish female voice with natural prosody perfect for narration. Speed: +0% rate (natural speed, no distortion). Per-segment generation with TRUNCATION to target_duration ensures perfect sync with video (Spanish TTS often longer than English slot - truncation prevents overlap with next segment). """ import edge_tts VOICE = "es-ES-XimenaNeural" # Colombian Spanish female, natural narration RATE = "+0%" # Natural speed VOLUME = "+0%" # Group chunks into sentence-level segments for better prosody sentence_segments = self._group_segments_into_sentences(segments) if not sentence_segments: return None print(f"[EDGE-TTS] {len(sentence_segments)} sentence-segments to generate") actual_tts_count = 0 for seg_idx,seg in enumerate(sentence_segments): text_es = seg.get("text_es","").strip() if not text_es: seg["audio_path"] = None seg["audio_duration"] = 0 continue text_es = self._clean_text_for_tts(text_es) target_duration = seg["end"] - seg["start"] seg_audio_path = str(work_dir / f"et_seg_{seg_idx}.mp3") try: # Generate with edge-tts communicate = edge_tts.Communicate(text_es,voice=VOICE,rate=RATE,volume=VOLUME) await communicate.save(seg_audio_path) if not Path(seg_audio_path).exists() or Path(seg_audio_path).stat().st_size < 500: print(f"[EDGE-TTS] Segment {seg_idx}: empty output") seg["audio_path"] = None seg["audio_duration"] = 0 continue # Convert to WAV for consistent processing wav_path = str(work_dir / f"et_seg_{seg_idx}.wav") proc = subprocess.run([ "ffmpeg","-y","-v","error", "-i",seg_audio_path, "-ar","44100","-ac","1","-c:a","pcm_s16le", wav_path ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(wav_path).exists(): seg_audio_path = wav_path actual_tts_count += 1 gen_duration = self._get_duration_sync(seg_audio_path) print(f"[EDGE-TTS] Segment {seg_idx}: gen={gen_duration:.2f}s,target={target_duration:.2f}s") # ============================================= # SPEED ADJUSTMENT - only if drift > 15%, capped at 1.3x # ============================================= if gen_duration > 0 and target_duration > 0: ratio = gen_duration / target_duration if abs(ratio - 1.0) > 0.15: tempo = target_duration / gen_duration tempo = max(0.7, min(1.3, tempo)) tempo_path = str(work_dir / f"et_seg_{seg_idx}_tempo.wav") proc = subprocess.run([ "ffmpeg","-y","-v","error", "-i",seg_audio_path, "-filter:a",f"atempo={tempo:.4f}", tempo_path ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(tempo_path).exists(): shutil.move(tempo_path,seg_audio_path) gen_duration = self._get_duration_sync(seg_audio_path) # TRUNCATE to target_duration to prevent overlap with next segment # This is CRITICAL: if Spanish TTS is longer than English slot, # it bleeds into next segment causing dubbing desync if gen_duration > target_duration: trunc_path = str(work_dir / f"et_seg_{seg_idx}_trunc.wav") proc = subprocess.run([ "ffmpeg","-y","-v","error", "-i",seg_audio_path, "-t",str(target_duration), "-c:a","pcm_s16le",trunc_path ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(trunc_path).exists(): shutil.move(trunc_path,seg_audio_path) gen_duration = self._get_duration_sync(seg_audio_path) seg["audio_path"] = seg_audio_path seg["audio_duration"] = min(gen_duration,target_duration) except Exception as e: print(f"[EDGE-TTS] Segment {seg_idx}: error: {e}") seg["audio_path"] = None seg["audio_duration"] = 0 if actual_tts_count == 0: print("[EDGE-TTS] No segments generated") return None # ============================================= # SYNC: adelay + amix for perfect timestamp positioning # ============================================= valid_segments = [(i,seg) for i,seg in enumerate(sentence_segments) if seg.get("audio_path") and Path(seg["audio_path"]).exists()] if not valid_segments: return None voice_path = str(work_dir / "voice_es.mp3") try: input_args = [] filter_parts = [] for input_idx,(idx,seg) in enumerate(valid_segments): seg_wav = str(work_dir / f"et_seg_{idx}_aligned.wav") proc = subprocess.run([ "ffmpeg","-y","-v","error", "-i",seg["audio_path"], "-ar","44100","-ac","1","-c:a","pcm_s16le", seg_wav ],capture_output=True,timeout=15) if not (proc.returncode == 0 and Path(seg_wav).exists()): continue input_args.extend(["-i",seg_wav]) delay_ms = int(seg["start"] * 1000) filter_parts.append(f"[{input_idx}:a]adelay={delay_ms}|{delay_ms}[d{input_idx}]") if not filter_parts: return None mix_inputs = "".join(f"[d{i}]" for i in range(len(filter_parts))) n_inputs = len(filter_parts) filter_complex = ";".join(filter_parts) + f";{mix_inputs}amix=inputs={n_inputs}:duration=longest:dropout_transition=0:normalize=0,apad=whole_dur={video_duration}[a]" cmd = [ "ffmpeg","-y","-v","error", ] + input_args + [ "-filter_complex",filter_complex, "-map","[a]", "-c:a","libmp3lame","-b:a","128k", voice_path ] proc = subprocess.run(cmd,capture_output=True,text=True,timeout=120) if proc.returncode != 0: print(f"[EDGE-TTS] amix failed: {proc.stderr[-500:]}") # Fallback: sequential concat concat_parts = [] current_time = 0.0 for idx,seg in valid_segments: seg_start = seg["start"] gap = seg_start - current_time if gap > 0.01: silence_path = str(work_dir / f"et_gap_{idx}.wav") gap_duration = min(gap,video_duration - current_time) if gap_duration > 0: subprocess.run([ "ffmpeg","-y","-v","error", "-f","lavfi","-i","anullsrc=r=44100:cl=mono", "-t",str(gap_duration), "-c:a","pcm_s16le",silence_path ],capture_output=True,timeout=15) if Path(silence_path).exists(): concat_parts.append(silence_path) current_time += gap_duration seg_wav = str(work_dir / f"et_seg_{idx}_aligned.wav") if Path(seg_wav).exists(): concat_parts.append(seg_wav) current_time += self._get_duration_sync(seg_wav) if concat_parts: list_file = str(work_dir / "et_concat_list.txt") with open(list_file,"w") as f: for part in concat_parts: f.write(f"file '{part}'\n") subprocess.run([ "ffmpeg","-y","-v","error", "-f","concat","-safe","0","-i",list_file, "-af",f"apad=whole_dur={video_duration}", "-c:a","libmp3lame","-b:a","128k", voice_path ],capture_output=True,text=True,timeout=60) if not Path(voice_path).exists(): return None print(f"[EDGE-TTS] Voice generated (amix sync): {Path(voice_path).stat().st_size} bytes") return {"path": voice_path,"segments": segments,"provider": "edge-tts-ximena"} except Exception as e: print(f"[EDGE-TTS] Sync error: {e}") return None def _group_segments_into_sentences(self,segments: list) -> list: """Group 2-word chunks into sentence-level segments for better TTS prosody. edge-tts produces more natural speech when given full sentences vs 2-word chunks. Groups chunks separated by <0.3s gap or after 3 chunks max. FIX (2026-07-03): Reduced max chunks per sentence from 6 to 3. With 6 chunks, the Spanish translation was often longer than the time slot, causing the voice to fall behind the video (dubbing desync). With 3 chunks max, each sentence is shorter and fits better in its slot. """ if not segments: return [] sentences = [] current = [] for seg in segments: if not current: current.append(seg) else: gap = seg["start"] - current[-1]["end"] if gap > 0.3 or len(current) >= 3: sentences.append(current) current = [seg] else: current.append(seg) if current: sentences.append(current) result = [] for sent in sentences: text = " ".join(s.get("text_es","").strip() for s in sent) result.append({ "start": sent[0]["start"], "end": sent[-1]["end"], "text_es": text, }) return result async def _generate_supertonic_async(self,segments: list,work_dir: Path,video_duration: float) -> Optional[dict]: """Generate Spanish voice per-segment using Supertonic 3 TTS with sequential concatenation. Uses sequential concat (NOT amix) to prevent the audio overlap bug where multiple segments play simultaneously over each other. Tries REMOTE Supertonic Space first,then LOCAL fallback. """ # Check if remote or local Supertonic is available use_remote = bool(SUPERTONIC_REMOTE_URLS) model = None if not use_remote: model = _get_supertonic_model() if model is None: print("[SUPERTONIC] Model not available (neither remote nor local)") return None segment_audio_paths = [] actual_tts_count = 0 for seg_idx,seg in enumerate(segments): text_es = seg.get("text_es","").strip() if not text_es: seg["audio_path"] = None seg["audio_duration"] = 0 continue text_es = self._clean_text_for_tts(text_es) target_duration = seg["end"] - seg["start"] seg_audio_path = str(work_dir / f"st_seg_{seg_idx}.wav") generated = False # Try REMOTE Supertonic first if use_remote: try: loop = asyncio.get_event_loop() wav_bytes = await loop.run_in_executor( None, lambda: _supertonic_remote_tts(text_es,SUPERTONIC_VOICE,SUPERTONIC_LANG,1.2) ) if wav_bytes and len(wav_bytes) > 500: Path(seg_audio_path).parent.mkdir(parents=True,exist_ok=True) with open(seg_audio_path,"wb") as f: f.write(wav_bytes) if Path(seg_audio_path).exists() and Path(seg_audio_path).stat().st_size > 500: generated = True actual_tts_count += 1 print(f"[SUPERTONIC] Segment {seg_idx}: REMOTE OK") except Exception as e: print(f"[SUPERTONIC] Segment {seg_idx}: REMOTE error: {e}") # LOCAL Supertonic fallback if not generated and model is not None: try: # Get voice style style = model.get_voice_style(voice_name=SUPERTONIC_VOICE) # Run Supertonic synthesis in executor (it's sync/blocking) loop = asyncio.get_event_loop() wav,duration = await loop.run_in_executor( None, lambda: model.synthesize(text_es,voice_style=style,lang=SUPERTONIC_LANG) ) model.save_audio(wav,seg_audio_path) if Path(seg_audio_path).exists() and Path(seg_audio_path).stat().st_size > 500: generated = True actual_tts_count += 1 print(f"[SUPERTONIC] Segment {seg_idx}: LOCAL OK") else: print(f"[SUPERTONIC] Segment {seg_idx}: output file too small") except Exception as e: print(f"[SUPERTONIC] Segment {seg_idx}: LOCAL error: {e}") if not generated: # Try gTTS for this single segment try: from gtts import gTTS gtts_path = str(work_dir / f"st_seg_{seg_idx}_gtts.mp3") tts = gTTS(text=text_es[:500],lang='es',slow=False) tts.save(gtts_path) if Path(gtts_path).exists() and Path(gtts_path).stat().st_size > 500: # Convert to WAV for consistency proc = subprocess.run([ "ffmpeg","-y","-i",gtts_path, "-ar","44100","-ac","1",seg_audio_path, ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(seg_audio_path).exists(): generated = True actual_tts_count += 1 print(f"[SUPERTONIC] Segment {seg_idx}: gTTS per-segment fallback OK") except Exception as ge: print(f"[SUPERTONIC] Segment {seg_idx}: gTTS fallback failed: {ge}") if not generated: # Generate silence for this segment try: proc = subprocess.run([ "ffmpeg","-y", "-f","lavfi","-i","anullsrc=r=44100:cl=mono", "-t",str(target_duration), seg_audio_path, ],capture_output=True,timeout=15) except Exception: pass if not Path(seg_audio_path).exists(): seg["audio_path"] = None seg["audio_duration"] = 0 continue # Speed adjustment if needed gen_duration = self._get_duration_sync(seg_audio_path) print(f"[SUPERTONIC] Segment {seg_idx}: generated={gen_duration:.1f}s,target={target_duration:.1f}s") # ============================================= # SPEED ADJUSTMENT - ALWAYS fit to segment duration # ============================================= # FIX: Always adjust speed to fit target_duration,not just when ratio > 1.05. # The old 5% tolerance caused audio drift accumulation across segments, # resulting in 1-2 second voice desynchronization. if gen_duration > 0 and target_duration > 0: ratio = gen_duration / target_duration # FIX: Always adjust speed if > 0.5% off (was only when > 1% OVER) # The old code only adjusted when TTS was longer than target, # but Spanish TTS is often SHORTER,causing progressive drift. # With 10 segments at 300ms drift each = 1.5-2s desync. if abs(ratio - 1.0) > 0.005: # Adjust if > 0.5% off (both shorter and longer) tempo = target_duration / gen_duration # Build atempo filter chain (atempo range is 0.5-2.0) # For extreme ratios,chain multiple atempo filters atempo_filters = [] remaining_tempo = tempo while remaining_tempo < 0.5: atempo_filters.append("atempo=0.5") remaining_tempo = remaining_tempo / 0.5 while remaining_tempo > 2.0: atempo_filters.append("atempo=2.0") remaining_tempo = remaining_tempo / 2.0 atempo_filters.append(f"atempo={remaining_tempo:.4f}") atempo_filter = ",".join(atempo_filters) print(f"[SUPERTONIC] Segment {seg_idx}: speed {tempo:.3f}x (gen={gen_duration:.1f}s → target={target_duration:.1f}s)") try: tempo_path = str(work_dir / f"st_seg_{seg_idx}_tempo.wav") proc = subprocess.run([ "ffmpeg","-i",seg_audio_path,"-filter:a",atempo_filter, "-y",tempo_path, ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(tempo_path).exists(): shutil.move(tempo_path,seg_audio_path) gen_duration = self._get_duration_sync(seg_audio_path) print(f"[SUPERTONIC] Segment {seg_idx}: after speed={gen_duration:.3f}s (target={target_duration:.1f}s)") except Exception as e: print(f"[SUPERTONIC] Segment {seg_idx}: speed adjustment error: {e}") # SAFETY: Truncate audio to target duration if still too long # This is the ultimate safety net against overlap # FIX: Lowered tolerance from 50ms to 10ms to reduce drift accumulation if gen_duration > target_duration + 0.01: print(f"[SUPERTONIC] Segment {seg_idx}: TRUNCATING {gen_duration:.3f}s → {target_duration:.3f}s") try: trunc_path = str(work_dir / f"st_seg_{seg_idx}_trunc.wav") proc = subprocess.run([ "ffmpeg","-y","-i",seg_audio_path, "-t",str(target_duration), "-c:a","pcm_s16le",trunc_path, ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(trunc_path).exists(): shutil.move(trunc_path,seg_audio_path) gen_duration = self._get_duration_sync(seg_audio_path) except Exception as e: print(f"[SUPERTONIC] Segment {seg_idx}: truncation error: {e}") seg["audio_path"] = seg_audio_path # FIX: Use min of generated duration and target to prevent drift seg["audio_duration"] = min(gen_duration,target_duration) if actual_tts_count == 0: print("[SUPERTONIC] No actual TTS generated") return None # ============================================= # SEQUENTIAL CONCATENATION (fixes audio overlap bug) # ============================================= # Build voice track by sequentially placing each segment at its correct # position with silence gaps,then concatenating everything. # This is DIFFERENT from amix which plays all streams simultaneously. valid_segments = [(i,seg) for i,seg in enumerate(segments) if seg.get("audio_path") and Path(seg["audio_path"]).exists()] if not valid_segments: print("[SUPERTONIC] No valid segment audio files") return None voice_path = str(work_dir / "voice_es.mp3") if len(valid_segments) == 1: idx,seg = valid_segments[0] if seg["start"] > 0.05: delay_ms = round(seg["start"] * 1000) # Round instead of truncating try: delayed_path = str(work_dir / "st_voice_delayed.wav") proc = subprocess.run([ "ffmpeg","-y","-i",seg["audio_path"], "-af",f"adelay={delay_ms}|{delay_ms}", delayed_path, ],capture_output=True,timeout=15) if proc.returncode == 0: proc2 = subprocess.run([ "ffmpeg","-y","-i",delayed_path, "-af",f"apad=whole_dur={video_duration}", "-c:a","libmp3lame","-b:a","128k",voice_path, ],capture_output=True,timeout=30) if proc2.returncode == 0: pass else: shutil.copy2(seg["audio_path"],voice_path) else: shutil.copy2(seg["audio_path"],voice_path) except Exception: shutil.copy2(seg["audio_path"],voice_path) else: shutil.copy2(seg["audio_path"],voice_path) else: # SEQUENTIAL approach: place each segment in order with silence gaps # This prevents the overlap bug that amix causes # FIX: Track current_time using ACTUAL audio duration,NOT Whisper timestamps. # Using seg["end"] caused progressive drift because TTS audio is often shorter # than the Whisper segment slot,so the actual audio position diverges from # the Whisper timestamps. With 10 segments,this accumulates 1-2s desync. try: concat_parts = [] # List of audio file paths for sequential concat current_time = 0.0 # Tracks ACTUAL position in output audio stream for idx,seg in valid_segments: seg_start = seg["start"] gap = seg_start - current_time # If there's a gap before this segment,insert silence # FIX: Lowered threshold from 50ms to 10ms to reduce drift if gap > 0.01: silence_path = str(work_dir / f"st_gap_{idx}.wav") gap_duration = min(gap,video_duration - current_time) if gap_duration > 0: proc = subprocess.run([ "ffmpeg","-y", "-f","lavfi","-i","anullsrc=r=44100:cl=mono", "-t",str(gap_duration), "-c:a","pcm_s16le",silence_path, ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(silence_path).exists(): concat_parts.append(silence_path) current_time += gap_duration elif gap < 0: # OVERLAP: Previous audio extends past this segment's start. # This shouldn't happen with proper speed adjustment, # but if it does,log it. Audio starts late by |gap| seconds. print(f"[SUPERTONIC] WARNING: Segment {idx} overlap by {-gap:.3f}s") # Add the segment audio (convert to same format first) seg_wav_path = str(work_dir / f"st_seg_{idx}_aligned.wav") proc = subprocess.run([ "ffmpeg","-y","-i",seg["audio_path"], "-ar","44100","-ac","1","-c:a","pcm_s16le", seg_wav_path, ],capture_output=True,timeout=15) if proc.returncode == 0 and Path(seg_wav_path).exists(): concat_parts.append(seg_wav_path) # FIX: Track current_time using ACTUAL audio duration, # NOT Whisper timestamps. This is the critical desync fix. # The concat builds a physical audio stream — position = sum of all # placed durations. Using seg["end"] ignores actual audio position. actual_seg_duration = self._get_duration_sync(seg_wav_path) current_time += actual_seg_duration if not concat_parts: print("[SUPERTONIC] No audio parts to concatenate") return None # Concatenate all parts sequentially using FFmpeg concat demuxer list_file = str(work_dir / "st_concat_list.txt") with open(list_file,"w") as f: for part_path in concat_parts: f.write(f"file '{part_path}'\n") proc = subprocess.run([ "ffmpeg","-y", "-f","concat","-safe","0","-i",list_file, "-af",f"apad=whole_dur={video_duration}", "-c:a","libmp3lame","-b:a","128k", voice_path, ],capture_output=True,text=True,timeout=60) if proc.returncode != 0: print(f"[SUPERTONIC] Concat failed: {proc.stderr[-300:]}") # Last resort: just concatenate without padding proc2 = subprocess.run([ "ffmpeg","-y", "-f","concat","-safe","0","-i",list_file, "-c:a","libmp3lame","-b:a","128k", voice_path, ],capture_output=True,timeout=30) if proc2.returncode != 0: print(f"[SUPERTONIC] Simple concat also failed: {proc2.stderr[-200:]}") return None except Exception as e: print(f"[SUPERTONIC] Concat error: {e}") return None if not Path(voice_path).exists() or Path(voice_path).stat().st_size <= 1000: print("[SUPERTONIC] Final voice file missing") return None size_kb = Path(voice_path).stat().st_size / 1024 voice_dur = self._get_duration_sync(voice_path) print(f"[SUPERTONIC] Final voice: {size_kb:.0f} KB,{voice_dur:.1f}s") return {"path": voice_path,"segments": segments} async def _generate_gtts_async(self,segments: list,work_dir: Path,video_duration: float) -> Optional[dict]: """Generate Spanish voice using gTTS (Google Translate TTS). Used as fallback when Supertonic 3 is not available. gTTS uses HTTP requests,so it's reliable and harder to block. The voice quality is lower than Supertonic 3 but it's very reliable. Returns dict with "path" and "segments" on success,None on failure. """ if not segments: return None voice_path = str(work_dir / "voice_es.mp3") try: from gtts import gTTS except ImportError: print("[TTS-gTTS] gTTS not installed!") return None # gTTS works best with full text (not per-segment) # Generate full audio,then create timed voice track full_text = " ".join(seg.get("text_es","") for seg in segments if seg.get("text_es")) full_text = self._clean_text_for_tts(full_text) if not full_text: print("[TTS-gTTS] No text to generate") return None # Generate full audio with gTTS gtts_raw_path = str(work_dir / "gtts_raw.mp3") try: print(f"[TTS-gTTS] Generating audio for {len(full_text)} chars...") tts = gTTS(text=full_text[:5000],lang='es',slow=False) tts.save(gtts_raw_path) if not Path(gtts_raw_path).exists() or Path(gtts_raw_path).stat().st_size < 500: print("[TTS-gTTS] Generated file too small or missing") return None gtts_duration = self._get_duration_sync(gtts_raw_path) print(f"[TTS-gTTS] Generated {gtts_duration:.1f}s of audio") except Exception as e: print(f"[TTS-gTTS] Generation failed: {e}") return None # Adjust speed to match video duration if needed if gtts_duration > video_duration * 1.1: # Speed up TTS to fit within video duration tempo = video_duration / gtts_duration tempo = max(tempo,0.5) print(f"[TTS-gTTS] Speeding up to {tempo:.2f}x to fit {video_duration:.1f}s video") try: tempo_path = str(work_dir / "gtts_tempo.mp3") atempo_filter = f"atempo={tempo}" if tempo >= 0.5 else f"atempo=0.5,atempo={tempo/0.5}" proc = subprocess.run([ "ffmpeg","-y","-i",gtts_raw_path, "-filter:a",atempo_filter, tempo_path, ],capture_output=True,timeout=30) if proc.returncode == 0 and Path(tempo_path).exists(): shutil.move(tempo_path,gtts_raw_path) gtts_duration = self._get_duration_sync(gtts_raw_path) print(f"[TTS-gTTS] After speed adjust: {gtts_duration:.1f}s") except Exception as e: print(f"[TTS-gTTS] Speed adjust failed: {e}") # Pad to video duration try: pad_cmd = [ "ffmpeg","-y", "-i",gtts_raw_path, "-af",f"apad=whole_dur={video_duration}", "-c:a","libmp3lame","-b:a","128k", voice_path, ] proc = subprocess.run(pad_cmd,capture_output=True,timeout=30) if proc.returncode == 0 and Path(voice_path).exists(): print(f"[TTS-gTTS] Final voice: {Path(voice_path).stat().st_size / 1024:.0f} KB,{self._get_duration_sync(voice_path):.1f}s") return {"path": voice_path,"segments": segments} else: # If padding failed,just use the raw file shutil.copy2(gtts_raw_path,voice_path) print(f"[TTS-gTTS] Used unpadded audio: {gtts_duration:.1f}s") return {"path": voice_path,"segments": segments} except Exception as e: print(f"[TTS-gTTS] Final processing failed: {e}") shutil.copy2(gtts_raw_path,voice_path) return {"path": voice_path,"segments": segments} def _clean_text_for_tts(self,text: str) -> str: """Clean text before TTS to prevent garbled audio artifacts. TTS engines can produce garbled output when text contains: - Number sequences that get read as digits - Special characters or formatting - Excessive punctuation """ # Remove any timestamp-like patterns text = re.sub(r'\d{1,2}:\d{2}(?::\d{2})?(?:[,\.]\d+)?','',text) # Remove standalone number sequences (like "000101010") text = re.sub(r'\b\d{4,}\b','',text) # Remove SRT-style numbering text = re.sub(r'^\d+\s*$','',text,flags=re.MULTILINE) # Remove arrow patterns --> text = re.sub(r'-->\s*\d','',text) # Remove any remaining timestamp artifacts text = re.sub(r'\d{2},\d{3}','',text) # Remove excessive whitespace text = re.sub(r'\s+',' ',text).strip() # Remove markdown-style formatting text = re.sub(r'[*_#`]','',text) # Ensure text ends with period for natural TTS pacing if text and text[-1] not in '.!?': text += '.' # Add commas for natural pauses (every ~15 words if no punctuation) words = text.split() result = [] since_punct = 0 for w in words: result.append(w) since_punct += 1 if since_punct >= 12 and w[-1] not in ',.;:!?': result.append(',') since_punct = 0 elif w[-1] in ',.;:!?': since_punct = 0 text = ' '.join(result) # Clean up double commas/periods text = re.sub(r',',',',text) text = re.sub(r'\.\.','.',text) text = re.sub(r',\.','.',text) return text.strip() def _split_text_for_tts(self,text: str,max_chars: int = 3000) -> list[str]: """Split text into chunks at sentence boundaries for TTS processing.""" if len(text) <= max_chars: return [text] chunks = [] sentences = re.split(r'(?<=[.!?])\s+',text) current_chunk = "" for sentence in sentences: if len(current_chunk) + len(sentence) + 1 > max_chars: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = sentence else: current_chunk += " " + sentence if current_chunk else sentence if current_chunk.strip(): chunks.append(current_chunk.strip()) return chunks def _get_duration_sync(self,path: str) -> float: """Get media file duration in seconds using ffprobe.""" try: proc = subprocess.run([ "ffprobe","-v","quiet", "-show_entries","format=duration", "-of","default=noprint_wrappers=1:nokey=1", path, ],capture_output=True,text=True,timeout=30) return float(proc.stdout.strip()) except Exception: return 0.0 # Return 0 instead of 60.0 — safer default (60 would cause catastrophic speed adjustment) # ================================================================ # Step 7: Compose Video (REDESIGNED - uses segments,not word_boundaries) # ================================================================ # Background music path (shipped with the app) BG_MUSIC_PATH = Path(os.getenv("BG_MUSIC_PATH","/app/assets/background_music.mp3")) def _ensure_bg_music(self): """Ensure background music file exists. Generate a subtle ambient track if missing.""" if self.BG_MUSIC_PATH.exists() and self.BG_MUSIC_PATH.stat().st_size > 1000: return True # Generate a subtle cinematic ambient background track print(f"[COMPOSE] Generating ambient background music at {self.BG_MUSIC_PATH}...") try: self.BG_MUSIC_PATH.parent.mkdir(parents=True,exist_ok=True) proc = subprocess.run([ "ffmpeg","-y", "-f","lavfi","-i","sine=frequency=110:duration=120", "-f","lavfi","-i","sine=frequency=165:duration=120", "-f","lavfi","-i","sine=frequency=220:duration=120", "-filter_complex", "[0:a]volume=0.06[a1];[1:a]volume=0.04[a2];[2:a]volume=0.02[a3];" "[a1][a2][a3]amix=inputs=3:duration=longest,lowpass=f=600,volume=0.15," "afade=t=in:st=0:d=3,afade=t=out:st=115:d=5[out]", "-map","[out]","-c:a","libmp3lame","-b:a","64k", str(self.BG_MUSIC_PATH), ],capture_output=True,timeout=60) if proc.returncode == 0 and self.BG_MUSIC_PATH.exists(): print(f"[COMPOSE] Background music generated: {self.BG_MUSIC_PATH.stat().st_size / 1024:.0f} KB") return True else: print(f"[COMPOSE] BG music generation failed: {proc.stderr[-200:] if proc.stderr else 'unknown'}") except Exception as e: print(f"[COMPOSE] BG music generation error: {e}") return False def _auto_detect_sub_positions(self,video_path: str) -> Optional[dict]: """Auto-detect English subtitle position in a video frame using pixel analysis. Strategy: 1. Extract MULTIPLE frames from different timestamps (more reliable than single frame) 2. Analyze the bottom 40% of the frame for high-contrast text regions 3. English subtitles are typically white text on a semi-transparent background 4. Return the detected region for blur placement Returns dict with blur_x,blur_y,blur_w,blur_h or None if detection fails. """ try: import numpy as np from PIL import Image duration = self._get_duration_sync(video_path) if duration <= 0: return None # Extract frames from 3 different timestamps for more reliable detection frame_paths = [] seek_times = [ max(0,duration * 0.25), # 25% into the video max(0,duration * 0.5), # middle max(0,duration * 0.75), # 75% into the video ] for i,seek_time in enumerate(seek_times): frame_path = str(Path(video_path).parent / f"detect_frame_{i}.png") proc = subprocess.run([ "ffmpeg","-y", "-ss",str(seek_time), "-i",video_path, "-vframes","1", "-vf",f"scale={TARGET_WIDTH}:{TARGET_HEIGHT}:force_original_aspect_ratio=increase,crop={TARGET_WIDTH}:{TARGET_HEIGHT}", frame_path, ],capture_output=True,timeout=30) if Path(frame_path).exists(): frame_paths.append(frame_path) if not frame_paths: print("[DETECT] Frame extraction failed") return None # Analyze all frames and find the most common subtitle region all_detected_regions = [] for frame_path in frame_paths: try: img = Image.open(frame_path).convert("L") arr = np.array(img) height,width = arr.shape # Analyze the bottom 40% of the frame (where subs typically appear) bottom_start = int(height * 0.60) bottom_region = arr[bottom_start:,:] # Find rows with high brightness (subtitle text is bright) # Subtitle background is often semi-transparent with bright text row_max_brightness = np.max(bottom_region,axis=1) row_std = np.std(bottom_region,axis=1) # Combine: high brightness AND high variance = subtitle text # Use a weighted score brightness_score = row_max_brightness / 255.0 variance_score = row_std / (np.max(row_std) + 1) combined_score = brightness_score * 0.6 + variance_score * 0.4 # Find contiguous high-score region (subtitle band) threshold = np.mean(combined_score) + np.std(combined_score) high_score_rows = combined_score > threshold if not np.any(high_score_rows): continue # Find the largest contiguous block of high-score rows # This is the subtitle region changes = np.diff(high_score_rows.astype(int)) starts = np.where(changes == 1)[0] + 1 ends = np.where(changes == -1)[0] + 1 # Handle edge cases if high_score_rows[0]: starts = np.insert(starts,0,0) if high_score_rows[-1]: ends = np.append(ends,len(high_score_rows)) if len(starts) == 0 or len(ends) == 0: continue # Find the largest contiguous region lengths = ends - starts best_idx = np.argmax(lengths) if lengths[best_idx] < 20: # Too small,probably not subtitles continue detected_y_local = starts[best_idx] detected_h_local = int(lengths[best_idx]) # Expand region slightly for better coverage padding_top = max(5,int(detected_h_local * 0.3)) padding_bottom = max(5,int(detected_h_local * 0.3)) detected_y_local = max(0,detected_y_local - padding_top) detected_h_local = min(bottom_region.shape[0] - detected_y_local, detected_h_local + padding_top + padding_bottom) # Convert back to full-frame coordinates detected_y = bottom_start + detected_y_local detected_h = detected_h_local # Ensure minimum height for the blur detected_h = max(detected_h,80) all_detected_regions.append({ "y": detected_y, "h": detected_h, }) finally: try: Path(frame_path).unlink() except: pass if not all_detected_regions: print("[DETECT] No subtitle region detected in any frame") return None # Use the MEDIAN position from all detected regions (most robust) ys = [r["y"] for r in all_detected_regions] hs = [r["h"] for r in all_detected_regions] final_y = int(np.median(ys)) final_h = int(np.median(hs)) # Full width for the blur (subs typically span the full width) result = { "blur_x": 0, "blur_y": final_y, "blur_w": TARGET_WIDTH, "blur_h": final_h, } print(f"[DETECT] Subtitle region: y={final_y},h={final_h} (from {len(all_detected_regions)} frames)") return result except ImportError: print("[DETECT] Pillow/numpy not available for auto-detection") return None except Exception as e: print(f"[DETECT] Auto-detect error: {e}") import traceback traceback.print_exc() return None def _get_visual_config(self) -> dict: """Get visual config from state,falling back to defaults.""" try: vc = self.state.get_visual_config() # Merge with defaults for any missing keys result = dict(DEFAULT_VISUAL_CONFIG) result.update(vc) return result except Exception: return dict(DEFAULT_VISUAL_CONFIG) def _compose_video_sync(self,video_path: str,voice_path: str,segments: list,work_dir: Path) -> Optional[str]: """Compose final video: flip,blur English subs,ASS Spanish subs,Spanish voice,bg music. ROBUST SINGLE-PASS approach: 1. Take original video 2. Scale + crop + flip (visual processing) 3. Add transparent blur over English subtitle area 4. Overlay ASS Spanish subtitles 5. Replace audio ENTIRELY with Spanish TTS voice (+ optional low bg music) CRITICAL: Original English audio is NEVER included. Only Spanish TTS + bg music. VERIFICATION: Checks that voice_path has actual audio content before composing. If voice is missing/silent,generates an emergency voice from the translated text. """ final_path = str(work_dir / "final.mp4") # Read visual config from state (saved from dashboard) vc = self._get_visual_config() # Skip auto-detect - use fixed values for consistent results print(f"[COMPOSE] Using fixed visual config (no auto-detect)") blur_x = int(vc.get("blur_x",0)) blur_y = int(vc.get("blur_y",1100)) blur_w = int(vc.get("blur_w",720)) blur_h = int(vc.get("blur_h",180)) blur_strength = int(vc.get("blur_strength",25)) print(f"[COMPOSE] Visual config: blur=({blur_x},{blur_y},{blur_w},{blur_h}) str={blur_strength}") # POSITION Spanish subtitles ON the blur region (replacing English captions) # Spanish subs appear in the SAME position where English captions were. # If user did NOT explicitly set sub_margin_v, auto-compute it from blur_y/blur_h # so subtitles sit centered on the blur. # MarginV (ASS) = distance from BOTTOM of frame to BOTTOM of subtitle text. # subtitle_bottom_y = blur_y + blur_h - 18 (small padding from blur bottom) # MarginV = PlayResY - subtitle_bottom_y = PlayResY - (blur_y + blur_h - 18) sub_font_size = int(vc.get("sub_font_size", 42)) if "sub_margin_v" in vc and vc.get("sub_margin_v") is not None: sub_margin_v = int(vc.get("sub_margin_v", 288)) else: # Auto-compute so subs sit ON the blur (replacing English captions) sub_margin_v = TARGET_HEIGHT - (blur_y + blur_h) + 18 print(f"[COMPOSE] Auto sub_margin_v={sub_margin_v} (subs ON blur at y={blur_y}-{blur_y+blur_h})") print(f"[COMPOSE] Subtitle config: font={sub_font_size} margin_v={sub_margin_v}") # Generate ASS subtitle file from segments (uses visual config for font/margin) ass_path = str(work_dir / "subtitles.ass") if segments: # Log how many segments have text_es with_es = sum(1 for s in segments if s.get("text_es","").strip()) print(f"[COMPOSE] Segments: {len(segments)}, with text_es: {with_es}") self._write_ass(segments,ass_path,work_dir,visual_config=vc) # Log ASS content for debugging try: ass_content = Path(ass_path).read_text() event_count = ass_content.count("Dialogue:") print(f"[COMPOSE] ASS generated: {event_count} events, {len(ass_content)} chars") # Print first 3 events for line in ass_content.split('\n'): if line.startswith('Dialogue:'): print(f" ASS: {line[:120]}") # Save ASS to dataset for debugging try: from huggingface_hub import HfApi as _HfApi2 _api2 = _HfApi2(token=os.getenv("HF_TOKEN","")) _ds = os.getenv("HF_STATE_DATASET","") if _ds: _api2.upload_file( path_or_fileobj=ass_path, path_in_repo="processed/last_subtitles.ass", repo_id=_ds, repo_type="dataset", token=os.getenv("HF_TOKEN","") ) print("[COMPOSE] ASS saved to dataset") except Exception as _e2: print(f"[COMPOSE] ASS save failed: {_e2}") except Exception as e: print(f"[COMPOSE] Could not read ASS: {e}") else: print("[COMPOSE] WARNING: No segments! Writing empty ASS") self._write_empty_ass(ass_path) try: video_duration = self._get_duration_sync(video_path) voice_duration = self._get_duration_sync(voice_path) final_duration = max(video_duration,voice_duration) print(f"[COMPOSE] Video: {video_duration:.1f}s,Voice: {voice_duration:.1f}s") # ============================================= # VERIFY voice file has actual audio content # ============================================= # RELAXED CHECK: We prefer having audio (even quiet) over silence. # The old -60dB threshold was too strict and could reject valid TTS. voice_has_audio = False voice_file_exists = Path(voice_path).exists() and Path(voice_path).stat().st_size > 500 if voice_file_exists: try: probe_cmd = ["ffprobe","-v","quiet","-select_streams","a", "-show_entries","stream=duration,codec_type", "-of","csv=s=x:p=0",voice_path] probe_result = subprocess.run(probe_cmd,capture_output=True,text=True,timeout=10) print(f"[COMPOSE] Voice probe result: '{probe_result.stdout.strip()}'") if 'audio' in probe_result.stdout: # File has an audio stream - use it! # Only skip if volumedetect shows truly dead silence (< -90dB) try: vol_cmd = ["ffmpeg","-i",voice_path,"-af","volumedetect", "-f","null","-"] vol_result = subprocess.run(vol_cmd,capture_output=True,text=True,timeout=15) vol_output = vol_result.stderr if "max_volume" in vol_output: vol_match = re.search(r'max_volume\s*:\s*([-.\d]+)\s*dB',vol_output) if vol_match: max_vol = float(vol_match.group(1)) # -90dB = essentially digital silence (16-bit noise floor) # -60dB was too strict,rejecting quiet but valid TTS if max_vol > -90: voice_has_audio = True print(f"[COMPOSE] Voice audio OK: max_volume={max_vol:.1f}dB (threshold=-90dB)") else: print(f"[COMPOSE] WARNING: Voice appears silent (max_volume={max_vol:.1f}dB < -90dB)") else: voice_has_audio = True print("[COMPOSE] Voice volumedetect: no max_volume match,assuming OK") else: voice_has_audio = True print("[COMPOSE] Voice volumedetect: no output,assuming OK") except Exception as ve: voice_has_audio = True print(f"[COMPOSE] Volumedetect error (non-fatal): {ve},assuming voice is OK") else: print(f"[COMPOSE] WARNING: Voice file has no audio stream! Size={Path(voice_path).stat().st_size}") # File exists but no audio stream - try to re-encode it try: reencoded = str(work_dir / "voice_reencoded.mp3") reenc_cmd = ["ffmpeg","-y","-i",voice_path, "-ar","44100","-ac","1","-b:a","128k", reencoded] reenc_proc = subprocess.run(reenc_cmd,capture_output=True,text=True,timeout=30) if reenc_proc.returncode == 0 and Path(reencoded).exists() and Path(reencoded).stat().st_size > 500: shutil.move(reencoded,voice_path) voice_has_audio = True print(f"[COMPOSE] Voice re-encoded successfully") except Exception as re: print(f"[COMPOSE] Voice re-encode failed: {re}") except Exception as e: print(f"[COMPOSE] Voice verification error: {e}") # On error,assume audio exists (safer than -an) voice_has_audio = True else: print(f"[COMPOSE] WARNING: Voice file missing or too small (size={Path(voice_path).stat().st_size if Path(voice_path).exists() else 0})") # If voice is missing/silent,generate emergency voice if not voice_has_audio: print("[COMPOSE] EMERGENCY: Generating voice from translated text...") full_text = " ".join(seg.get("text_es","") for seg in segments if seg.get("text_es")) full_text = self._clean_text_for_tts(full_text) if full_text: # Try Remote Supertonic first if SUPERTONIC_REMOTE_URLS: try: wav_bytes = _supertonic_remote_tts(full_text[:3000],SUPERTONIC_VOICE,SUPERTONIC_LANG) if wav_bytes and len(wav_bytes) > 1000: emergency_path = str(work_dir / "emergency_supertonic.wav") with open(emergency_path,"wb") as f: f.write(wav_bytes) if Path(emergency_path).exists() and Path(emergency_path).stat().st_size > 1000: shutil.move(emergency_path,voice_path) voice_duration = self._get_duration_sync(voice_path) final_duration = max(video_duration,voice_duration) voice_has_audio = True print(f"[COMPOSE] Emergency REMOTE Supertonic voice OK: {voice_duration:.1f}s") except Exception as se: print(f"[COMPOSE] Emergency remote Supertonic failed: {se}") # Try local Supertonic fallback if not voice_has_audio and SUPERTONIC_AVAILABLE: try: model = _get_supertonic_model() if model is not None: style = model.get_voice_style(voice_name=SUPERTONIC_VOICE) wav,duration = model.synthesize(full_text[:3000],voice_style=style,lang=SUPERTONIC_LANG) emergency_path = str(work_dir / "emergency_supertonic.wav") model.save_audio(wav,emergency_path) if Path(emergency_path).exists() and Path(emergency_path).stat().st_size > 1000: shutil.move(emergency_path,voice_path) voice_duration = self._get_duration_sync(voice_path) final_duration = max(video_duration,voice_duration) voice_has_audio = True print(f"[COMPOSE] Emergency LOCAL Supertonic voice OK: {voice_duration:.1f}s") except Exception as se: print(f"[COMPOSE] Emergency local Supertonic failed: {se}") # If Supertonic failed,try gTTS if not voice_has_audio: try: from gtts import gTTS gtts_path = str(work_dir / "emergency_gtts.mp3") tts = gTTS(text=full_text[:5000],lang='es',slow=False) tts.save(gtts_path) if Path(gtts_path).exists() and Path(gtts_path).stat().st_size > 1000: shutil.move(gtts_path,voice_path) voice_duration = self._get_duration_sync(voice_path) final_duration = max(video_duration,voice_duration) voice_has_audio = True print(f"[COMPOSE] Emergency gTTS OK: {voice_duration:.1f}s") except Exception as ge: print(f"[COMPOSE] Emergency gTTS also failed: {ge}") # Find font directories for ASS subtitle rendering # FIX: More comprehensive font search + auto-download Montserrat if missing font_dirs = [ "/usr/share/fonts/truetype/montserrat", "/usr/share/fonts/truetype/dejavu", "/app/fonts", str(Path(__file__).parent / "fonts"), "/usr/share/fonts/truetype/english", # Additional system font dirs "/usr/share/fonts/truetype/freefont", "/usr/share/fonts/truetype/liberation", ] fontsdir = "" for fd in font_dirs: if Path(fd).exists() and any(Path(fd).glob("*.ttf")): fontsdir = fd break # FIX (2026-07-03): Use Tahoma font (user-requested). # Tahoma Bold (Bold=-1 in ASS style) for normal text, # Tahoma Bold + outline 4px (simulated extra-bold) for active word. # Tahoma is a Microsoft font - auto-download from github.com/dolbydu/font if missing. tahoma_found = False if fontsdir: tahoma_found = any( "tahoma" in f.name.lower() for f in Path(fontsdir).glob("*.ttf") ) # Also check system fonts if not tahoma_found: for sys_font_dir in ["/usr/share/fonts/truetype/tahoma", "/usr/share/fonts/truetype/ofl/tahoma", "/usr/share/fonts/opentype/tahoma", "/usr/share/fonts/tahoma"]: if Path(sys_font_dir).exists() and any(Path(sys_font_dir).glob("*.ttf")): fontsdir = sys_font_dir tahoma_found = True break # Also check if Tahoma is registered in fontconfig if not tahoma_found: try: result = subprocess.run(["fc-list"], capture_output=True, text=True, timeout=5) if "Tahoma" in result.stdout: tahoma_found = True fontsdir = "" except Exception: pass if not tahoma_found: # Try to download Tahoma + Tahoma Bold to the fonts directory fonts_target = Path("/app/fonts/tahoma") fonts_target.mkdir(parents=True,exist_ok=True) regular_file = fonts_target / "Tahoma-Regular.ttf" bold_file = fonts_target / "Tahoma-Bold.ttf" if not regular_file.exists() or not bold_file.exists(): try: print("[COMPOSE] Downloading Tahoma fonts...") import urllib.request if not regular_file.exists(): urllib.request.urlretrieve( "https://raw.githubusercontent.com/dolbydu/font/master/Sans/Tahoma/tahoma.ttf", str(regular_file), ) if not bold_file.exists(): urllib.request.urlretrieve( "https://raw.githubusercontent.com/dolbydu/font/master/Sans/Tahoma/tahomabd.ttf", str(bold_file), ) print(f"[COMPOSE] Tahoma downloaded: {fonts_target}") try: subprocess.run(["fc-cache", "-f"], capture_output=True, timeout=10) except Exception: pass except Exception as e: print(f"[COMPOSE] Font download failed: {e}") if regular_file.exists() or bold_file.exists(): fontsdir = str(fonts_target) tahoma_found = True print(f"[COMPOSE] Using Tahoma from: {fontsdir}") if not tahoma_found: print("[COMPOSE] WARNING: Tahoma not found, subtitles may use fallback font") # Build ass filter with fontsdir ass_escaped = ass_path.replace("\\","/").replace(":","\\:") ass_filter = f"ass={ass_escaped}" if fontsdir: fontsdir_escaped = fontsdir.replace("\\","/").replace(":","\\:") ass_filter += f":fontsdir={fontsdir_escaped}" # Target resolution target_w,target_h = TARGET_WIDTH,TARGET_HEIGHT # ============================================= # SINGLE PASS: Scale + Crop + Flip + Blur + ASS + Audio # ============================================= # More reliable than two-pass because: # - No intermediate file to corrupt # - No chance of audio leaking from original # - Faster (one encode instead of two) print("[COMPOSE] Single pass: Scale + Flip + Blur + ASS + Audio...") # Ensure background music exists self._ensure_bg_music() bg_music_path = str(self.BG_MUSIC_PATH) has_bg_music = self.BG_MUSIC_PATH.exists() and self.BG_MUSIC_PATH.stat().st_size > 1000 # Video filter chain: scale -> crop -> GRADIENT blur -> ASS subs # FIX (2026-07-03): Replaced rectangular crop+blur+overlay with # gblur + alpha gradient (geq). The old approach created SHARP EDGES # at the blur rectangle boundary, which looked like a "second blur" # to the user. The gradient alpha approach fades the blur in smoothly # over 100px, eliminating the visible transition line. # # Gradient: alpha=0 (transparent) for Y < (blur_y - 100) # gradient 0→255 for (blur_y - 100) ≤ Y < blur_y # alpha=255 (opaque) for Y ≥ blur_y # This means the blur is fully opaque where English captions are # (y=920-960) and fades smoothly to sharp above. gradient_start = max(0, blur_y - 100) video_filter = ( f"scale={target_w}:{target_h}:force_original_aspect_ratio=increase," f"crop={target_w}:{target_h}," f"split[base][forblur];" f"[forblur]gblur=sigma=15,format=yuva420p," f"geq=lum='p(X,Y)':cb='p(X,Y)':cr='p(X,Y)':" f"a='if(gt(Y,{gradient_start}),min(255,(Y-{gradient_start})*2.55),0)'[blurred_grad];" f"[base][blurred_grad]overlay=0:0:format=auto[withblur];" f"[withblur]{ass_filter}[v]" ) # Audio filter: Spanish voice (+ optional low bg music) if voice_has_audio and has_bg_music: audio_filter = ( f"[1:a]aresample=44100,volume=1.5,apad=whole_dur={final_duration}[voice];" f"[2:a]aresample=44100,volume=0.06,afade=t=in:st=0:d=1.5," f"afade=t=out:st={max(0,final_duration - 2)}:d=2," f"apad=whole_dur={final_duration}[bgm];" f"[voice][bgm]amix=inputs=2:duration=first:dropout_transition=3:normalize=0[a]" ) cmd = [ "ffmpeg","-y", "-i",video_path, "-i",voice_path, "-i",bg_music_path, "-filter_complex",video_filter + ";" + audio_filter, "-map","[v]","-map","[a]", "-c:v","libx264","-preset","veryfast","-crf","23", "-pix_fmt","yuv420p", "-c:a","aac","-b:a","192k", "-movflags","+faststart", "-shortest",final_path, ] elif voice_has_audio: audio_filter = ( f"[1:a]aresample=44100,volume=1.5,apad=whole_dur={final_duration}[a]" ) cmd = [ "ffmpeg","-y", "-i",video_path, "-i",voice_path, "-filter_complex",video_filter + ";" + audio_filter, "-map","[v]","-map","[a]", "-c:v","libx264","-preset","veryfast","-crf","23", "-pix_fmt","yuv420p", "-c:a","aac","-b:a","192k", "-movflags","+faststart", "-shortest",final_path, ] else: # NO TTS voice - use original video audio as fallback (better than silence!) # This prevents the "NO AUDIO" bug where videos upload completely mute print("[COMPOSE] WARNING: No TTS voice! Using original video audio as fallback (English audio,low volume)") audio_filter_fallback = ( f"[0:a]aresample=44100,volume=0.5,apad=whole_dur={final_duration}[a]" ) cmd = [ "ffmpeg","-y", "-i",video_path, "-filter_complex",video_filter + ";" + audio_filter_fallback, "-map","[v]","-map","[a]", "-c:v","libx264","-preset","veryfast","-crf","23", "-pix_fmt","yuv420p", "-c:a","aac","-b:a","192k", "-movflags","+faststart", "-shortest",final_path, ] proc = subprocess.run(cmd,capture_output=True,text=True,timeout=180) if proc.returncode != 0: print(f"[COMPOSE] Main pass FAILED: {proc.stderr[-500:]}") # Fallback 1: Simpler video filter (no blur,just ASS + audio) print("[COMPOSE] Fallback 1: Scale + Flip + ASS + Audio (no blur)...") simple_vf = ( f"scale={target_w}:{target_h}:force_original_aspect_ratio=increase," f"crop={target_w}:{target_h},{ass_filter}[v]" ) if voice_has_audio: simple_audio = f"[1:a]aresample=44100,volume=1.5,apad=whole_dur={final_duration}[a]" fb1_cmd = [ "ffmpeg","-y", "-i",video_path,"-i",voice_path, "-filter_complex",simple_vf + ";" + simple_audio, "-map","[v]","-map","[a]", "-c:v","libx264","-preset","veryfast","-crf","23", "-pix_fmt","yuv420p", "-c:a","aac","-b:a","192k", "-movflags","+faststart", "-shortest",final_path, ] else: # No TTS voice - use original audio instead of silence fb1_cmd = [ "ffmpeg","-y", "-i",video_path, "-vf",f"scale={target_w}:{target_h}:force_original_aspect_ratio=increase,crop={target_w}:{target_h}", "-c:v","libx264","-preset","veryfast","-crf","23", "-pix_fmt","yuv420p", "-c:a","aac","-b:a","192k", "-movflags","+faststart",final_path, ] proc = subprocess.run(fb1_cmd,capture_output=True,text=True,timeout=180) if proc.returncode != 0: print(f"[COMPOSE] Fallback 1 FAILED: {proc.stderr[-300:]}") # Fallback 2: Simplest - scale + audio,no subs if voice_has_audio: fb2_cmd = [ "ffmpeg","-y", "-i",video_path,"-i",voice_path, "-vf",f"scale={target_w}:{target_h}:force_original_aspect_ratio=increase,crop={target_w}:{target_h}", "-map","0:v","-map","1:a", "-c:v","libx264","-preset","veryfast","-crf","23", "-pix_fmt","yuv420p", "-c:a","aac","-b:a","192k", "-shortest",final_path, ] proc = subprocess.run(fb2_cmd,capture_output=True,text=True,timeout=120) if proc.returncode != 0: print(f"[COMPOSE] All approaches failed!") return None else: # Last resort: just re-encode with original audio fb2_cmd = [ "ffmpeg","-y", "-i",video_path, "-vf",f"scale={target_w}:{target_h}:force_original_aspect_ratio=increase,crop={target_w}:{target_h}", "-c:v","libx264","-preset","veryfast","-crf","23", "-pix_fmt","yuv420p", "-c:a","aac","-b:a","192k", "-shortest",final_path, ] proc = subprocess.run(fb2_cmd,capture_output=True,text=True,timeout=120) if proc.returncode != 0: print(f"[COMPOSE] All approaches failed!") return None # Verify output if not Path(final_path).exists(): print("[COMPOSE] Final file not found!") return None size_mb = Path(final_path).stat().st_size / 1024 / 1024 # ============================================= # POST-COMPOSITION AUDIO RESCUE # ============================================= # If the final video has no audio but we have a voice file, # remux the audio in as a safety net. This is the LAST line # of defense against the "no audio" bug. has_audio_stream = False verify_cmd = [ "ffprobe","-v","quiet", "-show_entries","stream=codec_type,width,height,duration", "-of","csv=s=x:p=0", final_path, ] try: verify_result = subprocess.run(verify_cmd,capture_output=True,text=True,timeout=10) streams = verify_result.stdout.strip() print(f"[COMPOSE] Output streams: {streams}") has_audio_stream = 'audio' in streams if not has_audio_stream: print("[COMPOSE] CRITICAL: No audio in output! Attempting audio rescue...") else: print("[COMPOSE] Audio track present in output") except: pass if not has_audio_stream: # RESCUE: Try to add voice audio via simple remux rescue_path = str(work_dir / "final_rescued.mp4") rescue_success = False # Try 1: Remux with TTS voice if voice_has_audio and Path(voice_path).exists(): try: rescue_cmd = [ "ffmpeg","-y", "-i",final_path,"-i",voice_path, "-c:v","copy","-c:a","aac","-b:a","192k", "-map","0:v","-map","1:a", "-shortest",rescue_path, ] rescue_proc = subprocess.run(rescue_cmd,capture_output=True,text=True,timeout=60) if rescue_proc.returncode == 0 and Path(rescue_path).exists(): shutil.move(rescue_path,final_path) rescue_success = True print("[COMPOSE] Audio RESCUE: Added TTS voice via remux!") except Exception as e: print(f"[COMPOSE] Audio rescue with voice failed: {e}") # Try 2: Remux with original video audio if not rescue_success: try: rescue_cmd = [ "ffmpeg","-y", "-i",video_path,"-i",final_path, "-c:v","copy","-c:a","aac","-b:a","192k", "-map","1:v","-map","0:a", "-shortest",rescue_path, ] rescue_proc = subprocess.run(rescue_cmd,capture_output=True,text=True,timeout=60) if rescue_proc.returncode == 0 and Path(rescue_path).exists(): shutil.move(rescue_path,final_path) rescue_success = True print("[COMPOSE] Audio RESCUE: Added original audio via remux!") except Exception as e: print(f"[COMPOSE] Audio rescue with original audio failed: {e}") # Try 3: Generate silence track (at least the video won't be completely mute) if not rescue_success: try: silence_path = str(work_dir / "silence_track.aac") vid_dur = self._get_duration_sync(final_path) silence_cmd = [ "ffmpeg","-y", "-f","lavfi","-i",f"anullsrc=r=44100:cl=stereo", "-t",str(vid_dur),"-c:a","aac","-b:a","128k", silence_path, ] subprocess.run(silence_cmd,capture_output=True,timeout=30) rescue_cmd = [ "ffmpeg","-y", "-i",final_path,"-i",silence_path, "-c:v","copy","-c:a","copy", "-map","0:v","-map","1:a", "-shortest",rescue_path, ] rescue_proc = subprocess.run(rescue_cmd,capture_output=True,text=True,timeout=60) if rescue_proc.returncode == 0 and Path(rescue_path).exists(): shutil.move(rescue_path,final_path) print("[COMPOSE] Audio RESCUE: Added silence track (last resort)") except Exception as e: print(f"[COMPOSE] Even silence rescue failed: {e}") print(f"[COMPOSE] Final video: {size_mb:.1f} MB (Voice: {voice_has_audio},BG music: {has_bg_music})") return final_path except subprocess.TimeoutExpired: print("[COMPOSE] Timeout (5 min)") return None except Exception as e: print(f"[COMPOSE] Error: {e}") import traceback traceback.print_exc() return None # ================================================================ # ASS Subtitle Generation (REDESIGNED - uses segments,not word_boundaries) # ================================================================ def _write_ass(self,segments: list,path: str,work_dir: Path,visual_config: dict = None): """Generate ASS subtitle file with @LaHistoriaDeEl style captions. Style matches @LaHistoriaDeEl (1M+ subs reference channel): - Font: Montserrat Black (weight 900) - ALL CAPS for impact - White text with thick black outline (high readability) - Yellow highlight for emphasis word - 2 words per subtitle change (karaoke style) - Position read from visual_config (saved from dashboard) """ if not segments: self._write_empty_ass(path) return # Read visual config for subtitle positioning if visual_config is None: visual_config = self._get_visual_config() sub_font_size = int(visual_config.get("sub_font_size",42)) sub_margin_v = int(visual_config.get("sub_margin_v",288)) sub_alignment = int(visual_config.get("sub_alignment",2)) sub_margin_l = int(visual_config.get("sub_margin_l",20)) sub_margin_r = int(visual_config.get("sub_margin_r",20)) # Build subtitle events from segments # FIX (2026-07-03): Use SAME sentence grouping as voice generation. # Voice uses _group_segments_into_sentences() to merge short Whisper segments # into full sentences for better prosody. Subs MUST use the same grouping # or they will be desynced from the voice. # FIX 2 (2026-07-03): Do NOT split sentences into halves. The voice reads # the FULL sentence, so if subs split into 2 halves with proportional timing, # the voice reaches the second half BEFORE the sub switches = voice ahead of subs. # Now: one sub = one full sentence = one voice segment, perfectly synced 1:1. sentence_segments = self._group_segments_into_sentences(segments) events = [] for seg_idx,seg in enumerate(sentence_segments): text_es = seg.get("text_es","").strip() if not text_es: continue start = seg["start"] end = seg["end"] is_last_segment = (seg_idx == len(sentence_segments) - 1) # One sub per sentence (NO splitting) - matches voice exactly events.append({ "text": text_es, "start": start, "end": end, "is_last_in_segment": True, "is_last_overall": is_last_segment, }) with open(path,"w",encoding="utf-8") as f: # ASS Header - resolution matches video (720x1280) f.write("[Script Info]\n") f.write("Title: Spanish Subtitles - La Historia de Ella\n") f.write("ScriptType: v4.00+\n") f.write("WrapStyle: 0\n") f.write(f"PlayResX: {TARGET_WIDTH}\n") f.write(f"PlayResY: {TARGET_HEIGHT}\n") f.write("ScaledBorderAndShadow: yes\n") f.write("YCbCr Matrix: TV.709\n") f.write("\n") # Styles - Tahoma Bold (user-requested font) # POP effect: BackColour=&H00000000 (NO shadow box) eliminates the dark # rectangle behind text that looked like a "second blur". # Outline=3 (thicker) for readability without shadow. f.write("[V4+ Styles]\n") f.write("Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,BackColour,Bold,Italic,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding\n") # Default style: WHITE text, Tahoma Bold, outline 3px, NO shadow ol = int(visual_config.get("sub_outline", 3)) f.write(f"Style: Default,Tahoma,{sub_font_size},&H00FFFFFF,&H0000FFFF,&H00000000,&H00000000,-1,0,1,{ol},0,{sub_alignment},{sub_margin_l},{sub_margin_r},{sub_margin_v},1\n") # Highlight style (legacy, kept for compatibility) f.write(f"Style: Highlight,Tahoma,{sub_font_size},&H0000FFFF,&H0000FFFF,&H00000000,&H00000000,-1,0,1,{ol},0,{sub_alignment},{sub_margin_l},{sub_margin_r},{sub_margin_v},1\n") f.write("\n") # Events f.write("[Events]\n") f.write("Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text\n") for event in events: text = event["text"] start = event["start"] end = event["end"] is_last_in_segment = event.get("is_last_in_segment",False) is_last_overall = event.get("is_last_overall",False) # WORD-BY-WORD HIGHLIGHTING (2026-07-03, single-layer color, no flicker): # One Dialogue per word time-slot, each with the FULL TEXT but different # word highlighted in yellow. Since text is identical in all frames, # positioning NEVER changes - only the yellow word moves through the text. # FIX (2026-07-03): Slow down highlighting to 80% speed so it stays # BEHIND the voice (voice is faster than proportional highlighting). # Voice doesn't speak at uniform speed, so proportional highlighting # races ahead. Slowing it to 80% keeps it behind the voice. words_in_event = text.split() if not words_in_event: continue duration = end - start # SLOW DOWN highlighting: use 85% of duration, start 7.5% later # This ensures the yellow word is always BEHIND what the voice is saying highlight_start_offset = duration * 0.075 # start 7.5% into the segment highlight_duration = duration * 0.85 # use 85% of duration for highlighting word_duration = highlight_duration / len(words_in_event) if words_in_event else duration n_words = len(words_in_event) # Generate one Dialogue per word time-slot for i, word in enumerate(words_in_event): word_start = start + highlight_start_offset + i * word_duration word_end = start + highlight_start_offset + (i + 1) * word_duration # Don't exceed segment end word_end = min(word_end, end) word_start_str = self._seconds_to_ass(word_start) word_end_str = self._seconds_to_ass(word_end) # Build text with word i in yellow, rest in white parts = [] for j, w in enumerate(words_in_event): w_upper = w.upper() if j == i: # Active word: yellow + thicker outline (4px vs 3px) parts.append(r"{\c&H0000FFFF&\bord4}" + w_upper + r"{\c&H00FFFFFF&\bord3}") else: parts.append(w_upper) display_text = " ".join(parts) # Fade ONLY on first word (fade in) and last word (fade out) # This prevents flickering between words if n_words == 1: fade = r"{\fad(80,80)}" elif i == 0: fade = r"{\fad(80,0)}" elif i == n_words - 1: fade = r"{\fad(0,80)}" else: fade = "" # ASS Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text # Name and Effect MUST be empty (just consecutive commas) - otherwise # libass misaligns fields and Text becomes empty (subtitles don't render). f.write(f"Dialogue: 0,{word_start_str},{word_end_str},Default,,0,0,0,,{fade}{display_text}\n") print(f"[ASS] Generated {len(events)} subtitle events (karaoke style,2 words each,ALL CAPS)") def _group_words_karaoke(self,words: list,start: float,end: float,seg_idx: int = 0,total_segments: int = 1) -> list: """Split words into 1-2 word chunks with proportional timing (karaoke style). Matches @LaHistoriaDeEl style: subtitles change every 1-2 words, keeping the audience engaged and forced to read continuously. Includes is_last_in_segment and is_last_overall markers for "..." feature. """ if not words: return [] total_words = len(words) duration = end - start is_last_segment = (seg_idx == total_segments - 1) chunks = [] i = 0 while i < total_words: # Take 2 words if available,1 if only 1 left chunk_end = min(i + SUBTITLE_WORDS_PER_EVENT,total_words) chunk_words = words[i:chunk_end] chunk_text = " ".join(chunk_words) # Proportional timing based on word count chunk_start_time = start + (i / total_words) * duration chunk_end_time = start + (chunk_end / total_words) * duration # Small gap between chunks (0.02s) if i > 0 and chunks: chunk_start_time = max(chunk_start_time,chunks[-1]["end"] + 0.02) is_last_in_segment = (chunk_end >= total_words) is_last_overall = is_last_in_segment and is_last_segment chunks.append({ "text": chunk_text, "start": round(chunk_start_time,3), "end": round(chunk_end_time,3), "is_last_in_segment": is_last_in_segment, "is_last_overall": is_last_overall, }) i = chunk_end return chunks def _group_words_for_subtitles_from_text(self,words: list,start: float,end: float) -> list: """Split a list of words into subtitle chunks of ~5 words with proportional timing. Each chunk gets a proportion of the total segment duration based on word count. """ total_words = len(words) duration = end - start chunks = [] chunk_size = 5 # ~5 words per subtitle line for i in range(0,total_words,chunk_size): chunk_words = words[i:i + chunk_size] chunk_text = " ".join(chunk_words) # Proportional timing chunk_start = start + (i / total_words) * duration chunk_end = start + (min(i + chunk_size,total_words) / total_words) * duration # Add small gap between chunks (0.05s) if i > 0: chunk_start = max(chunk_start,chunks[-1]["end"] + 0.01) chunks.append({ "text": chunk_text, "start": round(chunk_start,3), "end": round(chunk_end,3), }) return chunks def _group_words_for_subtitles(self,word_boundaries: list) -> list: """Group word boundaries into 1-3 word chunks for subtitle display. Legacy method kept for backward compatibility. Used when falling back to word_boundaries from TTS. """ chunks = [] current_words = [] chunk_start = None for i,wb in enumerate(word_boundaries): word = wb["word"] start = wb["start"] end = wb["end"] if chunk_start is None: chunk_start = start current_words.append(word) should_close = False if i == len(word_boundaries) - 1: should_close = True elif word and word[-1] in '.,!?;:': should_close = True elif len(current_words) >= 3: should_close = True elif len(current_words) >= 2 and i + 1 < len(word_boundaries) and len(word_boundaries[i + 1]["word"]) >= 9: should_close = True if should_close: chunks.append({ "words": current_words, "start": chunk_start, "end": end, }) current_words = [] chunk_start = None return chunks def _write_empty_ass(self,path: str): """Write a minimal empty ASS file.""" with open(path,"w",encoding="utf-8") as f: f.write(f"[Script Info]\nScriptType: v4.00+\nPlayResX: {TARGET_WIDTH}\nPlayResY: {TARGET_HEIGHT}\n\n") f.write("[V4+ Styles]\nFormat: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,BackColour,Bold,Italic,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding\n") f.write(f"Style: Default,Tahoma,42,&H00FFFFFF,&H0000FFFF,&H00000000,&H00000000,-1,0,1,8,2,2,20,20,40,1\n\n") f.write("[Events]\nFormat: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text\n") def _seconds_to_ass(self,seconds: float) -> str: """Convert seconds to ASS timestamp format (H:MM:SS.CC).""" h = int(seconds // 3600) m = int((seconds % 3600) // 60) s = int(seconds % 60) cs = int((seconds % 1) * 100) return f"{h}:{m:02d}:{s:02d}.{cs:02d}" # ================================================================ # Step 8: Upload to YouTube # ================================================================ def _upload_youtube_sync(self,video_path: str,seo: dict) -> dict: """Upload video to YouTube using the API. OPTIMIZATION: Checks for duplicate uploads before uploading to save YouTube API quota (each upload costs ~1600 units,daily limit 10,000). Uses WARP SOCKS5 proxy if available for reliable YouTube access. """ try: from google.oauth2.credentials import Credentials from google.auth.transport.requests import Request from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload import httplib2 # No proxy needed (WARP VPN removed for HF compliance) _proxies = None tokens = self.state.get_youtube_tokens() if not tokens or "refresh_token" not in tokens: return {"status": "needs_auth","message": "YouTube not authorized yet. Visit the dashboard to connect."} # Use client_id/secret from stored tokens (more reliable than env vars) # The env vars may be truncated or incorrect,but the tokens store # the original credentials from the OAuth flow stored_client_id = tokens.get("client_id","") or os.getenv("GOOGLE_CLIENT_ID","") stored_client_secret = tokens.get("client_secret","") or os.getenv("GOOGLE_CLIENT_SECRET","") creds = Credentials( token=tokens.get("access_token",""), refresh_token=tokens["refresh_token"], token_uri="https://oauth2.googleapis.com/token", client_id=stored_client_id, client_secret=stored_client_secret, ) # Refresh if needed - use proxy for token refresh too if creds.expired: refresh_kwargs = {} if _proxies: refresh_kwargs["proxies"] = {"https": _proxies} creds.refresh(Request(**refresh_kwargs)) self.state.save_youtube_tokens({ "access_token": creds.token, "refresh_token": creds.refresh_token, "token_uri": creds.token_uri, "client_id": creds.client_id, "client_secret": creds.client_secret, }) # Build YouTube service with SOCKS5 proxy if available http = None if _proxies: # Configure httplib2 with SOCKS5 proxy for Google API try: import socks proxy_parts = _proxies.replace("socks5://","").replace("socks5h://","").split(":") proxy_host = proxy_parts[0] proxy_port = int(proxy_parts[1]) if len(proxy_parts) > 1 else 40000 http = httplib2.Http( proxy_info=httplib2.ProxyInfo( socks.PROXY_TYPE_SOCKS5, proxy_host, proxy_port ) ) print(f"[UPLOAD] Using SOCKS5 proxy: {proxy_host}:{proxy_port}") except ImportError: print("[UPLOAD] PySocks not available,uploading without proxy") except Exception as proxy_err: print(f"[UPLOAD] Proxy setup failed: {proxy_err},uploading directly") youtube = build("youtube","v3",credentials=creds,http=http) # OPTIMIZATION: Check for duplicate uploads before burning 1600 quota units # A search.list call costs only ~100 units vs 1600 for an upload upload_title = (seo.get("title","") or "Historia Increible")[:100] try: search_resp = youtube.search().list( part="snippet", forMine=True, type="video", q=upload_title, maxResults=3, ).execute() for item in search_resp.get("items",[]): if item["snippet"]["title"] == upload_title: print(f"[UPLOAD] DUPLICATE DETECTED: '{upload_title}' already uploaded (ID: {item['id']['videoId']})") return {"status": "duplicate","message": f"Video already exists: {item['id']['videoId']}","video_id": item["id"]["videoId"]} except Exception as dup_check_err: print(f"[UPLOAD] Duplicate check failed (non-fatal): {dup_check_err}") # Continue with upload even if check fails body = { "snippet": { "title": upload_title, "description": seo.get("description",""), "tags": ["historia","shorts","español","herstory","doblaje"], "categoryId": "22", }, "status": { "privacyStatus": "public", "selfDeclaredMadeForKids": False, }, } media = MediaFileUpload(video_path,mimetype="video/mp4",resumable=True) request = youtube.videos().insert(part="snippet,status",body=body,media_body=media) response = None while response is None: status,response = request.next_chunk() if status: print(f"[UPLOAD] Progress: {int(status.progress() * 100)}%") video_url = f"{_YT_SHORTS}{response['id']}" print(f"[UPLOAD] Success! {video_url}") return {"status": "success","video_id": response["id"],"url": video_url} except Exception as e: error_str = str(e) print(f"[UPLOAD] Error: {error_str}") # DETECT YouTube quota exceeded (daily limit 10,000 units, upload costs ~1600) # When quota is exceeded, YouTube returns 403 with quotaExceeded in the error quota_indicators = ["quotaExceeded", "quota", "exceeded", "dailyLimitExceeded", "rateLimitExceeded", "usageLimits", "403"] is_quota_error = any(ind in error_str.lower() for ind in [q.lower() for q in quota_indicators]) if is_quota_error: print(f"[UPLOAD] ⚠️ YouTube QUOTA EXCEEDED - daily limit reached") print(f"[UPLOAD] Pipeline will pause uploads and resume after quota reset (midnight PT)") return {"status": "quota_exceeded","error": error_str, "message": "YouTube daily quota exceeded. Will retry after reset."} # Try to save the video to HF Dataset so we can verify it later try: self._save_failed_upload(video_path,seo) except Exception as save_err: print(f"[UPLOAD] Save-to-dataset also failed: {save_err}") return {"status": "failed","error": error_str} def _save_failed_upload(self,video_path: str,seo: dict): """Save a video that failed to upload to HF Dataset for later verification.""" hf_token = os.getenv("HF_TOKEN","") if not hf_token or not Path(video_path).exists(): return try: import json repo_id = os.getenv("HF_DATASET_REPO","TomatitoToho/autodub-state") # Upload video file to the dataset from huggingface_hub import HfApi api = HfApi(token=hf_token) # Create a unique filename import time filename = f"failed_upload_{int(time.time())}.mp4" path_in_repo = f"failed_uploads/{filename}" api.upload_file( path_or_fileobj=video_path, path_in_repo=path_in_repo, repo_id=repo_id, repo_type="dataset", ) # Also upload metadata metadata = { "title": seo.get("title",""), "filename": filename, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), } meta_path = video_path.replace(".mp4","_meta.json") with open(meta_path,"w") as f: json.dump(metadata,f) api.upload_file( path_or_fileobj=meta_path, path_in_repo=f"failed_uploads/{filename}_meta.json", repo_id=repo_id, repo_type="dataset", ) Path(meta_path).unlink(missing_ok=True) print(f"[UPLOAD] Saved failed upload to HF Dataset: {path_in_repo}") except Exception as e: print(f"[UPLOAD] Save to dataset failed: {e}") # ================================================================ # Utility Methods # ================================================================ def _cleanup(self,work_dir: Path): """Clean up temporary files to free disk space.""" try: if work_dir.exists(): shutil.rmtree(work_dir) print(f"[CLEANUP] Removed: {work_dir}") except Exception as e: print(f"[CLEANUP] Error: {e}") # Also check total disk usage try: tmp = self.tmp_dir if tmp.exists(): total = sum(f.stat().st_size for f in tmp.rglob("*") if f.is_file()) print(f"[CLEANUP] Temp dir size: {total / 1024 / 1024:.1f} MB") except Exception: pass