import os import logging import subprocess import tempfile import uuid import shutil import time import random import re import json import requests from bs4 import BeautifulSoup from urllib.parse import urlparse, parse_qs class VideoDownloader: def __init__(self, download_path='downloads', public_path='/var/www/html/downloads', public_url='http://8.215.44.123/downloads'): self.download_path = download_path self.public_path = public_path self.public_url = public_url # Ensure download directory exists os.makedirs(download_path, exist_ok=True) # Configure logging logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) self.logger = logging.getLogger(__name__) # User agents for rotation self.user_agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.0 Safari/605.1.15', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:90.0) Gecko/20100101 Firefox/90.0', 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1' ] def get_platform(self, url): """Determine the platform from the URL""" if 'youtube.com' in url or 'youtu.be' in url: return 'youtube' elif 'facebook.com' in url or 'fb.com' in url or 'fb.watch' in url: return 'facebook' elif 'instagram.com' in url: return 'instagram' elif 'tiktok.com' in url: return 'tiktok' else: return 'generic' def get_random_user_agent(self): """Get a random user agent from the list""" return random.choice(self.user_agents) def get_youtube_video_id(self, url): """Extract YouTube video ID from URL""" if 'youtu.be' in url: # Short URL format: https://youtu.be/VIDEO_ID path = urlparse(url).path return path.strip('/').split('?')[0] else: # Regular URL format: https://www.youtube.com/watch?v=VIDEO_ID parsed_url = urlparse(url) query_params = parse_qs(parsed_url.query) return query_params.get('v', [''])[0] def get_instagram_shortcode(self, url): """Extract Instagram shortcode from URL""" # URL format: https://www.instagram.com/p/SHORTCODE/ match = re.search(r'instagram\.com/(?:p|reel)/([^/?]+)', url) if match: return match.group(1) return None def compress_video(self, input_file, target_size_mb=10, output_file=None): """Compress video to target size in MB""" try: self.logger.info(f"Compressing video: {input_file} to target size: {target_size_mb}MB") # Create output file if not provided if output_file is None: file_name, file_ext = os.path.splitext(input_file) output_file = f"{file_name}_compressed{file_ext}" # Check if ffmpeg is installed try: subprocess.run(['ffmpeg', '-version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True) except (subprocess.SubprocessError, FileNotFoundError): self.logger.error("ffmpeg not found. Please install ffmpeg.") return input_file # Get video duration cmd = ['ffprobe', '-v', 'error', '-show_entries', 'format=duration', '-of', 'default=noprint_wrappers=1:nokey=1', input_file] result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) if result.returncode != 0: self.logger.error(f"Error getting video duration: {result.stderr}") return input_file duration = float(result.stdout.strip()) # Calculate target bitrate (80% for video, 20% for audio) target_size_bits = target_size_mb * 8 * 1024 * 1024 bitrate = int(target_size_bits / (duration * 1.2)) # 1.2 factor for overhead # Compress video cmd = [ 'ffmpeg', '-i', input_file, '-b:v', str(bitrate), '-maxrate', str(bitrate), '-bufsize', str(bitrate*2), '-c:a', 'aac', '-b:a', '128k', # Compress audio too '-y', output_file ] self.logger.info(f"Running ffmpeg command: {' '.join(cmd)}") result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode != 0: self.logger.error(f"Error compressing video: {result.stderr}") return input_file # Verify the compressed file exists and is smaller if os.path.exists(output_file): original_size = os.path.getsize(input_file) / (1024 * 1024) # MB compressed_size = os.path.getsize(output_file) / (1024 * 1024) # MB self.logger.info(f"Original size: {original_size:.2f}MB, Compressed size: {compressed_size:.2f}MB") if compressed_size < original_size: return output_file else: self.logger.warning("Compressed file is not smaller than original. Using original file.") os.remove(output_file) return input_file else: self.logger.error("Compressed file was not created") return input_file except Exception as e: self.logger.error(f"Error in compress_video: {str(e)}") return input_file # Return original file if compression fails def create_public_download_link(self, file_path, title, expiry_hours=24): """Create a public download link for a file""" try: # Ensure public directory exists os.makedirs(self.public_path, exist_ok=True) # Generate a unique filename to avoid collisions file_name, file_ext = os.path.splitext(os.path.basename(file_path)) unique_id = str(uuid.uuid4())[:8] safe_title = ''.join(c if c.isalnum() or c in ['-', '_'] else '_' for c in title) public_filename = f"{safe_title}_{unique_id}{file_ext}" public_file_path = os.path.join(self.public_path, public_filename) # Copy the file to the public directory shutil.copy2(file_path, public_file_path) # Generate the public URL download_url = f"{self.public_url}/{public_filename}" # Create a .info file with metadata info_file_path = f"{public_file_path}.info" with open(info_file_path, 'w') as f: f.write(f"Title: {title}\n") f.write(f"Original file: {os.path.basename(file_path)}\n") f.write(f"Created: {subprocess.check_output(['date']).decode('utf-8')}") f.write(f"Expires: After {expiry_hours} hours\n") # Schedule file deletion (if possible) try: # Use at command to schedule deletion delete_cmd = f"rm -f {public_file_path} {info_file_path}" at_time = f"now + {expiry_hours} hours" subprocess.run(f'echo "{delete_cmd}" | at {at_time}', shell=True) self.logger.info(f"Scheduled deletion of {public_file_path} after {expiry_hours} hours") except Exception as e: self.logger.warning(f"Could not schedule file deletion: {e}") return { 'success': True, 'url': download_url, 'file_path': public_file_path, 'expires_in_hours': expiry_hours } except Exception as e: self.logger.error(f"Error creating download link: {str(e)}") return { 'success': False, 'error': str(e) } def download_with_y2mate(self, url, audio_only=False): """Download YouTube video using y2mate service""" try: video_id = self.get_youtube_video_id(url) if not video_id: return {'success': False, 'error': 'Could not extract YouTube video ID'} self.logger.info(f"Attempting to download YouTube video with ID: {video_id} using y2mate") # Step 1: Get video info session = requests.Session() session.headers.update({ 'User-Agent': self.get_random_user_agent(), 'Accept-Language': 'en-US,en;q=0.9', 'Origin': 'https://www.y2mate.com', 'Referer': 'https://www.y2mate.com/en68' }) # Try multiple y2mate domains y2mate_domains = [ 'https://www.y2mate.com/mates/analyze/ajax', 'https://www.y2mate.com/mates/en68/analyze/ajax', 'https://www.y2mate.com/mates/en/analyze/ajax', 'https://www.y2mate.com/mates/en419/analyze/ajax' ] analyze_data = None for analyze_url in y2mate_domains: try: data = { 'url': f'https://www.youtube.com/watch?v={video_id}', 'q_auto': 0, 'ajax': 1 } response = session.post(analyze_url, data=data, timeout=15) if response.status_code == 200 and 'result' in response.json(): analyze_data = response.json() self.logger.info(f"Successfully got video info from {analyze_url}") break except Exception as e: self.logger.warning(f"Failed to get video info from {analyze_url}: {str(e)}") time.sleep(1) # Wait before trying next domain if not analyze_data: return {'success': False, 'error': 'Failed to get video info from y2mate'} # Parse the HTML response to get video title and k parameter soup = BeautifulSoup(analyze_data['result'], 'html.parser') title_elem = soup.select_one('.caption > b') title = title_elem.text.strip() if title_elem else f'YouTube Video {video_id}' # Find the appropriate format button if audio_only: format_btn = soup.select_one('a[data-ftype="mp3"][data-fquality="128"]') else: # Try to find the highest quality available for quality in ['1080p', '720p', '480p', '360p', '240p']: format_btn = soup.select_one(f'a[data-ftype="mp4"][data-fquality="{quality}"]') if format_btn: break if not format_btn: return {'success': False, 'error': 'Could not find suitable format button'} k = format_btn.get('data-id') if not k: return {'success': False, 'error': 'Could not extract k parameter'} # Step 2: Convert and get download link convert_domains = [ 'https://www.y2mate.com/mates/convert', 'https://www.y2mate.com/mates/en68/convert', 'https://www.y2mate.com/mates/en/convert', 'https://www.y2mate.com/mates/en419/convert' ] download_url = None for convert_url in convert_domains: try: convert_data = { 'type': 'youtube', '_id': k, 'v_id': video_id, 'ajax': '1', 'token': '', 'ftype': 'mp3' if audio_only else 'mp4', 'fquality': '128' if audio_only else format_btn.get('data-fquality', '720p') } response = session.post(convert_url, data=convert_data, timeout=15) if response.status_code == 200 and 'result' in response.json(): convert_result = response.json() soup = BeautifulSoup(convert_result['result'], 'html.parser') download_link = soup.select_one('a.btn-success') if download_link and download_link.get('href'): download_url = download_link['href'] self.logger.info(f"Successfully got download link from {convert_url}") break except Exception as e: self.logger.warning(f"Failed to get download link from {convert_url}: {str(e)}") time.sleep(1) # Wait before trying next domain if not download_url: return {'success': False, 'error': 'Failed to get download link from y2mate'} # Step 3: Download the file ext = 'mp3' if audio_only else 'mp4' output_file = os.path.join(self.download_path, f"youtube_{video_id}.{ext}") # Try multiple times with different user agents for attempt in range(3): try: session.headers.update({'User-Agent': self.get_random_user_agent()}) response = session.get(download_url, stream=True, timeout=30) if response.status_code == 200: with open(output_file, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) # Verify file was downloaded successfully if os.path.exists(output_file) and os.path.getsize(output_file) > 0: return { 'success': True, 'file_path': output_file, 'title': title, 'format': ext } except Exception as e: self.logger.warning(f"Download attempt {attempt+1} failed: {str(e)}") time.sleep(2) # Wait before retrying return {'success': False, 'error': 'Failed to download file from y2mate'} except Exception as e: self.logger.error(f"Error in y2mate download: {str(e)}") return {'success': False, 'error': str(e)} def download_with_savefrom(self, url, audio_only=False): """Download video using savefrom.net service""" try: self.logger.info(f"Attempting to download video from {url} using savefrom.net") # Try multiple savefrom API endpoints api_endpoints = [ 'https://api.savefrom.net/api/convert', 'https://api.savefrom.net/api/convert/v2', 'https://en.savefrom.net/api/convert', 'https://ssyoutube.com/api/convert' ] for api_url in api_endpoints: try: headers = { 'User-Agent': self.get_random_user_agent(), 'Content-Type': 'application/json', 'Accept': 'application/json', 'Origin': 'https://savefrom.net', 'Referer': 'https://savefrom.net/' } data = { 'url': url, 'extension': 'mp3' if audio_only else 'mp4' } response = requests.post(api_url, headers=headers, json=data, timeout=20) if response.status_code == 200: api_data = response.json() # Check if we got a valid response if 'url' in api_data and api_data['url']: download_url = api_data['url'] title = api_data.get('title', f'Video {os.urandom(4).hex()}') # Download the file platform = self.get_platform(url) ext = 'mp3' if audio_only else 'mp4' output_file = os.path.join(self.download_path, f"{platform}_{os.urandom(4).hex()}.{ext}") # Try multiple times with different user agents for attempt in range(3): try: download_headers = { 'User-Agent': self.get_random_user_agent(), 'Referer': 'https://savefrom.net/' } response = requests.get(download_url, headers=download_headers, stream=True, timeout=30) if response.status_code == 200: with open(output_file, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) # Verify file was downloaded successfully if os.path.exists(output_file) and os.path.getsize(output_file) > 0: # Convert to mp3 if audio_only if audio_only and ext != 'mp3': mp3_file = os.path.join(self.download_path, f"{platform}_{os.urandom(4).hex()}.mp3") cmd = [ 'ffmpeg', '-i', output_file, '-vn', '-ab', '128k', '-ar', '44100', '-y', mp3_file ] subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) os.remove(output_file) output_file = mp3_file ext = 'mp3' return { 'success': True, 'file_path': output_file, 'title': title, 'format': ext } except Exception as e: self.logger.warning(f"Download attempt {attempt+1} failed: {str(e)}") time.sleep(2) # Wait before retrying except Exception as e: self.logger.warning(f"Failed to use API endpoint {api_url}: {str(e)}") time.sleep(1) # Wait before trying next endpoint return {'success': False, 'error': 'Failed to download using savefrom.net'} except Exception as e: self.logger.error(f"Error in savefrom download: {str(e)}") return {'success': False, 'error': str(e)} def download_with_ssyoutube(self, url, audio_only=False): """Download YouTube video using ssyoutube.com (direct scraping)""" try: video_id = self.get_youtube_video_id(url) if not video_id: return {'success': False, 'error': 'Could not extract YouTube video ID'} self.logger.info(f"Attempting to download YouTube video with ID: {video_id} using ssyoutube") # Convert to ssyoutube URL if 'youtube.com' in url: ss_url = url.replace('youtube.com', 'ssyoutube.com') elif 'youtu.be' in url: ss_url = f"https://ssyoutube.com/watch?v={video_id}" else: ss_url = f"https://ssyoutube.com/watch?v={video_id}" # Fetch the ssyoutube page headers = { 'User-Agent': self.get_random_user_agent(), 'Accept-Language': 'en-US,en;q=0.9', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8' } response = requests.get(ss_url, headers=headers, timeout=20) if response.status_code != 200: return {'success': False, 'error': f'Failed to access ssyoutube page, status code: {response.status_code}'} # Parse the page to find download links soup = BeautifulSoup(response.text, 'html.parser') # Try to find the title title_elem = soup.select_one('h1.title') title = title_elem.text.strip() if title_elem else f'YouTube Video {video_id}' # Find download links download_links = [] # Method 1: Look for download buttons download_buttons = soup.select('a.download-btn') for button in download_buttons: quality = button.select_one('.quality') quality_text = quality.text.strip() if quality else 'Unknown' if audio_only and 'mp3' in quality_text.lower(): download_links.append({ 'url': button['href'], 'quality': quality_text, 'format': 'mp3' }) elif not audio_only and 'mp4' in quality_text.lower(): download_links.append({ 'url': button['href'], 'quality': quality_text, 'format': 'mp4' }) # Method 2: Look for download tables download_tables = soup.select('table.download-table') for table in download_tables: rows = table.select('tr') for row in rows: format_cell = row.select_one('td.format') quality_cell = row.select_one('td.quality') link_cell = row.select_one('td.download a') if format_cell and quality_cell and link_cell and link_cell.has_attr('href'): format_text = format_cell.text.strip() quality_text = quality_cell.text.strip() if audio_only and 'mp3' in format_text.lower(): download_links.append({ 'url': link_cell['href'], 'quality': quality_text, 'format': 'mp3' }) elif not audio_only and 'mp4' in format_text.lower(): download_links.append({ 'url': link_cell['href'], 'quality': quality_text, 'format': 'mp4' }) # Method 3: Look for JSON data in script tags script_tags = soup.select('script') for script in script_tags: script_text = script.string if script_text and 'downloadLinks' in script_text: try: # Extract JSON data json_match = re.search(r'var\s+downloadLinks\s*=\s*(\{.*?\});', script_text, re.DOTALL) if json_match: json_str = json_match.group(1) # Clean up the JSON string json_str = re.sub(r'(\w+):', r'"\1":', json_str) json_str = re.sub(r',\s*}', '}', json_str) download_data = json.loads(json_str) # Extract links for format_key, format_data in download_data.items(): if audio_only and 'mp3' in format_key.lower(): for quality, url in format_data.items(): download_links.append({ 'url': url, 'quality': quality, 'format': 'mp3' }) elif not audio_only and 'mp4' in format_key.lower(): for quality, url in format_data.items(): download_links.append({ 'url': url, 'quality': quality, 'format': 'mp4' }) except Exception as e: self.logger.warning(f"Failed to parse JSON data from script: {str(e)}") if not download_links: return {'success': False, 'error': 'Could not find any download links'} # Sort download links by quality (prefer higher quality) def quality_score(link): quality = link.get('quality', '').lower() if '1080' in quality: return 5 elif '720' in quality: return 4 elif '480' in quality: return 3 elif '360' in quality: return 2 elif '240' in quality: return 1 else: return 0 download_links.sort(key=quality_score, reverse=True) # Try to download from each link until successful for link_data in download_links: try: download_url = link_data['url'] format_ext = link_data['format'] # Download the file output_file = os.path.join(self.download_path, f"youtube_{video_id}.{format_ext}") headers = { 'User-Agent': self.get_random_user_agent(), 'Referer': ss_url } response = requests.get(download_url, headers=headers, stream=True, timeout=30) if response.status_code == 200: with open(output_file, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) # Verify file was downloaded successfully if os.path.exists(output_file) and os.path.getsize(output_file) > 0: return { 'success': True, 'file_path': output_file, 'title': title, 'format': format_ext } except Exception as e: self.logger.warning(f"Failed to download from link {download_url}: {str(e)}") time.sleep(1) # Wait before trying next link return {'success': False, 'error': 'Failed to download from any available links'} except Exception as e: self.logger.error(f"Error in ssyoutube download: {str(e)}") return {'success': False, 'error': str(e)} def download_with_ytdlp(self, url, audio_only=False): """Download video using yt-dlp with improved error handling""" try: import yt_dlp platform = self.get_platform(url) filename = f"{platform}_{os.urandom(4).hex()}" output_template = os.path.join(self.download_path, filename) # Add random delay to avoid rate limiting time.sleep(random.uniform(0.5, 2)) # Enhanced yt-dlp options with better error handling ydl_opts = { 'outtmpl': output_template + '.%(ext)s', 'quiet': False, 'no_warnings': False, 'verbose': True, 'ignoreerrors': True, 'nocheckcertificate': True, 'geo_bypass': True, 'geo_bypass_country': 'US', 'retries': 15, 'fragment_retries': 15, 'socket_timeout': 60, 'http_headers': { 'User-Agent': self.get_random_user_agent(), 'Accept-Language': 'en-US,en;q=0.9', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Referer': 'https://www.google.com/', }, 'external_downloader_args': ['--max-retries', '10'], 'hls_prefer_native': False, 'hls_use_mpegts': True, 'force_generic_extractor': False, 'extractor_retries': 10, 'prefer_insecure': True, } # Add YouTube-specific options if platform == 'youtube': ydl_opts.update({ 'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best', 'merge_output_format': 'mp4', 'noplaylist': True, 'extractor_args': { 'youtube': { 'player_client': ['android', 'web', 'tv_embedded', 'ios'], 'skip': ['dash', 'hls'], } }, 'youtube_include_dash_manifest': False, }) if audio_only: ydl_opts.update({ 'format': 'bestaudio/best', 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '192', }], }) final_ext = 'mp3' else: if platform != 'youtube': ydl_opts.update({ 'format': 'best', }) final_ext = 'mp4' # Try with multiple configurations if needed error_messages = [] # First attempt with standard configuration try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: self.logger.info(f"Attempting to download {url} with yt-dlp (attempt 1)") info = ydl.extract_info(url, download=True) if info: # Success on first attempt if 'entries' in info: # Playlist if not info['entries']: raise Exception("Empty playlist or all videos failed") info = info['entries'][0] # Get the actual file extension if not audio_only: final_ext = info.get('ext', 'mp4') title = info.get('title', 'Video') file_path = f"{output_template}.{final_ext}" # Verify file exists if not os.path.exists(file_path): # Try to find the actual file possible_files = [f for f in os.listdir(self.download_path) if f.startswith(filename)] if possible_files: file_path = os.path.join(self.download_path, possible_files[0]) else: raise Exception("Downloaded file not found") return { 'success': True, 'file_path': file_path, 'title': title, 'format': final_ext } except Exception as e: error_messages.append(f"Attempt 1 failed: {str(e)}") self.logger.warning(f"First download attempt failed: {str(e)}") # Second attempt with different options try: # Modify options for second attempt second_attempt_opts = ydl_opts.copy() second_attempt_opts.update({ 'format': 'best', 'force_generic_extractor': True, }) with yt_dlp.YoutubeDL(second_attempt_opts) as ydl: self.logger.info(f"Attempting to download {url} with yt-dlp (attempt 2)") info = ydl.extract_info(url, download=True) if info: # Success on second attempt if 'entries' in info: # Playlist if not info['entries']: raise Exception("Empty playlist or all videos failed") info = info['entries'][0] # Get the actual file extension if not audio_only: final_ext = info.get('ext', 'mp4') title = info.get('title', 'Video') file_path = f"{output_template}.{final_ext}" # Verify file exists if not os.path.exists(file_path): # Try to find the actual file possible_files = [f for f in os.listdir(self.download_path) if f.startswith(filename)] if possible_files: file_path = os.path.join(self.download_path, possible_files[0]) else: raise Exception("Downloaded file not found") return { 'success': True, 'file_path': file_path, 'title': title, 'format': final_ext } except Exception as e: error_messages.append(f"Attempt 2 failed: {str(e)}") self.logger.warning(f"Second download attempt failed: {str(e)}") # If we get here, all attempts failed error_summary = "; ".join(error_messages) self.logger.error(f"All download attempts failed: {error_summary}") return {'success': False, 'error': f"Failed to extract video information after multiple attempts: {error_summary}"} except Exception as e: self.logger.error(f"Error in download_with_ytdlp: {str(e)}") return {'success': False, 'error': str(e)} def download_youtube_aggressive(self, url, audio_only=False): """Aggressively try multiple methods to download YouTube video""" self.logger.info(f"Starting aggressive YouTube download for {url}") # Try all methods in sequence until one succeeds methods = [ ('y2mate', self.download_with_y2mate), ('savefrom', self.download_with_savefrom), ('ssyoutube', self.download_with_ssyoutube), ('yt-dlp', self.download_with_ytdlp) ] all_errors = [] # Try different URL formats urls_to_try = [url] video_id = self.get_youtube_video_id(url) if video_id: # Add alternative URL formats if 'youtu.be' in url: urls_to_try.append(f"https://www.youtube.com/watch?v={video_id}") else: urls_to_try.append(f"https://youtu.be/{video_id}") # Add URL with timestamp parameter urls_to_try.append(f"https://www.youtube.com/watch?v={video_id}&t=0s") # Add URL with additional parameters urls_to_try.append(f"https://www.youtube.com/watch?v={video_id}&feature=youtu.be") # Try each URL with each method for current_url in urls_to_try: self.logger.info(f"Trying URL format: {current_url}") for method_name, method_func in methods: try: self.logger.info(f"Attempting download with method: {method_name}") result = method_func(current_url, audio_only) if result['success']: self.logger.info(f"Successfully downloaded with method: {method_name}") return result else: all_errors.append(f"{method_name}: {result.get('error', 'Unknown error')}") self.logger.warning(f"Method {method_name} failed: {result.get('error', 'Unknown error')}") except Exception as e: all_errors.append(f"{method_name}: {str(e)}") self.logger.error(f"Exception in method {method_name}: {str(e)}") # Add delay between attempts time.sleep(random.uniform(1, 3)) # If all methods failed, return error error_summary = "; ".join(all_errors) self.logger.error(f"All download methods failed: {error_summary}") return {'success': False, 'error': f"Failed to extract video information after multiple attempts"} def download(self, url, audio_only=False, compress=True, target_size_mb=10): """Main download method with multiple fallback options""" self.logger.info(f"Starting download for {url}, audio_only={audio_only}, compress={compress}") platform = self.get_platform(url) self.logger.info(f"Detected platform: {platform}") # Try platform-specific methods first if platform == 'youtube': result = self.download_youtube_aggressive(url, audio_only) elif platform == 'instagram': # Try direct methods first, then fallback to yt-dlp result = self.download_with_savefrom(url, audio_only) if not result['success']: result = self.download_with_ytdlp(url, audio_only) else: # For other platforms, try yt-dlp result = self.download_with_ytdlp(url, audio_only) # If download was successful and it's a video (not audio) and compression is enabled if result['success'] and not audio_only and compress: file_path = result['file_path'] file_size_mb = os.path.getsize(file_path) / (1024 * 1024) # Only compress if file is larger than target size if file_size_mb > target_size_mb: self.logger.info(f"File size ({file_size_mb:.2f}MB) exceeds target size ({target_size_mb}MB). Compressing...") compressed_path = self.compress_video(file_path, target_size_mb) # Update file path if compression was successful if compressed_path != file_path: result['file_path'] = compressed_path result['compressed'] = True # Get new file size new_size_mb = os.path.getsize(compressed_path) / (1024 * 1024) self.logger.info(f"Compression complete. New size: {new_size_mb:.2f}MB") else: result['compressed'] = False else: self.logger.info(f"File size ({file_size_mb:.2f}MB) is already below target size ({target_size_mb}MB). Skipping compression.") result['compressed'] = False return result