import os import logging import subprocess import tempfile import uuid import shutil from pytube import YouTube 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__) 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 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_youtube(self, url, audio_only=False): """Download video from YouTube using pytube with improved error handling""" try: yt = YouTube(url) if audio_only: # Download audio only stream = yt.streams.filter(only_audio=True).first() if not stream: self.logger.error(f"No audio stream found for {url}") return {'success': False, 'error': 'No audio stream found'} out_file = stream.download(output_path=self.download_path) # Convert to mp3 base, _ = os.path.splitext(out_file) mp3_file = base + '.mp3' os.rename(out_file, mp3_file) return {'success': True, 'file_path': mp3_file, 'title': yt.title, 'format': 'mp3'} else: # Download video with highest resolution stream = yt.streams.get_highest_resolution() if not stream: self.logger.error(f"No video stream found for {url}") return {'success': False, 'error': 'No video stream found'} out_file = stream.download(output_path=self.download_path) return {'success': True, 'file_path': out_file, 'title': yt.title, 'format': 'mp4'} except Exception as e: self.logger.error(f"Error downloading from YouTube with pytube: {str(e)}") # If pytube fails, try with yt-dlp as fallback self.logger.info(f"Falling back to yt-dlp for {url}") return self.download_with_ytdlp(url, audio_only) def download_with_ytdlp(self, url, audio_only=False): """Download video using yt-dlp for all platforms 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) # Enhanced yt-dlp options with better error handling ydl_opts = { 'outtmpl': output_template + '.%(ext)s', 'quiet': False, # Changed to False to see more info 'no_warnings': False, # Changed to False to see warnings 'verbose': True, # Added for more detailed logs 'ignoreerrors': True, # Skip unavailable videos in a playlist 'nocheckcertificate': True, # Ignore SSL cert verification 'geo_bypass': True, # Try to bypass geo-restrictions 'retries': 10, # Retry 10 times if download fails 'socket_timeout': 30, # Timeout after 30 seconds } if audio_only: ydl_opts.update({ 'format': 'bestaudio/best', 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '192', }], }) final_ext = 'mp3' else: ydl_opts.update({ 'format': 'best', }) final_ext = 'mp4' # Most common, but might be different with yt_dlp.YoutubeDL(ydl_opts) as ydl: self.logger.info(f"Attempting to download {url} with yt-dlp") info = ydl.extract_info(url, download=True) if not info: self.logger.error(f"No information extracted for {url}") return {'success': False, 'error': 'Failed to extract video information'} if 'entries' in info: # Playlist if not info['entries']: self.logger.error(f"Empty playlist or all videos failed for {url}") return {'success': False, 'error': '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]) self.logger.info(f"Found alternative file: {file_path}") else: self.logger.error(f"Downloaded file not found at {file_path}") return {'success': False, 'error': 'Downloaded file not found'} return { 'success': True, 'file_path': file_path, 'title': title, 'format': final_ext } except Exception as e: self.logger.error(f"Error downloading with yt-dlp: {str(e)}") return {'success': False, 'error': str(e)} def download(self, url, audio_only=False, compress=True, target_size_mb=10): """Main download method with compression option""" 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 yt-dlp first for all platforms as it's more robust result = self.download_with_ytdlp(url, audio_only) # If yt-dlp fails for YouTube and we're downloading video, try pytube as fallback if not result['success'] and platform == 'youtube' and not audio_only: self.logger.info(f"yt-dlp failed, trying pytube for {url}") result = self.download_youtube(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