import os import time import datetime from datetime import datetime, time as dt_time import pickle import threading import queue import logging import logging.handlers import glob import json from concurrent.futures import ThreadPoolExecutor from googleapiclient.discovery import build import subprocess from google.auth.transport.requests import Request import sys from auth import get_token, get_token_non_interactive, get_service_account_token # Home Assistant Integration try: from homeassistant_integration import ( init_homeassistant_integration, update_homeassistant_stats, stop_homeassistant_integration, set_service_waiting, set_service_running, set_service_completed, set_service_error ) HOMEASSISTANT_ENABLED = True print("šŸ  Home Assistant integration available") except ImportError: HOMEASSISTANT_ENABLED = False print("āš ļø Home Assistant integration not available (install paho-mqtt and psutil)") # Dummy functions if Home Assistant is not available def set_service_waiting(next_run): pass def set_service_running(): pass def set_service_completed(): pass def set_service_error(): pass # Dummy functions if not available def init_homeassistant_integration(*args, **kwargs): return None def update_homeassistant_stats(*args, **kwargs): pass def stop_homeassistant_integration(*args, **kwargs): pass # Constants OUTPUT_DIR = "/media/media/YouTube" # Change this to your Emby folder DAYS_BACK = 5 # Videos published in the last 5 days DAYS_TO_KEEP = 10 # Delete videos older than 10 days TOKEN_FILE = "youtube_token.pickle" # Token file from the OAuth script EXIT_TIME = dt_time(6, 0) # Exit the script at 06:00 POPULAR_VIDEOS_MAX_RESULTS = 10 # Number of popular videos to fetch (you can change this) REGION_CODES = ["US", "GB", "LU", "LK"] # Regions to fetch popular videos from # Proxy Configuration (set to None to disable) HTTP_PROXY = os.environ.get("HTTP_PROXY", None) # e.g., "http://proxy.example.com:8080" HTTPS_PROXY = os.environ.get("HTTPS_PROXY", None) # e.g., "http://proxy.example.com:8080" # Alternative: Set directly in code # HTTP_PROXY = "http://proxy.example.com:8080" # HTTPS_PROXY = "http://proxy.example.com:8080" # Logging Configuration LOG_DIR = "logs" LOG_FILE = "youtube_predownloader.log" LOG_LEVEL = logging.INFO LOG_MAX_SIZE = 50 * 1024 * 1024 # 50MB per log file LOG_BACKUP_COUNT = 5 # Keep 5 backup files LOG_RETENTION_DAYS = 10 # Delete logs older than 10 days # Conversion Queue Settings MAX_CONCURRENT_CONVERSIONS = 2 # Maximum number of simultaneous video conversions CONVERSION_QUEUE_SIZE = 50 # Maximum items in conversion queue def setup_logging(): """Set up comprehensive logging with file rotation and cleanup.""" # Create logs directory if it doesn't exist if not os.path.exists(LOG_DIR): os.makedirs(LOG_DIR) # Clean up old log files first cleanup_old_logs() # Create logger logger = logging.getLogger() logger.setLevel(LOG_LEVEL) # Clear any existing handlers for handler in logger.handlers[:]: logger.removeHandler(handler) # Create formatters detailed_formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) console_formatter = logging.Formatter( '%(asctime)s - %(levelname)s - %(message)s', datefmt='%H:%M:%S' ) # File handler with rotation log_file_path = os.path.join(LOG_DIR, LOG_FILE) file_handler = logging.handlers.RotatingFileHandler( log_file_path, maxBytes=LOG_MAX_SIZE, backupCount=LOG_BACKUP_COUNT, encoding='utf-8' ) file_handler.setLevel(logging.DEBUG) # Log everything to file file_handler.setFormatter(detailed_formatter) # Console handler (less verbose) console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(LOG_LEVEL) console_handler.setFormatter(console_formatter) # Add handlers to logger logger.addHandler(file_handler) logger.addHandler(console_handler) # Log startup information logger.info("=" * 80) logger.info("YouTube Predownloader Starting") logger.info("=" * 80) logger.info(f"Log file: {log_file_path}") logger.info(f"Log level: {logging.getLevelName(LOG_LEVEL)}") logger.info(f"Output directory: {OUTPUT_DIR}") logger.info(f"Days back to fetch: {DAYS_BACK}") logger.info(f"Days to keep videos: {DAYS_TO_KEEP}") logger.info(f"Log retention days: {LOG_RETENTION_DAYS}") return logger def cleanup_old_logs(): """Remove log files older than LOG_RETENTION_DAYS.""" try: cutoff_time = time.time() - (LOG_RETENTION_DAYS * 86400) log_pattern = os.path.join(LOG_DIR, "*.log*") deleted_count = 0 for log_file in glob.glob(log_pattern): try: if os.path.getmtime(log_file) < cutoff_time: os.remove(log_file) deleted_count += 1 print(f"Deleted old log file: {log_file}") except (OSError, IOError) as e: print(f"Failed to delete log file {log_file}: {e}") if deleted_count > 0: print(f"Cleaned up {deleted_count} old log files") except Exception as e: print(f"Error during log cleanup: {e}") def log_system_info(): """Log system and environment information.""" logger = logging.getLogger(__name__) try: # Log Python and system info import platform import psutil logger.info("System Information:") logger.info(f" Python version: {platform.python_version()}") logger.info(f" Platform: {platform.platform()}") logger.info(f" CPU count: {os.cpu_count()}") # Memory info memory = psutil.virtual_memory() logger.info(f" Total memory: {memory.total // (1024**3):.1f} GB") logger.info(f" Available memory: {memory.available // (1024**3):.1f} GB") # Disk space for output directory if os.path.exists(OUTPUT_DIR): disk_usage = psutil.disk_usage(OUTPUT_DIR) logger.info(f" Output disk free: {disk_usage.free // (1024**3):.1f} GB") # Check required tools tools = [ ('yt-dlp', ['yt-dlp', '--version']), ('ffmpeg', ['ffmpeg', '-version']), ('ffprobe', ['ffprobe', '-version']) ] for tool_name, cmd in tools: try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode == 0: version_line = result.stdout.split('\n')[0] if result.stdout else "unknown" logger.info(f" {tool_name}: {version_line}") else: logger.warning(f" {tool_name}: Not found or error") except (subprocess.TimeoutExpired, subprocess.SubprocessError, FileNotFoundError): logger.error(f" {tool_name}: Not available") except Exception as e: logger.error(f"Error collecting system info: {e}") # Initialize logging at module level logger = setup_logging() log_system_info() # FFmpeg/FFprobe paths FFPROBE_PATH = "/usr/bin/ffprobe" FFMPEG_PATH = "/usr/bin/ffmpeg" # Global conversion queue and executor conversion_queue = queue.Queue(maxsize=CONVERSION_QUEUE_SIZE) conversion_executor = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_CONVERSIONS, thread_name_prefix="VideoConverter") # Progress tracking download_stats = { 'total_videos': 0, 'completed_videos': 0, 'current_video': '', 'download_speed': 0, 'active_downloads': 0 } conversion_stats = { 'active_conversions': 0, 'completed_conversions': 0, 'current_conversion': '', 'conversion_progress': 0 } def update_yt_dlp(): """Update yt-dlp to the latest version.""" logger = logging.getLogger(__name__) logger.info("Updating yt-dlp to the latest version...") try: result = subprocess.run(["pip", "install", "--upgrade", "yt-dlp"], capture_output=True, text=True, check=True) logger.info("yt-dlp updated successfully!") logger.debug(f"yt-dlp update output: {result.stdout}") except subprocess.CalledProcessError as e: logger.error(f"Failed to update yt-dlp: {e}") logger.debug(f"yt-dlp update error: {e.stderr}") raise def authenticate_youtube_api(): """Authenticate using our enhanced auth system (tries non-interactive first).""" logger = logging.getLogger(__name__) logger.info("Attempting non-interactive authentication...") # Try non-interactive authentication first creds = get_token_non_interactive() if not creds: logger.error("Non-interactive authentication failed.") logger.info("You need to run: python3 auth.py --interactive") logger.info("Or set up service account authentication.") raise Exception("Authentication failed. No valid credentials available.") logger.info("Authentication successful!") # Build the YouTube API client youtube = build("youtube", "v3", credentials=creds) logger.debug("YouTube API client created successfully") return youtube def authenticate_with_token(): """Legacy function - kept for compatibility but redirects to new auth system.""" return authenticate_youtube_api() def get_subscribed_channels(youtube): """Fetch the list of subscribed channels.""" logger = logging.getLogger(__name__) logger.info("Fetching subscribed channels...") channels = [] request = youtube.subscriptions().list( part="snippet", mine=True, maxResults=50 ) page_count = 0 while request: page_count += 1 logger.debug(f"Fetching subscriptions page {page_count}") response = request.execute() page_channels = len(response["items"]) logger.debug(f"Found {page_channels} channels on page {page_count}") for item in response["items"]: channel_id = item["snippet"]["resourceId"]["channelId"] channel_title = item["snippet"]["title"] channels.append(channel_id) logger.debug(f"Added channel: {channel_title} ({channel_id})") request = youtube.subscriptions().list_next(request, response) logger.info(f"Found {len(channels)} subscribed channels") return channels def get_recent_videos(youtube, channel_id): """Fetch videos from a channel published in the last `DAYS_BACK` days.""" logger = logging.getLogger(__name__) recent_videos = [] published_after = (datetime.datetime.now() - datetime.timedelta(days=DAYS_BACK)).isoformat("T") + "Z" try: request = youtube.search().list( part="snippet", channelId=channel_id, publishedAfter=published_after, maxResults=50, type="video", order="date" ) response = request.execute() for item in response["items"]: video_data = { "video_id": item["id"]["videoId"], "title": item["snippet"]["title"], "description": item["snippet"]["description"], "published_at": item["snippet"]["publishedAt"], "channel_title": item["snippet"]["channelTitle"] } recent_videos.append(video_data) logger.debug(f"Found recent video: {video_data['title']} from {video_data['channel_title']}") logger.debug(f"Found {len(recent_videos)} recent videos from channel {channel_id}") except Exception as e: logger.error(f"Error fetching recent videos from channel {channel_id}: {e}") return recent_videos def get_popular_videos(youtube, region_code): """Fetch the most popular videos for a given region.""" logger = logging.getLogger(__name__) popular_videos = [] try: logger.debug(f"Fetching popular videos for region: {region_code}") request = youtube.videos().list( part="snippet,statistics", chart="mostPopular", regionCode=region_code, # Region code passed here maxResults=POPULAR_VIDEOS_MAX_RESULTS ) response = request.execute() for item in response["items"]: video_data = { "video_id": item["id"], "title": item["snippet"]["title"], "description": item["snippet"]["description"], "published_at": item["snippet"]["publishedAt"], "channel_title": item["snippet"]["channelTitle"] } popular_videos.append(video_data) logger.debug(f"Found popular video ({region_code}): {video_data['title']} from {video_data['channel_title']}") logger.info(f"Found {len(popular_videos)} popular videos from region {region_code}") except Exception as e: logger.error(f"Error fetching popular videos from region {region_code}: {e}") return popular_videos def list_available_formats(video_id): """List available formats for a video (for debugging).""" logger = logging.getLogger(__name__) try: command = [ "yt-dlp", "--list-formats", f"https://www.youtube.com/watch?v={video_id}" ] result = subprocess.run(command, capture_output=True, text=True, timeout=30) if result.returncode == 0: logger.debug(f"Available formats for {video_id}:") for line in result.stdout.split('\n'): if 'mp4' in line or 'webm' in line or '2160p' in line or '1440p' in line or '1080p' in line: logger.debug(f" {line.strip()}") else: logger.warning(f"Could not list formats for {video_id}: {result.stderr}") except Exception as e: logger.warning(f"Error listing formats for {video_id}: {e}") def get_proxy_args(): """Get proxy arguments for yt-dlp commands.""" proxy_args = [] if HTTP_PROXY or HTTPS_PROXY: # Use the first available proxy for both HTTP and HTTPS proxy_url = HTTP_PROXY or HTTPS_PROXY proxy_args.extend(["--proxy", proxy_url]) logger = logging.getLogger(__name__) logger.info(f"🌐 Using proxy: {proxy_url}") return proxy_args def download_video(video_id, output_path, video_title="Unknown"): """Download a YouTube video with quality priority - downloads 4K VP9/AV1 if needed for conversion.""" global download_stats logger = logging.getLogger(__name__) try: # Update stats download_stats['active_downloads'] += 1 download_stats['current_video'] = video_title[:50] + "..." if len(video_title) > 50 else video_title logger.info(f"Starting download: {video_title} (ID: {video_id})") logger.debug(f"Download path: {output_path}") # First, check what format would be downloaded to determine compatibility logger.debug(f"Checking video format compatibility for: {video_title}") simulate_command = [ "yt-dlp", "--simulate", "--print", "%(format_id)s|%(vcodec)s|%(height)s|%(ext)s", "-f", "(bestvideo[height>=2160][vcodec~='^(avc|h264|hevc|h265)']+bestaudio)/(bestvideo[height>=2160]+bestaudio)/(bestvideo[height>=1440]+bestaudio)/(bestvideo[height>=1080]+bestaudio)/best", f"https://www.youtube.com/watch?v={video_id}" ] will_be_compatible = True predicted_codec = "unknown" predicted_height = 0 temp_filename = None final_filename = None try: sim_result = subprocess.run(simulate_command, capture_output=True, text=True, timeout=30) if sim_result.returncode == 0 and sim_result.stdout.strip(): format_info = sim_result.stdout.strip().split('|') if len(format_info) >= 3: predicted_codec = format_info[1] if format_info[1] != 'none' else 'unknown' predicted_height = int(format_info[2]) if format_info[2].isdigit() else 0 # Check if the predicted codec is Apple TV incompatible if any(incompatible in predicted_codec.lower() for incompatible in ['vp9', 'av01', 'av1']): will_be_compatible = False # Create temporary filename for incompatible formats sanitized_title = sanitize_filename(video_title) temp_filename = f"{sanitized_title}.incompatible" final_filename = f"{sanitized_title}" logger.info(f"āš ļø Incompatible codec detected: {predicted_codec} - downloading as temporary file") logger.info(f" Temporary: {temp_filename}") logger.info(f" Final: {final_filename}") else: logger.info(f"āœ… Compatible codec detected: {predicted_codec}") except Exception as e: logger.debug(f"Could not simulate download format: {e}") # Set output filename based on compatibility prediction if temp_filename: output_template = os.path.join(output_path, f"{temp_filename}.%(ext)s") else: output_template = os.path.join(output_path, "%(title)s.%(ext)s") # List available formats for debugging (only in debug mode) if logger.isEnabledFor(logging.DEBUG): list_available_formats(video_id) # Update Home Assistant update_homeassistant_stats( download_queue=download_stats['total_videos'] - download_stats['completed_videos'], conversion_queue=conversion_queue.qsize(), download_progress=round((download_stats['completed_videos'] / max(download_stats['total_videos'], 1)) * 100, 1), active_downloads=download_stats['active_downloads'], active_conversions=conversion_stats['active_conversions'] ) # 4K-prioritized approach: Try Apple TV compatible 4K first, then any 4K, then lower resolutions command = [ "yt-dlp", "-f", "(bestvideo[height>=2160][vcodec~='^(avc|h264|hevc|h265)']+bestaudio)/(bestvideo[height>=2160]+bestaudio)/(bestvideo[height>=1440]+bestaudio)/(bestvideo[height>=1080]+bestaudio)/best", "--merge-output-format", "mp4", "--no-mtime", "--newline", # Better for parsing progress "--verbose" if logger.isEnabledFor(logging.DEBUG) else "--no-warnings", "--write-info-json", # Write metadata for analysis f"https://www.youtube.com/watch?v={video_id}", "-o", output_template ] logger.debug(f"yt-dlp command: {' '.join(command)}") try: start_time = time.time() logging.info(f"Starting download with command: {' '.join(command)}") # Add progress hook for yt-dlp progress tracking command.extend(["--progress-template", "download:%(progress._percent_str)s %(progress._eta_str)s %(progress._speed_str)s"]) # Start download with progress monitoring process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True) current_progress = 0 while True: output = process.stdout.readline() if output == '' and process.poll() is not None: break if output: output = output.strip() logging.debug(f"yt-dlp output: {output}") # Parse progress from yt-dlp output if "%" in output and "ETA" in output: try: # Extract percentage from progress line parts = output.split() for part in parts: if "%" in part: progress_str = part.replace("%", "") if progress_str.replace(".", "").isdigit(): new_progress = float(progress_str) if abs(new_progress - current_progress) >= 5: # Update every 5% current_progress = new_progress # Update Home Assistant with current progress update_homeassistant_stats( download_queue=download_stats['total_videos'] - download_stats['completed_videos'], conversion_queue=conversion_queue.qsize(), download_progress=round((download_stats['completed_videos'] / max(download_stats['total_videos'], 1)) * 100, 1), active_downloads=download_stats['active_downloads'], active_conversions=conversion_stats['active_conversions'], current_download_progress=current_progress, current_download_file=video_title[:50] ) logger.info(f"Download progress: {current_progress:.1f}% - {video_title}") break except (ValueError, IndexError): pass # Wait for process to complete process.wait() end_time = time.time() if process.returncode != 0: stderr_output = process.stderr.read() logging.error(f"yt-dlp failed with return code {process.returncode}: {stderr_output}") raise subprocess.CalledProcessError(process.returncode, command, stderr=stderr_output) download_time = end_time - start_time logger.info(f"Download completed in {download_time:.1f}s: {video_title}") # Clear current download progress update_homeassistant_stats( download_queue=download_stats['total_videos'] - download_stats['completed_videos'], conversion_queue=conversion_queue.qsize(), download_progress=round((download_stats['completed_videos'] / max(download_stats['total_videos'], 1)) * 100, 1), active_downloads=download_stats['active_downloads'], active_conversions=conversion_stats['active_conversions'], current_download_progress=100, current_download_file="" ) # Analyze what was actually downloaded and check Apple TV compatibility try: downloaded_files = [f for f in os.listdir(output_path) if f.endswith(('.mp4', '.mkv', '.webm', '.m4v'))] if downloaded_files: for file in downloaded_files: file_path = os.path.join(output_path, file) try: # Get video info probe_cmd = [FFPROBE_PATH, "-v", "quiet", "-print_format", "json", "-show_streams", file_path] probe_result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) video_info = json.loads(probe_result.stdout) # Find video stream video_stream = next((s for s in video_info['streams'] if s['codec_type'] == 'video'), None) if video_stream: height = video_stream.get('height', 'unknown') width = video_stream.get('width', 'unknown') codec = video_stream.get('codec_name', 'unknown') bitrate = video_stream.get('bit_rate', 'unknown') fps = video_stream.get('r_frame_rate', 'unknown') quality_label = "Unknown" if height >= 2160: quality_label = "4K" elif height >= 1440: quality_label = "1440p" elif height >= 1080: quality_label = "1080p" elif height >= 720: quality_label = "720p" logger.info(f"Downloaded quality: {quality_label} ({width}x{height}), codec: {codec}, fps: {fps}") if bitrate != 'unknown': bitrate_mbps = int(bitrate) / 1000000 if bitrate.isdigit() else 'unknown' logger.info(f"Video bitrate: {bitrate_mbps:.1f} Mbps") # Log file size file_size = os.path.getsize(file_path) file_size_mb = file_size / (1024 * 1024) logger.info(f"Downloaded file size: {file_size_mb:.1f} MB") # Check if this is a temporary incompatible file is_temp_file = ".incompatible." in file # Check Apple TV compatibility and queue for conversion if needed needs_conversion = False conversion_reason = "" target_filename = None if is_temp_file: # This was downloaded as incompatible - definitely needs conversion needs_conversion = True conversion_reason = f"Pre-identified incompatible format ({codec})" # Set target filename (remove .incompatible from the name but keep extension) base_name, ext = os.path.splitext(file) if ".incompatible" in base_name: clean_base = base_name.replace(".incompatible", "") target_filename = f"{clean_base}{ext}" else: target_filename = file target_path = os.path.join(output_path, target_filename) logger.info(f"šŸ”„ Temporary incompatible file detected - will convert to: {target_filename}") else: # Regular compatibility check for files downloaded with standard names if any(incompatible in codec.lower() for incompatible in ['vp9', 'av01', 'av1']): needs_conversion = True conversion_reason = f"Incompatible codec ({codec}) for Apple TV" elif not file.endswith('.mp4'): needs_conversion = True conversion_reason = f"Incompatible container ({file.split('.')[-1]}) for Apple TV" elif height >= 2160 and not any(compatible in codec.lower() for compatible in ['h264', 'avc1', 'hevc', 'h265']): needs_conversion = True conversion_reason = f"4K video with non-Apple TV codec ({codec})" if needs_conversion: logger.info(f"Queuing for conversion: {conversion_reason}") # Determine priority based on video resolution priority = 2 if height >= 2160 else 1 # 4K gets higher priority # Add to conversion queue with target filename if specified if target_filename: add_to_conversion_queue_with_target(file_path, target_path, priority) logger.info(f"šŸ“‹ Queued: {file} → {target_filename}") else: add_to_conversion_queue(file_path, priority) logger.info(f"šŸ“‹ Queued: {file}") print_queue_status() else: logger.info(f"āœ… File is Apple TV compatible: {codec} in MP4 container") except Exception as e: logger.debug(f"Could not analyze downloaded file {file}: {e}") except Exception as e: logger.debug(f"Could not analyze downloaded files: {e}") except subprocess.CalledProcessError as e: download_stats['active_downloads'] -= 1 logger.error(f"Download failed for {video_title} (ID: {video_id}): {e}") logger.debug(f"yt-dlp error output: {e.stderr}") return False except Exception as e: download_stats['active_downloads'] -= 1 logger.error(f"Unexpected error downloading {video_title} (ID: {video_id}): {e}") # Update completed count download_stats['completed_videos'] += 1 download_stats['active_downloads'] -= 1 return True # Update Home Assistant with completion update_homeassistant_stats( download_queue=download_stats['total_videos'] - download_stats['completed_videos'], download_progress=round((download_stats['completed_videos'] / max(download_stats['total_videos'], 1)) * 100, 1), active_downloads=download_stats['active_downloads'] ) return True except subprocess.CalledProcessError as e: download_stats['active_downloads'] -= 1 logger.error(f"Download failed for {video_title} (ID: {video_id}): {e}") logger.debug(f"yt-dlp error output: {e.stderr}") return False except Exception as e: download_stats['active_downloads'] -= 1 logger.error(f"Unexpected error downloading {video_title} (ID: {video_id}): {e}") return False def add_to_conversion_queue(input_file, priority=1): """Add a video file to the conversion queue with priority.""" logger = logging.getLogger(__name__) try: # Higher number = higher priority (4K gets priority 2, others get 1) conversion_queue.put((priority, input_file), timeout=1) logger.info(f"Added to conversion queue: {os.path.basename(input_file)} (priority: {priority})") logger.debug(f"Queue size after addition: {conversion_queue.qsize()}") return True except queue.Full: logger.warning(f"Conversion queue is full! Skipping: {os.path.basename(input_file)}") return False except Exception as e: logger.error(f"Error adding to conversion queue: {e}") return False def add_to_conversion_queue_with_target(input_file, target_file, priority=1): """Add a video file to the conversion queue with a specific target filename.""" logger = logging.getLogger(__name__) try: # Higher number = higher priority (4K gets priority 2, others get 1) # Store as tuple: (priority, input_file, target_file) conversion_queue.put((priority, input_file, target_file), timeout=1) logger.info(f"Added to conversion queue: {os.path.basename(input_file)} → {os.path.basename(target_file)} (priority: {priority})") logger.debug(f"Queue size after addition: {conversion_queue.qsize()}") return True except queue.Full: logger.warning(f"Conversion queue is full! Skipping: {os.path.basename(input_file)}") return False except Exception as e: logger.error(f"Error adding to conversion queue: {e}") return False def get_queue_status(): """Get current conversion queue status.""" queue_size = conversion_queue.qsize() active_conversions = conversion_executor._threads # Active thread count return { 'queued': queue_size, 'active': len(active_conversions) if active_conversions else 0, 'max_concurrent': MAX_CONCURRENT_CONVERSIONS } def print_queue_status(): """Print current queue status.""" logger = logging.getLogger(__name__) status = get_queue_status() logger.info(f"Conversion Status: {status['active']}/{status['max_concurrent']} active, {status['queued']} queued") def process_conversion_queue(): """Background worker to process the conversion queue.""" logger = logging.getLogger(__name__) logger.info("Conversion queue worker started") while True: try: # Get item from queue (blocks until available) queue_item = conversion_queue.get(timeout=30) # Handle both old format (priority, input_file) and new format (priority, input_file, target_file) if len(queue_item) == 2: priority, input_file = queue_item target_file = None elif len(queue_item) == 3: priority, input_file, target_file = queue_item else: logger.error(f"Invalid queue item format: {queue_item}") conversion_queue.task_done() continue # Check if file still exists if not os.path.exists(input_file): logger.warning(f"File no longer exists, skipping conversion: {input_file}") conversion_queue.task_done() continue if target_file: logger.info(f"Starting conversion (priority {priority}): {os.path.basename(input_file)} → {os.path.basename(target_file)}") # Perform the conversion with target filename success = convert_to_apple_tv_compatible_with_target(input_file, target_file) else: logger.info(f"Starting conversion (priority {priority}): {os.path.basename(input_file)}") # Perform the conversion with standard naming success = convert_to_apple_tv_compatible(input_file) if success: logger.info(f"Conversion completed successfully: {os.path.basename(input_file)}") else: logger.error(f"Conversion failed: {os.path.basename(input_file)}") # Mark task as done conversion_queue.task_done() except queue.Empty: # No items in queue for 30 seconds, continue waiting logger.debug("Conversion queue empty, waiting for new items...") continue except Exception as e: logger.error(f"Conversion queue worker error: {e}") try: conversion_queue.task_done() except: pass def sanitize_filename(filename): """Sanitize a filename to remove invalid characters.""" import re # Remove invalid filename characters sanitized = re.sub(r'[<>:"/\\|?*]', '_', filename) # Remove extra spaces and limit length sanitized = ' '.join(sanitized.split())[:200] return sanitized def test_vaapi_support(): """Test if Intel VAAPI/QSV hardware acceleration is available for both encoding and decoding.""" logger = logging.getLogger(__name__) try: # Test H.264 VAAPI encoding test_cmd = [FFMPEG_PATH, "-hide_banner", "-f", "lavfi", "-i", "testsrc=duration=1:size=320x240:rate=1", "-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload", "-c:v", "h264_vaapi", "-frames:v", "1", "-f", "null", "-"] result = subprocess.run(test_cmd, capture_output=True, text=True, timeout=10) h264_vaapi_available = result.returncode == 0 # Test HEVC VAAPI encoding (optional, may not work on all Intel GPUs) test_cmd = [FFMPEG_PATH, "-hide_banner", "-f", "lavfi", "-i", "testsrc=duration=1:size=320x240:rate=1", "-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload", "-c:v", "hevc_vaapi", "-frames:v", "1", "-f", "null", "-"] result = subprocess.run(test_cmd, capture_output=True, text=True, timeout=10) hevc_vaapi_available = result.returncode == 0 # Test H.264 VAAPI decoding - create a small test file and try to decode it h264_decode_available = False try: # Create a small test H.264 file create_test_cmd = [FFMPEG_PATH, "-hide_banner", "-f", "lavfi", "-i", "testsrc=duration=1:size=320x240:rate=1", "-c:v", "libx264", "-frames:v", "5", "-f", "mp4", "/tmp/vaapi_test.mp4"] subprocess.run(create_test_cmd, capture_output=True, text=True, timeout=10, check=True) # Test hardware decoding decode_test_cmd = [FFMPEG_PATH, "-hide_banner", "-hwaccel", "vaapi", "-hwaccel_device", "/dev/dri/renderD128", "-i", "/tmp/vaapi_test.mp4", "-f", "null", "-"] result = subprocess.run(decode_test_cmd, capture_output=True, text=True, timeout=10) h264_decode_available = result.returncode == 0 # Clean up test file try: os.remove("/tmp/vaapi_test.mp4") except: pass except: pass # Test VP9 VAAPI decoding capability (Intel HD Graphics 630 supports VP9 via VAAPI) vp9_vaapi_decode_available = False try: # Create a small VP9 test file create_vp9_cmd = [FFMPEG_PATH, "-hide_banner", "-f", "lavfi", "-i", "testsrc=duration=1:size=320x240:rate=1", "-c:v", "libvpx-vp9", "-frames:v", "5", "-f", "webm", "/tmp/vp9_test.webm"] result = subprocess.run(create_vp9_cmd, capture_output=True, text=True, timeout=15) if result.returncode == 0: # Test VP9 hardware decoding with VAAPI decode_vp9_cmd = [FFMPEG_PATH, "-hide_banner", "-hwaccel", "vaapi", "-hwaccel_device", "/dev/dri/renderD128", "-i", "/tmp/vp9_test.webm", "-f", "null", "-"] result = subprocess.run(decode_vp9_cmd, capture_output=True, text=True, timeout=10) vp9_vaapi_decode_available = result.returncode == 0 # Clean up test file try: os.remove("/tmp/vp9_test.webm") except: pass except: pass # AV1 is not supported on Intel HD Graphics 630, so we'll skip it av1_decode_available = False # Log results with comprehensive hardware support info hw_support = [] if h264_vaapi_available: hw_support.append("H.264 encode") if hevc_vaapi_available: hw_support.append("HEVC encode") if h264_decode_available: hw_support.append("H.264 decode") if vp9_vaapi_decode_available: hw_support.append("VP9 decode") if hw_support: logger.info(f"šŸš€ Intel hardware acceleration available: {', '.join(hw_support)}") else: logger.info("āš ļø Intel hardware acceleration not available, falling back to software encoding") return (h264_vaapi_available, hevc_vaapi_available, h264_decode_available, av1_decode_available, vp9_vaapi_decode_available) except Exception as e: logger.warning(f"Could not test hardware acceleration support: {e}") return False, False, False, False, False def convert_to_apple_tv_compatible(input_file): """Convert a video to Apple TV 4K compatible format with Intel VAAPI hardware acceleration.""" global conversion_stats logger = logging.getLogger(__name__) base_name = os.path.splitext(input_file)[0] mp4_file = base_name + ".mp4" # If input is already MP4, use a temporary output file to avoid overwriting during conversion temp_output = None if input_file.lower().endswith('.mp4'): temp_output = base_name + "_converted.mp4" final_output = mp4_file mp4_file = temp_output else: final_output = mp4_file # Update conversion stats conversion_stats['active_conversions'] += 1 conversion_stats['current_conversion'] = os.path.basename(input_file) logger.info(f"Converting {os.path.basename(input_file)} to Apple TV compatible format with GPU acceleration...") logger.debug(f"Input file: {input_file}") logger.debug(f"Output file: {mp4_file}") # Test VAAPI support vaapi_h264_available, vaapi_hevc_available, vaapi_decode_available, av1_decode_available, vp9_vaapi_decode_available = test_vaapi_support() # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) # Detect video properties to optimize conversion try: import json probe_cmd = [ FFPROBE_PATH, "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", input_file ] logger.debug(f"Probing video with command: {' '.join(probe_cmd)}") probe_result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) video_info = json.loads(probe_result.stdout) # Find video stream video_stream = next((s for s in video_info['streams'] if s['codec_type'] == 'video'), None) is_4k = video_stream and int(video_stream.get('height', 0)) >= 2160 is_high_res = video_stream and int(video_stream.get('height', 0)) >= 1080 input_codec = video_stream.get('codec_name', 'unknown').lower() if video_stream else 'unknown' if video_stream: height = video_stream.get('height', 'unknown') codec = video_stream.get('codec_name', 'unknown') bitrate = video_stream.get('bit_rate', 'unknown') logger.info(f"Video properties: {height}p, codec: {codec}, bitrate: {bitrate}") except (subprocess.CalledProcessError, json.JSONDecodeError, Exception) as e: logger.warning(f"Could not probe video properties: {e}, using default settings") is_4k = False is_high_res = True input_codec = 'unknown' # Intel VAAPI optimized encoding with hardware decoding based on source quality and hardware support: hwaccel_args = [] filter_args = [] # Determine optimal hardware decoding strategy based on input codec if input_codec == 'av1' and av1_decode_available: # AV1 is not supported on Intel HD Graphics 630, fallback to software filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"Using software decoding with GPU upload for AV1 input (no hardware AV1 support)") elif input_codec in ['vp9', 'vp09'] and vp9_vaapi_decode_available: # Use Intel VAAPI for VP9 hardware decoding hwaccel_args = ["-hwaccel", "vaapi", "-hwaccel_device", "/dev/dri/renderD128"] filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"šŸš€ Using Intel VAAPI hardware decoding for VP9 input") elif input_codec in ['h264', 'avc'] and vaapi_decode_available: # Use Intel VAAPI for H.264 hardware decoding (decode + encode on same device) hwaccel_args = ["-hwaccel", "vaapi", "-hwaccel_device", "/dev/dri/renderD128"] filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"šŸš€ Using Intel VAAPI hardware decoding for H.264 input") else: # Fallback: CPU decode with GPU upload for compatibility filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"Using software decoding with GPU upload for {input_codec} input") if is_4k and vaapi_hevc_available: # 4K content with VAAPI HEVC: Use hardware-accelerated H.265 codec_args = [ *filter_args, "-c:v", "hevc_vaapi", # Intel VAAPI HEVC encoder "-qp", "20", # Quality parameter (lower = better) "-rc_mode", "CQP", # Constant Quality mode ] logger.info("šŸš€ Using Intel VAAPI HEVC hardware acceleration for 4K content") elif (is_4k or is_high_res) and vaapi_h264_available: # 4K/1080p content with VAAPI H.264: Use hardware-accelerated H.264 quality = "20" if is_4k else "22" # Slightly lower quality for 1080p codec_args = [ *filter_args, "-c:v", "h264_vaapi", # Intel VAAPI H.264 encoder "-qp", quality, # Quality parameter "-rc_mode", "CQP", # Constant Quality mode "-profile:v", "high", # H.264 High Profile ] quality_label = "4K" if is_4k else "1080p" logger.info(f"šŸš€ Using Intel VAAPI H.264 hardware acceleration for {quality_label} content") elif is_4k: # 4K content without VAAPI: Fall back to software H.265 codec_args = [ "-c:v", "libx265", # Software H.265 for 4K "-preset", "medium", # Better compression for 4K "-crf", "20", # Higher quality for 4K content "-level", "5.1", # Level 5.1 supports 4K "-pix_fmt", "yuv420p", # Standard compatibility ] hwaccel_args = [] # No hardware acceleration logger.info("āš ļø Using software H.265 encoding for 4K content (VAAPI not available)") elif is_high_res: # 1080p content without VAAPI: Use software H.264 codec_args = [ "-c:v", "libx264", # Software H.264 for 1080p "-profile:v", "high", # High profile for better quality "-preset", "fast", # Faster encoding for 1080p "-crf", "21", # Good quality for 1080p "-level", "4.1", # Level 4.1 supports 1080p ] hwaccel_args = [] # No hardware acceleration logger.info("āš ļø Using software H.264 encoding for 1080p content (VAAPI not available)") else: # Lower resolution: Use software H.264 with faster settings codec_args = [ "-c:v", "libx264", # Software H.264 for lower res "-profile:v", "main", # Main profile for lower res "-preset", "fast", # Fast encoding "-crf", "23", # Standard quality "-level", "3.1", # Level 3.1 for lower res ] hwaccel_args = [] # No hardware acceleration logger.info("Using software H.264 encoding for lower resolution content") # Add progress reporting to FFmpeg command command = [ FFMPEG_PATH, *hwaccel_args, # Hardware acceleration args (decode) "-i", input_file, *codec_args, # Video codec args (encode + filters) "-c:a", "aac", # AAC audio codec "-b:a", "192k", # Higher audio bitrate for better quality "-movflags", "+faststart", # Optimize for streaming/quick playback "-progress", "pipe:1", # Output progress to stdout "-y", # Overwrite output file if exists mp4_file ] try: logger.debug(f"FFmpeg command: {' '.join(command)}") logger.info("Starting conversion... This may take a while for 4K content.") # Get total duration for progress calculation duration_cmd = [FFPROBE_PATH, "-v", "quiet", "-show_entries", "format=duration", "-of", "csv=p=0", input_file] try: duration_result = subprocess.run(duration_cmd, capture_output=True, text=True, check=True) total_duration = float(duration_result.stdout.strip()) except: total_duration = 0 logger.debug("Could not determine video duration for progress tracking") start_time = time.time() # Start FFmpeg with progress monitoring process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True) current_conversion_progress = 0 while True: output = process.stdout.readline() if output == '' and process.poll() is not None: break if output: output = output.strip() # Parse FFmpeg progress output if output.startswith("out_time_us=") and total_duration > 0: try: # Extract time in microseconds time_us = int(output.split("=")[1]) current_time = time_us / 1000000 # Convert to seconds progress = min(100, (current_time / total_duration) * 100) if abs(progress - current_conversion_progress) >= 5: # Update every 5% current_conversion_progress = progress # Update Home Assistant with conversion progress update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'], current_conversion_progress=current_conversion_progress, current_conversion_file=os.path.basename(input_file)[:50] ) logger.info(f"Conversion progress: {current_conversion_progress:.1f}% - {os.path.basename(input_file)}") except (ValueError, IndexError): pass # Wait for process to complete process.wait() end_time = time.time() if process.returncode != 0: stderr_output = process.stderr.read() logger.error(f"FFmpeg failed with return code {process.returncode}: {stderr_output}") raise subprocess.CalledProcessError(process.returncode, command, stderr=stderr_output) conversion_time = end_time - start_time logger.info(f"Conversion completed in {conversion_time:.1f}s: {mp4_file}") # Clear conversion progress update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'], current_conversion_progress=100, current_conversion_file="" ) # Check if output file was created and has reasonable size if os.path.exists(mp4_file): input_size = os.path.getsize(input_file) output_size = os.path.getsize(mp4_file) size_ratio = output_size / input_size if input_size > 0 else 0 logger.info(f"File sizes - Input: {input_size // (1024*1024):.1f}MB, Output: {output_size // (1024*1024):.1f}MB (ratio: {size_ratio:.2f})") # Only replace original if output seems reasonable if output_size > input_size * 0.1: # Output should be at least 10% of input size # If we used a temporary file, replace the original if temp_output: os.remove(input_file) os.rename(mp4_file, final_output) logger.info(f"Original file replaced with converted version: {os.path.basename(final_output)}") else: os.remove(input_file) logger.info(f"Original file deleted: {os.path.basename(input_file)}") else: logger.warning(f"Output file seems too small, keeping original: {os.path.basename(input_file)}") if temp_output and os.path.exists(mp4_file): os.remove(mp4_file) # Clean up failed temporary file # Update completion stats conversion_stats['active_conversions'] -= 1 conversion_stats['completed_conversions'] += 1 # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) return True except subprocess.CalledProcessError as e: conversion_stats['active_conversions'] -= 1 logger.error(f"Conversion failed for {os.path.basename(input_file)}: {e}") if e.stderr: logger.debug(f"FFmpeg error output: {e.stderr}") # Clean up temporary file if it exists if temp_output and os.path.exists(mp4_file): os.remove(mp4_file) logger.debug(f"Cleaned up temporary file: {mp4_file}") # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) return False except Exception as e: conversion_stats['active_conversions'] -= 1 logger.error(f"Unexpected error during conversion of {os.path.basename(input_file)}: {e}") # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) return False def convert_to_apple_tv_compatible_with_target(input_file, target_file): """Convert a video to Apple TV 4K compatible format with Intel VAAPI hardware acceleration and specific target filename.""" global conversion_stats logger = logging.getLogger(__name__) # Update conversion stats conversion_stats['active_conversions'] += 1 conversion_stats['current_conversion'] = os.path.basename(input_file) logger.info(f"Converting {os.path.basename(input_file)} to Apple TV compatible format with GPU acceleration...") logger.info(f"Target filename: {os.path.basename(target_file)}") logger.debug(f"Input file: {input_file}") logger.debug(f"Target file: {target_file}") # Use a temporary output file during conversion to avoid conflicts base_name, ext = os.path.splitext(target_file) temp_output = f"{base_name}.converting{ext}" # Test VAAPI support vaapi_h264_available, vaapi_hevc_available, vaapi_decode_available, av1_decode_available, vp9_vaapi_decode_available = test_vaapi_support() # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) # Detect video properties to optimize conversion try: import json probe_cmd = [ FFPROBE_PATH, "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", input_file ] logger.debug(f"Probing video with command: {' '.join(probe_cmd)}") probe_result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) video_info = json.loads(probe_result.stdout) # Find video stream video_stream = next((s for s in video_info['streams'] if s['codec_type'] == 'video'), None) is_4k = video_stream and int(video_stream.get('height', 0)) >= 2160 is_high_res = video_stream and int(video_stream.get('height', 0)) >= 1080 input_codec = video_stream.get('codec_name', 'unknown').lower() if video_stream else 'unknown' if video_stream: height = video_stream.get('height', 'unknown') codec = video_stream.get('codec_name', 'unknown') bitrate = video_stream.get('bit_rate', 'unknown') logger.info(f"Video properties: {height}p, codec: {codec}, bitrate: {bitrate}") except (subprocess.CalledProcessError, json.JSONDecodeError, Exception) as e: logger.warning(f"Could not probe video properties: {e}, using default settings") is_4k = False is_high_res = True input_codec = 'unknown' # Intel VAAPI optimized encoding with hardware decoding based on source quality and hardware support hwaccel_args = [] filter_args = [] # Determine optimal hardware decoding strategy based on input codec if input_codec == 'av1' and av1_decode_available: # AV1 is not supported on Intel HD Graphics 630, fallback to software filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"Using software decoding with GPU upload for AV1 input (no hardware AV1 support)") elif input_codec in ['vp9', 'vp09'] and vp9_vaapi_decode_available: # Use Intel VAAPI for VP9 hardware decoding hwaccel_args = ["-hwaccel", "vaapi", "-hwaccel_device", "/dev/dri/renderD128"] filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"šŸš€ Using Intel VAAPI hardware decoding for VP9 input") elif input_codec in ['h264', 'avc'] and vaapi_decode_available: # Use Intel VAAPI for H.264 hardware decoding (decode + encode on same device) hwaccel_args = ["-hwaccel", "vaapi", "-hwaccel_device", "/dev/dri/renderD128"] filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"šŸš€ Using Intel VAAPI hardware decoding for H.264 input") else: # Fallback: CPU decode with GPU upload for compatibility filter_args = ["-vaapi_device", "/dev/dri/renderD128", "-vf", "format=nv12,hwupload"] logger.info(f"Using software decoding with GPU upload for {input_codec} input") if is_4k and vaapi_hevc_available: # 4K content with VAAPI HEVC: Use hardware-accelerated H.265 codec_args = [ *filter_args, "-c:v", "hevc_vaapi", # Intel VAAPI HEVC encoder "-qp", "20", # Quality parameter (lower = better) "-rc_mode", "CQP", # Constant Quality mode ] logger.info("šŸš€ Using Intel VAAPI HEVC hardware acceleration for 4K content") elif (is_4k or is_high_res) and vaapi_h264_available: # 4K/1080p content with VAAPI H.264: Use hardware-accelerated H.264 quality = "20" if is_4k else "22" # Slightly lower quality for 1080p codec_args = [ *filter_args, "-c:v", "h264_vaapi", # Intel VAAPI H.264 encoder "-qp", quality, # Quality parameter "-rc_mode", "CQP", # Constant Quality mode "-profile:v", "high", # H.264 High Profile ] quality_label = "4K" if is_4k else "1080p" logger.info(f"šŸš€ Using Intel VAAPI H.264 hardware acceleration for {quality_label} content") elif is_4k: # 4K content without VAAPI: Fall back to software H.265 codec_args = [ "-c:v", "libx265", # Software H.265 for 4K "-preset", "medium", # Better compression for 4K "-crf", "20", # Higher quality for 4K content "-level", "5.1", # Level 5.1 supports 4K "-pix_fmt", "yuv420p", # Standard compatibility ] hwaccel_args = [] # No hardware acceleration logger.info("āš ļø Using software H.265 encoding for 4K content (VAAPI not available)") elif is_high_res: # 1080p content without VAAPI: Use software H.264 codec_args = [ "-c:v", "libx264", # Software H.264 for 1080p "-profile:v", "high", # High profile for better quality "-preset", "fast", # Faster encoding for 1080p "-crf", "21", # Good quality for 1080p "-level", "4.1", # Level 4.1 supports 1080p ] hwaccel_args = [] # No hardware acceleration logger.info("āš ļø Using software H.264 encoding for 1080p content (VAAPI not available)") else: # Lower resolution: Use software H.264 with faster settings codec_args = [ "-c:v", "libx264", # Software H.264 for lower res "-profile:v", "main", # Main profile for lower res "-preset", "fast", # Fast encoding "-crf", "23", # Standard quality "-level", "3.1", # Level 3.1 for lower res ] hwaccel_args = [] # No hardware acceleration logger.info("Using software H.264 encoding for lower resolution content") # Add progress reporting to FFmpeg command command = [ FFMPEG_PATH, *hwaccel_args, # Hardware acceleration args (decode) "-i", input_file, *codec_args, # Video codec args (encode + filters) "-c:a", "aac", # AAC audio codec "-b:a", "192k", # Higher audio bitrate for better quality "-movflags", "+faststart", # Optimize for streaming/quick playback "-progress", "pipe:1", # Output progress to stdout "-y", # Overwrite output file if exists temp_output ] try: logger.debug(f"FFmpeg command: {' '.join(command)}") logger.info("Starting conversion... This may take a while for 4K content.") # Get total duration for progress calculation duration_cmd = [FFPROBE_PATH, "-v", "quiet", "-show_entries", "format=duration", "-of", "csv=p=0", input_file] try: duration_result = subprocess.run(duration_cmd, capture_output=True, text=True, check=True) total_duration = float(duration_result.stdout.strip()) except: total_duration = 0 logger.debug("Could not determine video duration for progress tracking") start_time = time.time() # Start FFmpeg with progress monitoring process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True) current_conversion_progress = 0 while True: output = process.stdout.readline() if output == '' and process.poll() is not None: break if output: output = output.strip() # Parse FFmpeg progress output if output.startswith("out_time_us=") and total_duration > 0: try: # Extract time in microseconds time_us = int(output.split("=")[1]) current_time = time_us / 1000000 # Convert to seconds progress = min(100, (current_time / total_duration) * 100) if abs(progress - current_conversion_progress) >= 5: # Update every 5% current_conversion_progress = progress # Update Home Assistant with conversion progress update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'], current_conversion_progress=current_conversion_progress, current_conversion_file=os.path.basename(input_file)[:50] ) logger.info(f"Conversion progress: {current_conversion_progress:.1f}% - {os.path.basename(input_file)}") except (ValueError, IndexError): pass # Wait for process to complete process.wait() end_time = time.time() if process.returncode != 0: stderr_output = process.stderr.read() logger.error(f"FFmpeg failed with return code {process.returncode}: {stderr_output}") raise subprocess.CalledProcessError(process.returncode, command, stderr=stderr_output) conversion_time = end_time - start_time logger.info(f"Conversion completed in {conversion_time:.1f}s") # Clear conversion progress update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'], current_conversion_progress=100, current_conversion_file="" ) # Check if output file was created and has reasonable size if os.path.exists(temp_output): input_size = os.path.getsize(input_file) output_size = os.path.getsize(temp_output) size_ratio = output_size / input_size if input_size > 0 else 0 logger.info(f"File sizes - Input: {input_size // (1024*1024):.1f}MB, Output: {output_size // (1024*1024):.1f}MB (ratio: {size_ratio:.2f})") # Only replace original if output seems reasonable if output_size > input_size * 0.1: # Move temporary file to final target location os.rename(temp_output, target_file) # Remove the original incompatible file os.remove(input_file) logger.info(f"āœ… Conversion successful: {os.path.basename(input_file)} → {os.path.basename(target_file)}") else: logger.warning(f"Output file seems too small, keeping original: {os.path.basename(input_file)}") if os.path.exists(temp_output): os.remove(temp_output) # Update completion stats conversion_stats['active_conversions'] -= 1 conversion_stats['completed_conversions'] += 1 # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) return True except subprocess.CalledProcessError as e: conversion_stats['active_conversions'] -= 1 logger.error(f"Conversion failed for {os.path.basename(input_file)}: {e}") if e.stderr: logger.debug(f"FFmpeg error output: {e.stderr}") # Clean up temporary file if it exists if os.path.exists(temp_output): os.remove(temp_output) logger.debug(f"Cleaned up temporary file: {temp_output}") # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) return False except Exception as e: conversion_stats['active_conversions'] -= 1 logger.error(f"Unexpected error during conversion of {os.path.basename(input_file)}: {e}") # Clean up temporary file if it exists if os.path.exists(temp_output): os.remove(temp_output) logger.debug(f"Cleaned up temporary file: {temp_output}") # Update Home Assistant update_homeassistant_stats( conversion_queue=conversion_queue.qsize(), active_conversions=conversion_stats['active_conversions'] ) return False # Temporarily commented out video conversion functionality # def convert_to_mp4(input_file): # """Convert a video to mp4 and delete the original file.""" # mp4_file = os.path.splitext(input_file)[0] + ".mp4" # print(f"Converting {input_file} to {mp4_file}...") # command = ["ffmpeg", "-i", input_file, "-c:v", "libx264", "-preset", "fast", "-c:a", "aac", mp4_file] # subprocess.run(command, check=True) # print(f"Conversion completed: {mp4_file}") # if os.path.exists(mp4_file): # os.remove(input_file) # print(f"Original file deleted: {input_file}") def create_nfo_file(video_data, file_path): """Create an Emby-compatible NFO file.""" logger = logging.getLogger(__name__) try: nfo_content = f""" {video_data['title']} {video_data['description']} {video_data['published_at'][:4]} {video_data['published_at']} {video_data['channel_title']} """ with open(file_path, "w", encoding="utf-8") as nfo_file: nfo_file.write(nfo_content) logger.debug(f"Created NFO file: {file_path}") except Exception as e: logger.error(f"Failed to create NFO file {file_path}: {e}") raise def check_exit_time(): """Check if the current time is past the exit time.""" logger = logging.getLogger(__name__) now = datetime.now().time() if now >= EXIT_TIME: logger.info(f"Current time ({now.strftime('%H:%M:%S')}) exceeds exit time ({EXIT_TIME.strftime('%H:%M:%S')}). Terminating script.") sys.exit(0) def delete_old_videos(folder, days=5): """Delete video files and their folders older than `days`.""" logger = logging.getLogger(__name__) logger.info(f"Starting cleanup of videos older than {days} days in {folder}") if not os.path.exists(folder): logger.warning(f"Output folder does not exist: {folder}") return now = time.time() cutoff = now - (days * 86400) # Convert days to seconds deleted_files = 0 deleted_dirs = 0 total_size_freed = 0 try: for root, dirs, files in os.walk(folder, topdown=False): # Delete files for file in files: file_path = os.path.join(root, file) try: if os.path.isfile(file_path) and os.path.getmtime(file_path) < cutoff: file_size = os.path.getsize(file_path) os.remove(file_path) deleted_files += 1 total_size_freed += file_size logger.debug(f"Deleted file: {file_path} ({file_size // (1024*1024):.1f}MB)") except (OSError, IOError) as e: logger.error(f"Failed to delete file {file_path}: {e}") # Remove empty directories for dir_name in dirs: dir_path = os.path.join(root, dir_name) try: if os.path.exists(dir_path) and not os.listdir(dir_path): # Check if the directory is empty os.rmdir(dir_path) deleted_dirs += 1 logger.debug(f"Deleted empty directory: {dir_path}") except (OSError, IOError) as e: logger.error(f"Failed to delete directory {dir_path}: {e}") if deleted_files > 0 or deleted_dirs > 0: size_mb = total_size_freed / (1024 * 1024) logger.info(f"Cleanup completed: {deleted_files} files, {deleted_dirs} directories deleted, {size_mb:.1f}MB freed") else: logger.info("No old files found to delete") except Exception as e: logger.error(f"Error during cleanup: {e}") def video_exists(output_path, video_id): """Check if a video already exists in the directory.""" return os.path.exists(os.path.join(output_path, video_id)) def check_exit_time(): """Check if the current time is past the exit time.""" now = datetime.now().time() if now >= EXIT_TIME: print("Current time exceeds exit time. Terminating script.") sys.exit(0) def main(): global download_stats logger = logging.getLogger(__name__) logger.info("Starting main download process") # Initialize Home Assistant integration if HOMEASSISTANT_ENABLED: logger.info("Initializing Home Assistant integration...") ha_integration = init_homeassistant_integration() if ha_integration: logger.info("Home Assistant integration started successfully") else: logger.warning("Home Assistant integration failed to start") # Start the conversion queue worker logger.info("Starting video conversion queue system...") logger.info(f"Max concurrent conversions: {MAX_CONCURRENT_CONVERSIONS}") # Start background conversion worker conversion_worker = threading.Thread(target=process_conversion_queue, daemon=True, name="ConversionQueueWorker") conversion_worker.start() logger.debug("Conversion queue worker thread started") # Update yt-dlp try: update_yt_dlp() except Exception as e: logger.error(f"Failed to update yt-dlp: {e}") # Continue anyway, existing version might work # Authenticate YouTube API logger.info("Authenticating with YouTube API...") try: youtube = authenticate_youtube_api() logger.info("YouTube API authentication successful") except Exception as e: logger.error(f"YouTube API authentication failed: {e}") raise # Ensure output directory exists if not os.path.exists(OUTPUT_DIR): logger.info(f"Creating output directory: {OUTPUT_DIR}") os.makedirs(OUTPUT_DIR) else: logger.debug(f"Output directory exists: {OUTPUT_DIR}") # Clean up old videos logger.info(f"Cleaning up videos older than {DAYS_TO_KEEP} days...") delete_old_videos(OUTPUT_DIR, days=DAYS_TO_KEEP) # Fetch subscribed channels logger.info("Fetching subscribed channels...") try: channels = get_subscribed_channels(youtube) logger.info(f"Successfully fetched {len(channels)} subscribed channels") except Exception as e: logger.error(f"Failed to fetch subscribed channels: {e}") channels = [] # Count total videos for progress tracking logger.info("Counting total videos to download...") total_videos = 0 all_videos = [] # Collect all videos first for counting for i, channel_id in enumerate(channels): try: logger.debug(f"Processing channel {i+1}/{len(channels)}: {channel_id}") videos = get_recent_videos(youtube, channel_id) for video in videos: video_dir = os.path.join(OUTPUT_DIR, video["video_id"]) if not os.path.exists(video_dir): all_videos.append(('subscription', video)) total_videos += 1 else: logger.debug(f"Video already exists, skipping: {video['title']}") except Exception as e: logger.error(f"Error processing channel {channel_id}: {e}") # Add popular videos for region_code in REGION_CODES: try: logger.debug(f"Fetching popular videos from region: {region_code}") popular_videos = get_popular_videos(youtube, region_code) for video in popular_videos: video_dir = os.path.join(OUTPUT_DIR, video["video_id"]) if not os.path.exists(video_dir): all_videos.append(('popular', video)) total_videos += 1 else: logger.debug(f"Popular video already exists, skipping: {video['title']}") except Exception as e: logger.error(f"Error fetching popular videos from region {region_code}: {e}") # Initialize download stats download_stats['total_videos'] = total_videos download_stats['completed_videos'] = 0 logger.info(f"Total videos to download: {total_videos}") if total_videos == 0: logger.info("No new videos found to download") return # Update initial Home Assistant stats update_homeassistant_stats( download_queue=total_videos, conversion_queue=0, download_progress=0, active_downloads=0, active_conversions=0 ) # Process all videos for i, (video_type, video) in enumerate(all_videos): # Check the exit time before processing each video check_exit_time() logger.info(f"Processing video {i+1}/{total_videos} ({video_type}): {video['title']}") video_dir = os.path.join(OUTPUT_DIR, video["video_id"]) try: # Create directory os.makedirs(video_dir, exist_ok=True) logger.debug(f"Created/verified directory: {video_dir}") # Download the video if not download_video(video["video_id"], video_dir, video["title"]): logger.warning(f"Skipping video {video['video_id']} due to download error") continue # Skip this video if download failed # Create the NFO file nfo_file_path = os.path.join(video_dir, "movie.nfo") try: create_nfo_file(video, nfo_file_path) logger.debug(f"Created NFO file: {nfo_file_path}") except Exception as e: logger.error(f"Failed to create NFO file: {e}") # Update completed videos count download_stats['completed_videos'] += 1 progress = (download_stats['completed_videos'] / download_stats['total_videos']) * 100 # Update Home Assistant stats update_homeassistant_stats( download_queue=max(0, download_stats['total_videos'] - download_stats['completed_videos']), download_progress=round(progress, 1) ) logger.debug(f"Progress: {download_stats['completed_videos']}/{download_stats['total_videos']} ({progress:.1f}%)") except Exception as e: logger.error(f"Error processing video {video['title']}: {e}") continue logger.info("All videos processed!") logger.info(f"Final stats: {download_stats['completed_videos']}/{download_stats['total_videos']} videos completed") # Wait for all conversions to complete logger.info("Waiting for all conversions to complete...") conversion_executor.shutdown(wait=True) logger.info("All conversions completed!") # Final Home Assistant update update_homeassistant_stats( download_queue=0, conversion_queue=0, download_progress=100, active_downloads=0, active_conversions=0 ) logger.info("Main download process completed successfully") logger.info("=" * 80) def schedule_downloads(): """Schedule downloads to run at specified times""" logger = logging.getLogger(__name__) # Configuration for scheduling START_TIME = dt_time(0, 0) # 00:00 END_TIME = dt_time(7, 0) # 07:00 logger.info("Starting YouTube Predownloader Service with scheduling") logger.info(f"Service will run between {START_TIME.strftime('%H:%M')} and {END_TIME.strftime('%H:%M')} daily") while True: current_time = datetime.now().time() # Check if we're in the allowed time window if START_TIME <= current_time <= END_TIME: try: set_service_running() logger.info(f"Starting download session at {datetime.now().strftime('%H:%M:%S')}") # Run the main download process main() set_service_completed() logger.info(f"Download session completed at {datetime.now().strftime('%H:%M:%S')}") # Calculate next run time (next day at START_TIME) next_run = datetime.now().replace(hour=START_TIME.hour, minute=START_TIME.minute, second=0, microsecond=0) if current_time > START_TIME: next_run = next_run.replace(day=next_run.day + 1) set_service_waiting(next_run.isoformat()) # Sleep until next day sleep_seconds = (next_run - datetime.now()).total_seconds() logger.info(f"Sleeping until next run at {next_run.strftime('%Y-%m-%d %H:%M:%S')} ({sleep_seconds:.0f} seconds)") time.sleep(sleep_seconds) except Exception as e: logger.error(f"Error during download session: {e}", exc_info=True) set_service_error() # Sleep for 1 hour before retrying logger.info("Sleeping for 1 hour before retrying") time.sleep(3600) else: # We're outside the allowed time window if current_time < START_TIME: # Before start time today next_run = datetime.now().replace(hour=START_TIME.hour, minute=START_TIME.minute, second=0, microsecond=0) else: # After end time today, schedule for tomorrow next_run = datetime.now().replace(hour=START_TIME.hour, minute=START_TIME.minute, second=0, microsecond=0) next_run = next_run.replace(day=next_run.day + 1) set_service_waiting(next_run.isoformat()) # Sleep until start time sleep_seconds = (next_run - datetime.now()).total_seconds() logger.info(f"Waiting until {next_run.strftime('%Y-%m-%d %H:%M:%S')} to start downloads ({sleep_seconds:.0f} seconds)") time.sleep(min(sleep_seconds, 3600)) # Check every hour maximum # Fetch and process most popular videos worldwide (from multiple regions) def fetch_popular_videos_old(): """Old function kept for reference - this logic is now integrated into main()""" for region_code in REGION_CODES: print(f"Fetching the most popular videos from {region_code}...") popular_videos = get_popular_videos(youtube, region_code) for video in popular_videos: # Check the exit time before processing each video check_exit_time() print(f"Processing popular video: {video['title']} ({region_code})") video_dir = os.path.join(OUTPUT_DIR, video["video_id"]) # Check if the video already exists if os.path.exists(video_dir): print(f"Skipping download, already exists: {video['title']} ({video['video_id']})") continue os.makedirs(video_dir, exist_ok=True) # Download the popular video if not download_video(video["video_id"], video_dir): print(f"Skipping popular video {video['video_id']} due to download error.") continue # Skip this video if download failed # Check if conversion is needed (VP9/AV1 codecs or non-MP4 files) needs_conversion = False conversion_file = None for file in os.listdir(video_dir): if file.endswith(('.mp4', '.mkv', '.webm', '.m4v')): file_path = os.path.join(video_dir, file) try: # Quick check for problematic codecs probe_cmd = [FFPROBE_PATH, "-v", "quiet", "-select_streams", "v:0", "-show_entries", "stream=codec_name", "-of", "csv=p=0", file_path] result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) codec = result.stdout.strip().lower() if codec in ['vp9', 'av01', 'av1'] or not file.endswith('.mp4'): needs_conversion = True conversion_file = file_path print(f"Found {codec} codec - will convert for Apple TV compatibility") break except subprocess.CalledProcessError: # If we can't probe, assume it might need conversion if not MP4 if not file.endswith('.mp4'): needs_conversion = True conversion_file = file_path break if needs_conversion and conversion_file: # Determine priority based on video resolution try: probe_cmd = [FFPROBE_PATH, "-v", "quiet", "-select_streams", "v:0", "-show_entries", "stream=height", "-of", "csv=p=0", conversion_file] result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) height = int(result.stdout.strip()) if result.stdout.strip().isdigit() else 1080 priority = 2 if height >= 2160 else 1 # 4K gets higher priority except: priority = 1 # Default priority add_to_conversion_queue(conversion_file, priority) print_queue_status() # Create the NFO file nfo_file_path = os.path.join(video_dir, "movie.nfo") create_nfo_file(video, nfo_file_path) print("All videos processed successfully!") # Final conversion queue status print("\nšŸŽ¬ Final Conversion Queue Status:") print_queue_status() # Final Home Assistant update update_homeassistant_stats( download_queue=0, conversion_queue=conversion_queue.qsize(), download_progress=100, active_downloads=0, active_conversions=conversion_stats['active_conversions'] ) queued_items = conversion_queue.qsize() if queued_items > 0: print(f"ā³ {queued_items} videos still in conversion queue") print("šŸ’” Conversions will continue in the background") print("šŸ’” You can monitor progress in Home Assistant") else: print("āœ… All conversions completed!") print("šŸ  Home Assistant integration will continue running in background") if __name__ == "__main__": import sys # Check for debug mode debug_mode = "--debug" in sys.argv if debug_mode: sys.argv.remove("--debug") # Set debug logging LOG_LEVEL = logging.DEBUG logger = setup_logging() logger.info("Debug mode enabled - verbose logging activated") logger = logging.getLogger(__name__) # Check if we should run as a service with scheduling if len(sys.argv) > 1 and sys.argv[1] == "--service": logger.info("Starting YouTube Predownloader Service with scheduling") try: schedule_downloads() except KeyboardInterrupt: logger.info("Service interrupted by user") except Exception as e: logger.error(f"Service error: {e}", exc_info=True) else: logger.info("Running YouTube Predownloader once") try: main() except KeyboardInterrupt: logger.info("Process interrupted by user") except Exception as e: logger.error(f"Process error: {e}", exc_info=True) finally: logger.info("Process finished")