import os import time import datetime from datetime import datetime, time as dt_time import pickle import threading import queue 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 # Conversion Queue Settings MAX_CONCURRENT_CONVERSIONS = 2 # Maximum number of simultaneous video conversions CONVERSION_QUEUE_SIZE = 50 # Maximum items in conversion queue # 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.""" print("Updating yt-dlp to the latest version...") subprocess.run(["pip", "install", "--upgrade", "yt-dlp"], check=True) print("yt-dlp updated successfully!") def authenticate_youtube_api(): """Authenticate using our enhanced auth system (tries non-interactive first).""" print("Attempting non-interactive authentication...") # Try non-interactive authentication first creds = get_token_non_interactive() if not creds: print("Non-interactive authentication failed.") print("You need to run: python3 auth.py --interactive") print("Or set up service account authentication.") raise Exception("Authentication failed. No valid credentials available.") print("Authentication successful!") # Build the YouTube API client youtube = build("youtube", "v3", credentials=creds) 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.""" channels = [] request = youtube.subscriptions().list( part="snippet", mine=True, maxResults=50 ) while request: response = request.execute() for item in response["items"]: channels.append(item["snippet"]["resourceId"]["channelId"]) request = youtube.subscriptions().list_next(request, response) return channels def get_recent_videos(youtube, channel_id): """Fetch videos from a channel published in the last `DAYS_BACK` days.""" recent_videos = [] published_after = (datetime.datetime.now() - datetime.timedelta(days=DAYS_BACK)).isoformat("T") + "Z" 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"]: recent_videos.append({ "video_id": item["id"]["videoId"], "title": item["snippet"]["title"], "description": item["snippet"]["description"], "published_at": item["snippet"]["publishedAt"], "channel_title": item["snippet"]["channelTitle"] }) return recent_videos def get_popular_videos(youtube, region_code): """Fetch the most popular videos for a given region.""" popular_videos = [] 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"]: popular_videos.append({ "video_id": item["id"], "title": item["snippet"]["title"], "description": item["snippet"]["description"], "published_at": item["snippet"]["publishedAt"], "channel_title": item["snippet"]["channelTitle"] }) return popular_videos 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 try: # Update stats download_stats['active_downloads'] += 1 download_stats['current_video'] = video_title[:50] + "..." if len(video_title) > 50 else video_title # 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'] ) # Quality-priority format selection: command = [ "yt-dlp", "-f", "(bv[vcodec~='^(avc|h264|hevc|h265)'] / bv[height>=2160] / bv[height>=1080] / best) + (ba[acodec~='^(aac|mp4a)'] / ba)", "--merge-output-format", "mp4", "--no-mtime", "--newline", # Better for parsing progress f"https://www.youtube.com/watch?v={video_id}", "-o", os.path.join(output_path, "%(title)s.%(ext)s") ] result = subprocess.run(command, check=True, capture_output=True, text=True) # Update completed count download_stats['completed_videos'] += 1 download_stats['active_downloads'] -= 1 print(f"Successfully downloaded video {video_id} (Quality priority - will convert VP9/AV1 if needed)") # 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 print(f"Error downloading video {video_id}: {e}") return False def add_to_conversion_queue(input_file, priority=1): """Add a video file to the conversion queue with priority.""" try: # Higher number = higher priority (4K gets priority 2, others get 1) conversion_queue.put((priority, input_file), timeout=1) print(f"Added to conversion queue: {os.path.basename(input_file)} (priority: {priority})") return True except queue.Full: print(f"āš ļø Conversion queue is full! Skipping: {os.path.basename(input_file)}") 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.""" status = get_queue_status() print(f"šŸŽ¬ Conversion Status: {status['active']}/{status['max_concurrent']} active, {status['queued']} queued") def process_conversion_queue(): """Background worker to process the conversion queue.""" while True: try: # Get item from queue (blocks until available) priority, input_file = conversion_queue.get(timeout=30) # Check if file still exists if not os.path.exists(input_file): print(f"āš ļø File no longer exists, skipping conversion: {input_file}") conversion_queue.task_done() continue print(f"šŸ”„ Starting conversion (priority {priority}): {os.path.basename(input_file)}") # Perform the conversion success = convert_to_apple_tv_compatible(input_file) if success: print(f"āœ… Conversion completed: {os.path.basename(input_file)}") else: print(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 continue except Exception as e: print(f"āŒ Conversion queue error: {e}") conversion_queue.task_done() def convert_to_apple_tv_compatible(input_file): """Convert a video to Apple TV 4K compatible format, optimized by resolution.""" global conversion_stats base_name = os.path.splitext(input_file)[0] mp4_file = base_name + ".mp4" # Update conversion stats conversion_stats['active_conversions'] += 1 conversion_stats['current_conversion'] = os.path.basename(input_file) print(f"Converting {input_file} to Apple TV compatible format...") # 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 ] 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 print(f"Video resolution: {video_stream.get('height', 'unknown')}p") except (subprocess.CalledProcessError, json.JSONDecodeError, Exception) as e: print(f"Could not probe video properties: {e}, using default settings") is_4k = False is_high_res = True # Apple TV 4K optimized encoding based on source quality: if is_4k: # 4K content: Use H.265 for better compression at high resolution codec_args = [ "-c:v", "libx265", # H.265 for 4K (better compression) "-preset", "medium", # Better compression for 4K "-crf", "20", # Higher quality for 4K content "-level", "5.1", # Level 5.1 supports 4K "-pix_fmt", "yuv420p", # Ensure compatibility ] print("Using H.265 encoding for 4K content") elif is_high_res: # 1080p content: Use H.264 for maximum compatibility codec_args = [ "-c:v", "libx264", # H.264 for 1080p (max compatibility) "-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 at high bitrates ] print("Using H.264 encoding for 1080p content") else: # Lower resolution: Use H.264 with faster settings codec_args = [ "-c:v", "libx264", # 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 ] print("Using H.264 encoding for lower resolution content") command = [ FFMPEG_PATH, "-i", input_file, *codec_args, "-c:a", "aac", # AAC audio codec "-b:a", "192k", # Higher audio bitrate for better quality "-movflags", "+faststart", # Optimize for streaming/quick playback "-y", # Overwrite output file if exists mp4_file ] try: print("Starting conversion... This may take a while for 4K content.") subprocess.run(command, check=True) print(f"Conversion completed: {mp4_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'] ) if os.path.exists(mp4_file): os.remove(input_file) print(f"Original file deleted: {input_file}") return True except subprocess.CalledProcessError as e: conversion_stats['active_conversions'] -= 1 print(f"Conversion failed for {input_file}: {e}") # 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.""" 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) def delete_old_videos(folder, days=5): """Delete video files and their folders older than `days`.""" now = time.time() cutoff = now - (days * 86400) # Convert days to seconds for root, dirs, files in os.walk(folder, topdown=False): # Delete files for file in files: file_path = os.path.join(root, file) if os.path.isfile(file_path): if os.path.getmtime(file_path) < cutoff: print(f"Deleting file: {file_path}") os.remove(file_path) # Remove empty directories for dir in dirs: dir_path = os.path.join(root, dir) if not os.listdir(dir_path): # Check if the directory is empty print(f"Deleting folder: {dir_path}") os.rmdir(dir_path) 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 # Initialize Home Assistant integration if HOMEASSISTANT_ENABLED: print("šŸ  Initializing Home Assistant integration...") ha_integration = init_homeassistant_integration() if ha_integration: print("āœ… Home Assistant integration started") else: print("āš ļø Home Assistant integration failed to start") # Start the conversion queue worker print("šŸŽ¬ Starting video conversion queue system...") print(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() # Update yt-dlp update_yt_dlp() # Authenticate YouTube API print("Authenticating with YouTube API...") youtube = authenticate_youtube_api() if not os.path.exists(OUTPUT_DIR): os.makedirs(OUTPUT_DIR) # Clean up old videos print(f"Deleting videos older than {DAYS_TO_KEEP} days in {OUTPUT_DIR}...") delete_old_videos(OUTPUT_DIR, days=DAYS_TO_KEEP) # Fetch subscribed channels print("Fetching subscribed channels...") channels = get_subscribed_channels(youtube) # Count total videos for progress tracking print("šŸ“Š Counting total videos to download...") total_videos = 0 all_videos = [] # Collect all videos first for counting for channel_id in channels: 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 # Add popular videos for region_code in REGION_CODES: 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 # Initialize download stats download_stats['total_videos'] = total_videos download_stats['completed_videos'] = 0 print(f"šŸ“ˆ Total videos to download: {total_videos}") # 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 video_type, video in all_videos: # Check the exit time before processing each video check_exit_time() print(f"Processing {video_type} video: {video['title']}") video_dir = os.path.join(OUTPUT_DIR, video["video_id"]) # Create directory os.makedirs(video_dir, exist_ok=True) # Download the video if not download_video(video["video_id"], video_dir, video["title"]): print(f"Skipping 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) # 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) ) print("āœ… All videos processed!") print(f"šŸ“Š Final stats: {download_stats['completed_videos']}/{download_stats['total_videos']} videos completed") # Wait for all conversions to complete print("ā³ Waiting for all conversions to complete...") conversion_executor.shutdown(wait=True) print("āœ… 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 ) def schedule_downloads(): """Schedule downloads to run at specified times""" # Configuration for scheduling START_TIME = dt_time(0, 0) # 00:00 END_TIME = dt_time(7, 0) # 07:00 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() print(f"šŸš€ Starting download session at {datetime.now().strftime('%H:%M:%S')}") # Run the main download process main() set_service_completed() print(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() print(f"😓 Sleeping until next run at {next_run.strftime('%Y-%m-%d %H:%M:%S')}") time.sleep(sleep_seconds) except Exception as e: print(f"āŒ Error during download session: {e}") set_service_error() # Sleep 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() print(f"ā° Waiting until {next_run.strftime('%Y-%m-%d %H:%M:%S')} to start downloads") 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 if we should run as a service with scheduling if len(sys.argv) > 1 and sys.argv[1] == "--service": print("šŸ”„ Starting YouTube Predownloader Service with scheduling") schedule_downloads() else: print("ā–¶ļø Running YouTube Predownloader once") main()