{error}
Back to dashboard") if not code: return HTMLResponse("{resp.text}
Back to dashboard") tokens = resp.json() # Save client_id and client_secret WITH the tokens so refresh works # even if env vars change or Space restarts without the same secrets tokens["client_id"] = GOOGLE_CLIENT_ID tokens["client_secret"] = GOOGLE_CLIENT_SECRET # Save tokens state.save_youtube_tokens(tokens) return HTMLResponse("""Your YouTube channel is now connected. The agent can upload videos.
Back to dashboard """) @app.get("/auth/status") async def auth_status(): """Check YouTube auth status.""" tokens = state.get_youtube_tokens() return { "connected": bool(tokens.get("refresh_token")), "has_access_token": bool(tokens.get("access_token")), } @app.post("/auth/save-tokens") async def save_tokens(tokens: dict): """Save YouTube OAuth tokens directly (admin endpoint).""" if not tokens.get("refresh_token"): return {"status": "error", "message": "Missing refresh_token"} state.save_youtube_tokens(tokens) return {"status": "ok", "message": "Tokens saved", "connected": True} @app.get("/auth/export-tokens") async def export_tokens(): """Export YouTube tokens for backup (admin only).""" tokens = state.get_youtube_tokens() if not tokens.get("refresh_token"): return {"status": "error", "message": "No tokens to export"} return {"status": "ok", "tokens": tokens} # ============================================================ # Routes - YouTube Cookies # ============================================================ @app.get("/cookies/status") async def cookies_status(): """Check if YouTube cookies are available.""" # Check local file (the primary method for yt-dlp) local_cookies = COOKIES_TXT.exists() # Check dataset dataset_cookies = bool(state.load_cookies_txt()) browser_cookies = False # browser not available has_any = local_cookies or dataset_cookies or browser_cookies return { "browser_ready": False, "has_cookies": has_any, "local_cookies": local_cookies, "dataset_cookies": dataset_cookies, "browser_cookies": browser_cookies, } @app.post("/cookies/upload") async def cookies_upload(request: Request): """Upload YouTube cookies in Netscape format (cookies.txt). User exports cookies from their browser using an extension like 'Get cookies.txt LOCALLY' for the youtube.com domain. Saves to HF Dataset for persistence, and to local file for yt-dlp. """ body = await request.body() cookies_text = body.decode("utf-8") # Validate it looks like Netscape cookies if not cookies_text.strip() or ("# Netscape" not in cookies_text and "\t" not in cookies_text): return {"status": "error", "message": "Invalid cookies format. Must be Netscape format."} # Save to HF Dataset (persistent) state.save_cookies_txt(cookies_text) # Save to local file for yt-dlp (with fallback to /tmp) try: BROWSER_DATA_DIR.mkdir(parents=True, exist_ok=True) COOKIES_TXT.write_text(cookies_text) except PermissionError: BROWSER_DATA_DIR = Path("/tmp/browser_data") COOKIES_TXT = BROWSER_DATA_DIR / "cookies.txt" BROWSER_DATA_DIR.mkdir(parents=True, exist_ok=True) COOKIES_TXT.write_text(cookies_text) # Browser no longer available (removed for HF compliance) has_cookies = any( line.strip() and not line.startswith("#") and "youtube.com" in line.lower() for line in cookies_text.split("\n") ) return { "status": "success", "message": "Cookies guardadas!" if has_cookies else "Cookies guardadas pero pueden necesitar verificación.", "cookies_valid": has_cookies, } @app.post("/cookies/refresh") async def cookies_refresh(): """Refresh cookies by visiting YouTube.""" if True: return {"status": "error", "message": "Browser not available (removed for HF Spaces compliance)"} success = False # browser not available has_cookies = False # browser not available return { "status": "success" if success else "failed", "has_cookies": has_cookies, } # ============================================================ # Routes - Manual Controls # ============================================================ @app.post("/trigger/check") async def trigger_check(): """Manually trigger a channel check.""" if pipeline.is_processing(): return {"status": "busy", "message": "Pipeline is currently processing a video"} # Run check in background asyncio.create_task(_check_and_process()) return {"status": "started", "message": "Channel check started"} @app.post("/trigger/process/{video_id}") async def trigger_process(video_id: str, url: str = "", title: str = ""): """Manually trigger processing for a specific video.""" if pipeline.is_processing(): return {"status": "busy", "message": "Pipeline is currently processing"} if not url: url = f"{_YT_SHORTS}{video_id}" # Try to get the real title from YouTube if not provided if not title: try: import httpx async with httpx.AsyncClient(timeout=10) as client: # Try YouTube oEmbed API to get the title resp = await client.get(f"{_YT_OEMBED}{_YT_SHORTS}{video_id}&format=json") if resp.status_code == 200: data = resp.json() title = data.get("title", "") except Exception: pass if not title: title = f"Short {video_id}" async def _run_pipeline(): log_entry = { "video_id": video_id, "title": title, "started_at": datetime.now().isoformat(), "status": "processing", } processing_log.append(log_entry) result = await pipeline.process_video(video_id, title, url) log_entry["status"] = result.get("status", "unknown") log_entry["completed_at"] = datetime.now().isoformat() log_entry["result"] = result # Update last video if successful if result.get("status") == "success": _update_last_video(result, {"video_id": video_id, "title": title, "url": url}) asyncio.create_task(_run_pipeline()) return {"status": "started", "video_id": video_id} @app.post("/trigger/process-queue") async def trigger_process_queue(): """Process the next item in the queue.""" if pipeline.is_processing(): return {"status": "busy", "message": "Pipeline is currently processing"} next_item = state.get_next_in_queue() if not next_item: return {"status": "empty", "message": "No items in queue"} video_id = next_item["video_id"] title = next_item.get("title", "Queued video") url = next_item.get("url", f"{_YT_SHORTS}{video_id}") # Mark as processing state.update_queue_item(video_id, "processing") async def _run_pipeline(): log_entry = { "video_id": video_id, "title": title, "started_at": datetime.now().isoformat(), "status": "processing", } processing_log.append(log_entry) result = await pipeline.process_video(video_id, title, url) status = result.get("status", "unknown") log_entry["status"] = status log_entry["completed_at"] = datetime.now().isoformat() log_entry["result"] = result if status == "success": state.update_queue_item(video_id, "done") _update_last_video(result, {"video_id": video_id, "title": title, "url": url}) else: state.update_queue_item(video_id, "failed", error=result.get("error", "Unknown error")) asyncio.create_task(_run_pipeline()) return {"status": "started", "video_id": video_id, "title": title} @app.post("/queue/clear") async def clear_queue(): """Clear the entire queue.""" state._state["queue"] = [] state.save() return {"status": "cleared"} @app.post("/brain/load") async def load_brain(): """Manually trigger brain loading.""" if brain.is_ready(): return {"status": "already_loaded"} asyncio.create_task(_load_brain_background()) return {"status": "loading_started"} @app.get("/queue") async def get_queue(): """Get the current processing queue.""" queue = state._state.get("queue", []) pending = [q for q in queue if q.get("status") == "pending"] retryable = state.get_retryable_failed_videos(max_retries=3) return { "queue": queue, "total": len(queue), "pending": len(pending), "retryable_failed": len(retryable), "processing": pipeline.is_processing(), } @app.get("/failed-videos") async def get_failed_videos(): """Get all failed videos with their error details and retry counts.""" failed = state._state.get("failed_videos", {}) retryable = state.get_retryable_failed_videos(max_retries=3) permanent = {vid: info for vid, info in failed.items() if info.get("retry_count", 0) >= 3} return { "total_failed": len(failed), "retryable": retryable, "retryable_count": len(retryable), "permanent": permanent, "permanent_count": len(permanent), } @app.post("/retry-failed/{video_id}") async def retry_failed_video(video_id: str): """Manually retry a specific failed video.""" if pipeline.is_processing(): return {"status": "busy", "message": "Pipeline is currently processing"} failed = state._state.get("failed_videos", {}).get(video_id) if not failed: return {"status": "error", "message": f"Video {video_id} not found in failed list"} title = failed.get("title", "Retry") url = f"{_YT_SHORTS}{video_id}" async def _run_pipeline(): log_entry = { "video_id": video_id, "title": title, "started_at": datetime.now().isoformat(), "status": "processing", } processing_log.append(log_entry) result = await pipeline.process_video(video_id, title, url) log_entry["status"] = result.get("status", "unknown") log_entry["completed_at"] = datetime.now().isoformat() log_entry["result"] = result if result.get("status") == "success": _update_last_video(result, {"video_id": video_id, "title": title, "url": url}) asyncio.create_task(_run_pipeline()) return {"status": "started", "video_id": video_id, "attempt": failed.get("retry_count", 0) + 1} @app.post("/retry-all-failed") async def retry_all_failed(): """Add all retryable failed videos back to the processing queue.""" retryable = state.get_retryable_failed_videos(max_retries=3) if not retryable: return {"status": "empty", "message": "No retryable failed videos"} added = 0 for video in retryable: # Add to queue if not already there already_queued = any( q.get("video_id") == video["video_id"] for q in state._state.get("queue", []) ) if not already_queued: state.add_to_queue( video_id=video["video_id"], title=video["title"], url=f"{_YT_SHORTS}{video['video_id']}", ) added += 1 return { "status": "success", "added_to_queue": added, "total_retryable": len(retryable), "message": f"Added {added} videos to queue for retry", } @app.get("/history") async def get_history(): """Get processing history.""" return { "processed": state._state.get("processed_videos", {}), "log": processing_log[-20:], "stats": state.get_stats(), "last_video": last_video, } @app.post("/config") async def update_config(check_interval: Optional[int] = None): """Update configuration.""" if check_interval and check_interval >= 5: # Reschedule the monitoring job scheduler.reschedule_job( "channel_monitor", trigger="interval", minutes=check_interval, ) state.update_config({"check_interval_minutes": check_interval}) return {"status": "updated", "check_interval": check_interval} return {"status": "no_changes"} @app.post("/config/visual") async def update_visual_config(request: Request): """Update visual config (blur + subtitle positions) from dashboard. These positions are saved in the HF Dataset and persist across restarts. The pipeline reads them dynamically when composing videos. """ try: config = await request.json() # Validate ranges (720x1280 canvas) if "blur_y" in config: config["blur_y"] = max(0, min(1280, int(config["blur_y"]))) if "blur_h" in config: config["blur_h"] = max(10, min(400, int(config["blur_h"]))) if "blur_x" in config: config["blur_x"] = max(0, min(360, int(config["blur_x"]))) if "blur_w" in config: config["blur_w"] = max(50, min(720, int(config["blur_w"]))) if "blur_strength" in config: config["blur_strength"] = max(2, min(50, int(config["blur_strength"]))) if "sub_font_size" in config: config["sub_font_size"] = max(20, min(150, int(config["sub_font_size"]))) if "sub_margin_v" in config: config["sub_margin_v"] = max(0, min(500, int(config["sub_margin_v"]))) if "sub_margin_l" in config: config["sub_margin_l"] = max(0, min(200, int(config["sub_margin_l"]))) if "sub_margin_r" in config: config["sub_margin_r"] = max(0, min(200, int(config["sub_margin_r"]))) state.update_visual_config(config) print(f"[APP] Visual config updated: {config}") return {"status": "ok", "config": state.get_visual_config()} except Exception as e: return {"status": "error", "message": str(e)} @app.get("/config/visual") async def get_visual_config(): """Get current visual config.""" return state.get_visual_config() @app.post("/trigger/backfill") async def trigger_backfill(count: int = 100): """Fetch older videos from the source channel and add them to the processing queue. This creates a backlog of videos to process, starting from the oldest available. Videos are added in reverse chronological order (newest first) so the queue processes from most recent backward. """ try: # Fetch videos from the source channel videos = monitor.fetch_channel_videos(max_results=min(count, 200)) if not videos: return {"status": "empty", "message": "No videos found on the channel"} # Filter out already-processed videos new_videos = [] for v in videos: if not state.is_video_processed(v["video_id"]): # Check not already in queue already_queued = any( q.get("video_id") == v["video_id"] for q in state._state.get("queue", []) ) if not already_queued: new_videos.append(v) if not new_videos: return {"status": "empty", "message": "All videos already processed or queued"} # Add to queue (newest first = they'll be processed in order) for v in new_videos: state.add_to_queue( video_id=v["video_id"], title=v["title"], url=v["url"], ) pending = [q for q in state._state.get("queue", []) if q.get("status") == "pending"] return { "status": "success", "added": len(new_videos), "total_in_queue": len(pending), "message": f"Added {len(new_videos)} videos to queue. {len(pending)} pending." } except Exception as e: return {"status": "error", "message": str(e)} @app.post("/trigger/delete-channel-videos") async def delete_channel_videos(): """Delete all videos from the connected YouTube channel (La Historia de Ella). Uses the stored OAuth tokens to authenticate. This is a destructive operation. OPTIMIZED: Uses playlistItems.list (3 units/page) instead of search.list (100 units/page) to save API quota. Can delete ~195 videos per day within the 10,000 unit quota. """ try: from google.oauth2.credentials import Credentials from google.auth.transport.requests import Request from googleapiclient.discovery import build tokens = state.get_youtube_tokens() if not tokens or "refresh_token" not in tokens: return {"status": "error", "message": "YouTube not authorized. Connect your account first."} # Use client_id/secret from stored tokens (more reliable than env vars) stored_client_id = tokens.get("client_id", "") or GOOGLE_CLIENT_ID stored_client_secret = tokens.get("client_secret", "") or 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, ) if creds.expired: creds.refresh(Request()) 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, }) youtube = build("youtube", "v3", credentials=creds) # Get the authenticated user's channel (OUR channel, not the source) # This ensures we delete from "La Historia de Ella", not "Her Story" channel_resp = youtube.channels().list(part="contentDetails", mine=True).execute() if not channel_resp.get("items"): return {"status": "error", "message": "Could not find your YouTube channel."} uploads_playlist = channel_resp["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"] channel_name = channel_resp["items"][0].get("snippet", {}).get("title", "Unknown") print(f"[DELETE] Deleting videos from: {channel_name} (uploads: {uploads_playlist})") # Use playlistItems.list (3 units/page) instead of search.list (100 units/page) # This saves 97 units per page = enough for ~2 more video deletes per page deleted = 0 failed = 0 page_token = "" video_ids_to_delete = [] # First, collect all video IDs while True: try: pl_response = youtube.playlistItems().list( part="snippet", playlistId=uploads_playlist, maxResults=50, pageToken=page_token, ).execute() except Exception as e: if "quotaExceeded" in str(e): return {"status": "error", "message": "YouTube API quota exceeded. Try again tomorrow."} raise for item in pl_response.get("items", []): video_id = item["snippet"]["resourceId"].get("videoId", "") if video_id: video_ids_to_delete.append(video_id) page_token = pl_response.get("nextPageToken", "") if not page_token: break print(f"[DELETE] Found {len(video_ids_to_delete)} videos to delete") # Now delete each video for video_id in video_ids_to_delete: try: youtube.videos().delete(id=video_id).execute() deleted += 1 if deleted % 10 == 0: print(f"[DELETE] Progress: {deleted}/{len(video_ids_to_delete)} deleted") except Exception as e: failed += 1 error_str = str(e) print(f"[DELETE] Failed: {video_id} - {error_str[:100]}") if "quotaExceeded" in error_str: return { "status": "partial", "deleted": deleted, "failed": failed, "total_found": len(video_ids_to_delete), "message": f"YouTube API quota exceeded. Deleted {deleted}/{len(video_ids_to_delete)} videos. Try again tomorrow for the rest." } return { "status": "complete", "deleted": deleted, "failed": failed, "message": f"Deleted {deleted} videos from channel." } except Exception as e: error_str = str(e) if "quotaExceeded" in error_str: return {"status": "error", "message": "YouTube API quota exceeded. Try again tomorrow (resets at midnight Pacific Time)."} return {"status": "error", "message": error_str} @app.post("/voice/upload-reference") async def upload_voice_reference(request: Request): """Upload a reference audio file for voice cloning (Chatterbox TTS). The reference audio determines the voice that Chatterbox will clone. Upload a short clip (5-15 seconds) of the desired voice. Supported formats: WAV, MP3, FLAC. """ body = await request.body() if not body or len(body) < 100: return {"status": "error", "message": "Audio file too small or empty"} if len(body) > 5 * 1024 * 1024: # 5MB max return {"status": "error", "message": "Audio file too large (max 5MB)"} # Save the reference audio ref_dir = Path(tempfile.gettempdir()) / "autodub_voices" ref_dir.mkdir(parents=True, exist_ok=True) ref_path = str(ref_dir / "custom_ref.wav") # Write the uploaded file with open(ref_path + ".tmp", "wb") as f: f.write(body) # Convert to WAV for Chatterbox compatibility try: proc = subprocess.run([ "ffmpeg", "-y", "-i", ref_path + ".tmp", "-ar", "24000", "-ac", "1", "-sample_fmt", "s16", ref_path, ], capture_output=True, timeout=15) if proc.returncode != 0: # Maybe it's already WAV, just rename import shutil shutil.move(ref_path + ".tmp", ref_path) except Exception: import shutil shutil.move(ref_path + ".tmp", ref_path) # Clean up temp file try: Path(ref_path + ".tmp").unlink() except: pass # Save path in visual config so pipeline uses it state.update_visual_config({ "voice_reference_path": ref_path, }) return { "status": "success", "message": "Voz de referencia guardada!", "path": ref_path, } @app.post("/voice/test") async def test_voice(text: str = "Hola, esta es una prueba de voz."): """Generate a test audio clip with the current voice settings.""" try: import tempfile test_dir = Path(tempfile.gettempdir()) / "autodub_test" test_dir.mkdir(parents=True, exist_ok=True) test_path = str(test_dir / "test_voice.mp3") # Use Supertonic 3 TTS for voice testing try: from pipeline import _get_supertonic_model, SUPERTONIC_AVAILABLE, SUPERTONIC_VOICE, SUPERTONIC_LANG if SUPERTONIC_AVAILABLE: model = _get_supertonic_model() if model is not None: style = model.get_voice_style(voice_name=SUPERTONIC_VOICE) wav, duration = model.synthesize(text, voice_style=style, lang=SUPERTONIC_LANG) model.save_audio(wav, test_path) if Path(test_path).exists(): return {"status": "success", "path": test_path, "voice": f"Supertonic 3 ({SUPERTONIC_VOICE})"} except Exception as se: print(f"[VOICE-TEST] Supertonic failed: {se}") # Fallback to gTTS try: from gtts import gTTS tts = gTTS(text=text, lang='es', slow=False) tts.save(test_path) if Path(test_path).exists(): return {"status": "success", "path": test_path, "voice": "gTTS (fallback)"} except Exception as ge: print(f"[VOICE-TEST] gTTS failed: {ge}") return {"status": "error", "message": "All TTS methods failed"} except Exception as e: return {"status": "error", "message": str(e)} @app.post("/trigger/process-uploaded") async def process_uploaded_video( video_id: str = "", title: str = "Uploaded video", original_url: str = "", ): """Process a video that has been manually uploaded to the state dataset. This is a fallback for when automatic YouTube download fails. The user can upload a video file to the state dataset, then trigger processing with this endpoint. """ if pipeline.is_processing(): return {"status": "busy", "message": "Pipeline is currently processing"} if not video_id: return {"status": "error", "message": "video_id is required"} # Check if video file exists in the dataset import tempfile try: from huggingface_hub import hf_hub_download video_path = hf_hub_download( repo_id=state.repo_id, filename=f"uploads/{video_id}.mp4", repo_type="dataset", token=state.token, ) except Exception: return {"status": "error", "message": f"Video file not found in dataset: uploads/{video_id}.mp4"} async def _run_pipeline(): log_entry = { "video_id": video_id, "title": title, "started_at": datetime.now().isoformat(), "status": "processing", } processing_log.append(log_entry) # Process with the uploaded video path result = await pipeline.process_video_from_file( video_id=video_id, title=title, video_path=video_path, original_url=original_url or f"{_YT_SHORTS}{video_id}", ) log_entry["status"] = result.get("status", "unknown") log_entry["completed_at"] = datetime.now().isoformat() log_entry["result"] = result if result.get("status") == "success": _update_last_video(result, {"video_id": video_id, "title": title, "url": original_url}) asyncio.create_task(_run_pipeline()) return {"status": "started", "video_id": video_id} @app.get("/test-download") async def test_download(video_id: str = "bD6N9ekGCZs"): """Quick test to see which download methods work from this Space.""" results = {} # Test Invidious instances results["invidious"] = {} instances = _os_url_const.getenv("INVIDIOUS_INSTANCES", "").split(",") if _os_url_const.getenv("INVIDIOUS_INSTANCES") else [] for inst in instances: try: async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client: resp = await client.get(f"{inst}/api/v1/videos/{video_id}") if resp.status_code == 200: data = resp.json() formats = len(data.get("formatStreams", [])) results["invidious"][inst] = f"OK ({formats} formats)" else: results["invidious"][inst] = f"HTTP {resp.status_code}" except Exception as e: results["invidious"][inst] = f"{type(e).__name__}: {str(e)[:60]}" # Test Piped instances results["piped"] = {} piped_instances = _os_url_const.getenv("PIPED_INSTANCES", "").split(",") if _os_url_const.getenv("PIPED_INSTANCES") else [] for inst in piped_instances: try: async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client: resp = await client.get(f"{inst}/streams/{video_id}") if resp.status_code == 200: data = resp.json() streams = len(data.get("videoStreams", [])) results["piped"][inst] = f"OK ({streams} streams)" else: results["piped"][inst] = f"HTTP {resp.status_code}" except Exception as e: results["piped"][inst] = f"{type(e).__name__}: {str(e)[:60]}" # Test custom proxy (Cloudflare Worker) if YTDLP_PROXY: proxy_secret = os.getenv("YTDLP_PROXY_SECRET", "") proxy_headers = {} if proxy_secret: proxy_headers["Authorization"] = f"Bearer {proxy_secret}" results["proxy"] = {} # Test Worker health (root endpoint) try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get(f"{YTDLP_PROXY}/", headers=proxy_headers) if resp.status_code == 200: data = resp.json() results["proxy"]["health"] = f"OK: v{data.get('version', '?')}, methods={data.get('methods', [])}" else: results["proxy"]["health"] = f"HTTP {resp.status_code}: {resp.text[:200]}" except Exception as e: results["proxy"]["health"] = f"{type(e).__name__}: {str(e)[:100]}" # Test Worker download endpoint try: async with httpx.AsyncClient(timeout=30) as client: download_url = f"{YTDLP_PROXY}/download/{video_id}" resp = await client.get(download_url, headers=proxy_headers) if resp.status_code == 200: data = resp.json() results["proxy"]["download"] = f"OK: status={data.get('status')}, method={data.get('method')}, quality={data.get('quality')}, hasAudio={data.get('hasAudio')}" else: results["proxy"]["download"] = f"HTTP {resp.status_code}: {resp.text[:200]}" except Exception as e: results["proxy"]["download"] = f"{type(e).__name__}: {str(e)[:100]}" return results @app.get("/debug/download/{video_id}") async def debug_download(video_id: str): """Debug endpoint que ejecuta el método _download_cloudflare_worker directamente y devuelve el resultado detallado.""" import asyncio import tempfile from pathlib import Path result = {"video_id": video_id, "steps": []} try: # Step 1: Verify YTDLP_PROXY is set proxy = os.getenv("YTDLP_PROXY", "") result["steps"].append({"step": "env_check", "YTDLP_PROXY": proxy[:50] if proxy else "EMPTY"}) if not proxy: result["error"] = "YTDLP_PROXY not configured" return result # Step 2: Try the info endpoint directly (like /test-download does) import httpx proxy_secret = os.getenv("YTDLP_PROXY_SECRET", "") headers = {} if proxy_secret: headers["Authorization"] = f"Bearer {proxy_secret}" async with httpx.AsyncClient(timeout=60) as client: info_url = f"{proxy}/download/{video_id}" resp = await client.get(info_url, headers=headers) result["steps"].append({ "step": "info_request", "url": info_url, "status_code": resp.status_code, "response_size": len(resp.text) }) if resp.status_code != 200: result["error"] = f"HTTP {resp.status_code}" result["response_preview"] = resp.text[:300] return result data = resp.json() result["steps"].append({ "step": "info_parsed", "status": data.get("status"), "method": data.get("method"), "quality": data.get("quality"), "hasAudio": data.get("hasAudio"), "url_length": len(data.get("url", "")), "audioUrl_length": len(data.get("audioUrl", "")) }) if data.get("status") != "success": result["error"] = data.get("error", "unknown") return result cdn_url = data.get("url", "") if not cdn_url: result["error"] = "No CDN URL in response" return result # Step 3: Try to download from CDN URL directly result["steps"].append({"step": "downloading_from_cdn", "url_length": len(cdn_url)}) # Create temp file import tempfile as _tempfile tmp_fd, tmp_path = _tempfile.mkstemp(suffix=".mp4") import os as _os_mod _os_mod.close(tmp_fd) try: async with httpx.AsyncClient(timeout=90, follow_redirects=True) as client: dl_resp = await client.get(cdn_url) result["steps"].append({ "step": "cdn_download_response", "status_code": dl_resp.status_code, "content_length": len(dl_resp.content), "content_type": dl_resp.headers.get("content-type", "?") }) if dl_resp.status_code == 200 and len(dl_resp.content) > 10000: with open(tmp_path, 'wb') as f: f.write(dl_resp.content) result["success"] = True result["file_path"] = tmp_path result["file_size"] = len(dl_resp.content) else: result["error"] = f"CDN download failed: HTTP {dl_resp.status_code}, size {len(dl_resp.content)}" except Exception as e: result["error"] = f"CDN download exception: {type(e).__name__}: {e}" except Exception as e: result["error"] = f"Unexpected: {type(e).__name__}: {e}" return result @app.get("/debug/cf-downloader/{video_id}") async def debug_cf_downloader(video_id: str): """Debug endpoint específico para cf-yt-downloader worker.""" import httpx as _httpx import os as _os result = {"video_id": video_id, "steps": []} downloader_url = _os.getenv("CF_YT_DOWNLOADER_URL", "https://cf-yt-downloader.t70512145.workers.dev") downloader_secret = _os.getenv("YTDLP_PROXY_SECRET", "") result["steps"].append({ "step": "env_check", "CF_YT_DOWNLOADER_URL": downloader_url, "has_secret": bool(downloader_secret) }) headers = {} if downloader_secret: headers["Authorization"] = f"Bearer {downloader_secret}" try: # Hacer request al cf-yt-downloader result["steps"].append({"step": "starting_request", "url": f"{downloader_url}/video/{video_id}"}) async with _httpx.AsyncClient(timeout=120, follow_redirects=True) as client: resp = await client.get(f"{downloader_url}/video/{video_id}", headers=headers) result["steps"].append({ "step": "response_received", "status_code": resp.status_code, "content_length": len(resp.content), "content_type": resp.headers.get("content-type", "?"), "x_video_quality": resp.headers.get("x-video-quality", "?"), "x_video_instance": resp.headers.get("x-video-instance", "?") }) if resp.status_code in (200, 206) and len(resp.content) > 10000: # Guardar el video import tempfile as _tempfile tmp_fd, tmp_path = _tempfile.mkstemp(suffix=".mp4") import os as _os_mod _os_mod.close(tmp_fd) with open(tmp_path, 'wb') as f: f.write(resp.content) result["success"] = True result["file_path"] = tmp_path result["file_size"] = len(resp.content) result["steps"].append({"step": "saved_to_file", "path": tmp_path}) else: result["error"] = f"Download failed: HTTP {resp.status_code}, size {len(resp.content)}" result["response_preview"] = resp.text[:500] except Exception as e: result["error"] = f"Exception: {type(e).__name__}: {e}" return result @app.get("/debug/test-download-method/{video_id}") async def debug_test_download_method(video_id: str): """Ejecuta SOLO el método _download_cloudflare_worker del pipeline.""" import tempfile as _tempfile from pathlib import Path as _Path result = {"video_id": video_id, "steps": []} try: # Crear work_dir temporal work_dir = _Path(_tempfile.mkdtemp()) output_path = str(work_dir / "original.mp4") result["steps"].append({"step": "created_workdir", "path": str(work_dir)}) # Obtener el pipeline if pipeline is None: result["error"] = "Pipeline not initialized" return result result["steps"].append({"step": "got_pipeline", "type": type(pipeline).__name__}) # Llamar al método directamente con captura de stdout result["steps"].append({"step": "calling_method"}) # Capturar prints del método import io as _io import contextlib as _contextlib captured = _io.StringIO() try: with _contextlib.redirect_stdout(captured): download_result = await pipeline._download_cloudflare_worker( video_id, output_path, "" ) result["captured_stdout"] = captured.getvalue()[-2000:] # Últimos 2000 chars except Exception as e: result["captured_stdout"] = captured.getvalue()[-2000:] raise e result["steps"].append({"step": "method_returned", "result": str(download_result)}) if download_result: import os as _os size = _os.path.getsize(download_result) result["success"] = True result["file_size"] = size result["file_path"] = download_result else: result["error"] = "Method returned None" except Exception as e: result["error"] = f"Exception: {type(e).__name__}: {e}" import traceback result["traceback"] = traceback.format_exc() return result @app.get("/debug/full-process/{video_id}") async def debug_full_process(video_id: str): """Ejecuta process_video completo con captura de stdout.""" import io as _io import contextlib as _contextlib result = {"video_id": video_id} try: if pipeline.is_processing(): return {"error": "Pipeline busy"} captured = _io.StringIO() url = f"{_YT_SHORTS}{video_id}" title = f"Short {video_id}" with _contextlib.redirect_stdout(captured): process_result = await pipeline.process_video(video_id, title, url) result["success"] = process_result.get("status") == "success" result["status"] = process_result.get("status") result["failed_stage"] = process_result.get("failed_stage") result["captured_stdout"] = captured.getvalue()[-3000:] # Últimos 3000 chars result["process_result"] = str(process_result)[:1000] except Exception as e: result["error"] = f"Exception: {type(e).__name__}: {e}" import traceback result["traceback"] = traceback.format_exc()[-1000:] result["captured_stdout"] = captured.getvalue()[-3000:] if 'captured' in dir() else "(no capture)" return result @app.get("/debug/innertube-direct/{video_id}") async def debug_innertube_direct(video_id: str): """Test Innertube direct download method.""" import tempfile as _tempfile from pathlib import Path as _Path import io as _io import contextlib as _contextlib work_dir = _Path(_tempfile.mkdtemp()) output_path = str(work_dir / "original.mp4") result = {"video_id": video_id} captured = _io.StringIO() try: with _contextlib.redirect_stdout(captured): download_result = await pipeline._innertube_direct(video_id, output_path) result["captured_stdout"] = captured.getvalue() result["result"] = str(download_result) if download_result: import os as _os result["file_size"] = _os.path.getsize(download_result) result["success"] = True except Exception as e: result["captured_stdout"] = captured.getvalue() result["error"] = str(e) return result # ============================================================ # Run # ============================================================ if __name__ == "__main__": import uvicorn port = int(os.getenv("PORT", 7860)) print(f"[APP] Starting on port {port}") uvicorn.run(app, host="0.0.0.0", port=port) # Force rebuild Sat Jul 4 22:28:11 UTC 2026