from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import requests import re import os import subprocess import json import ssl ssl._create_default_https_context = ssl._create_unverified_context app = FastAPI(title="YouTube Converter API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) COOKIES_STR = os.environ.get("YOUTUBE_COOKIES", "") def setup_cookies(): if COOKIES_STR: try: youtube_cookies = [] for line in COOKIES_STR.split('\n'): if '.youtube.com' in line or line.startswith('#') or not line.strip(): youtube_cookies.append(line) with open('/tmp/cookies.txt', 'w') as f: f.write('\n'.join(youtube_cookies)) return '/tmp/cookies.txt' except Exception as e: print(f"Error writing cookies: {e}") return None COOKIE_FILE = setup_cookies() class VideoRequest(BaseModel): url: str format_id: str = None def extract_video_id(url): patterns = [ r'(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})', r'^([a-zA-Z0-9_-]{11})$' ] for pattern in patterns: match = re.search(pattern, url) if match: return match.group(1) return None def run_ytdlp(cmd, timeout=300): """Run yt-dlp with error handling""" try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return result except subprocess.TimeoutExpired: return None @app.get("/") async def root(): return { "status": "ok", "message": "YouTube Converter API is running", "cookies_loaded": bool(COOKIES_STR) } @app.post("/api/info") async def get_video_info(request: VideoRequest): try: video_id = extract_video_id(request.url) if not video_id: return {"success": False, "message": "Invalid YouTube URL"} cmd = [ 'yt-dlp', '--dump-json', '--no-download', '--no-warnings', '--no-check-certificate', '--user-agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' ] cmd.append(request.url) result = run_ytdlp(cmd, 120) if result is None: return {"success": False, "message": "Request timed out"} if result.returncode != 0: # Try with cookies if COOKIE_FILE and os.path.exists(COOKIE_FILE): cmd = [ 'yt-dlp', '--dump-json', '--no-download', '--no-warnings', '--no-check-certificate', '--cookies', COOKIE_FILE, '--user-agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' ] cmd.append(request.url) result = run_ytdlp(cmd, 120) if result is None or result.returncode != 0: return {"success": False, "message": f"Error: {result.stderr[:300] if result else 'Timeout'}"} try: info = json.loads(result.stdout.split('\n')[0]) except: info = json.loads(result.stdout) title = info.get('title', 'Unknown') thumbnail = info.get('thumbnail', f"https://i.ytimg.com/vi/{video_id}/maxresdefault.jpg") duration_sec = info.get('duration', 0) duration = f"{duration_sec // 60}:{duration_sec % 60:02d}" if duration_sec else "0:00" author = info.get('uploader', '') or info.get('channel', 'Unknown') formats = [] seen = set() all_formats = sorted( info.get('formats', []), key=lambda x: (x.get('height') or 0, x.get('fps') or 0), reverse=True ) for f in all_formats: height = f.get('height') vcodec = f.get('vcodec', 'none') or 'none' acodec = f.get('acodec', 'none') or 'none' if height and vcodec != 'none': res = f"{height}p" fps = f.get('fps', 0) or 0 if f.get('dynamic_range') == 'HDR': res += " HDR" if fps > 30: res = f"{height}p{fps}" if res in seen: continue seen.add(res) filesize = f.get('filesize') or f.get('filesize_approx', 0) size_str = f"{filesize/1024/1024:.1f}MB" if filesize else "Unknown" formats.append({ "format_id": str(f.get('format_id', '')), "resolution": res, "fps": fps, "vcodec": vcodec, "acodec": acodec, "filesize": size_str, "ext": f.get('ext', 'mp4'), "is_video_only": acodec == 'none', "is_audio_only": False }) formats.append({ "format_id": "bestaudio", "resolution": "Audio Only (MP3)", "fps": 0, "vcodec": "none", "acodec": "mp3", "filesize": "~5MB", "ext": "mp3", "is_video_only": False, "is_audio_only": True }) formats = formats[:15] return { "success": True, "message": "Video info extracted", "download_url": "", "video_info": { "title": title, "thumbnail": thumbnail, "duration": duration, "author": author, "formats": formats } } except Exception as e: return {"success": False, "message": f"Error: {str(e)}"} @app.post("/api/download/video") async def download_video(request: VideoRequest): try: video_id = extract_video_id(request.url) if not video_id: return {"success": False, "message": "Invalid YouTube URL"} import uuid file_id = str(uuid.uuid4()) output_template = f"/tmp/{file_id}.%(ext)s" cmd = [ 'yt-dlp', '-o', output_template, '--no-warnings', '--no-check-certificate', '--user-agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' ] format_id = request.format_id if format_id and format_id != "bestaudio": cmd.extend(['-f', f'{format_id}+bestaudio/best']) else: cmd.extend(['-f', 'bestvideo+bestaudio/best', '--merge-output-format', 'mp4']) cmd.append(request.url) result = run_ytdlp(cmd, 600) if result is None: return {"success": False, "message": "Download timed out"} if result.returncode != 0: return {"success": False, "message": f"Error: {result.stderr[:300]}"} file_path = None for f in os.listdir('/tmp'): if f.startswith(file_id): file_path = f"/tmp/{f}" break if file_path and os.path.exists(file_path): upload_url = upload_to_gofile(file_path) if os.path.exists(file_path): os.remove(file_path) if upload_url: return {"success": True, "message": "Video converted", "download_url": upload_url} return {"success": False, "message": "Download failed"} except Exception as e: return {"success": False, "message": f"Error: {str(e)}"} @app.post("/api/download/audio") async def download_audio(request: VideoRequest): try: video_id = extract_video_id(request.url) if not video_id: return {"success": False, "message": "Invalid YouTube URL"} import uuid file_id = str(uuid.uuid4()) output_template = f"/tmp/{file_id}.%(ext)s" cmd = [ 'yt-dlp', '-o', output_template, '-f', 'bestaudio/best', '--no-warnings', '--no-check-certificate', '--user-agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' ] cmd.append(request.url) result = run_ytdlp(cmd, 600) if result is None: return {"success": False, "message": "Download timed out"} if result.returncode != 0: return {"success": False, "message": f"Error: {result.stderr[:300]}"} file_path = None for f in os.listdir('/tmp'): if f.startswith(file_id): file_path = f"/tmp/{f}" break if file_path and os.path.exists(file_path): mp3_path = f"/tmp/{file_id}.mp3" ffmpeg_result = subprocess.run( ['ffmpeg', '-y', '-i', file_path, '-vn', '-acodec', 'libmp3lame', '-q:a', '2', mp3_path], capture_output=True, text=True, timeout=120 ) if os.path.exists(file_path): os.remove(file_path) if os.path.exists(mp3_path): upload_url = upload_to_gofile(mp3_path) if os.path.exists(mp3_path): os.remove(mp3_path) if upload_url: return {"success": True, "message": "Audio converted", "download_url": upload_url} return {"success": False, "message": "Audio conversion failed"} except Exception as e: return {"success": False, "message": f"Error: {str(e)}"} def upload_to_gofile(file_path): try: servers_resp = requests.get("https://api.gofile.io/servers", timeout=30) server_data = servers_resp.json() if server_data.get("status") != "ok": return None server = server_data["data"]["servers"][0]["name"] with open(file_path, 'rb') as f: upload_resp = requests.post( f"https://{server}.gofile.io/uploadFile", files={'file': f}, timeout=300 ) upload_data = upload_resp.json() if upload_data.get("status") == "ok": return upload_data["data"]["downloadPage"] return None except Exception as e: print(f"GoFile error: {e}") return None if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)