import hashlib import logging import re from typing import Optional from django.core.cache import cache logger = logging.getLogger(__name__) # Cache duration for YouTube stream URLs (3 hours = 10,800 seconds) YOUTUBE_STREAM_CACHE_TTL = 3 * 3600 def normalize_youtube_url(url: str) -> str: """ Extract YouTube 11-character video ID and return clean canonical URL. Strips tracking parameters (like ?si=...) and converts short/embed links to standard watch URL. """ if not url: return url match = re.search(r'(?:v=|\/|embed\/|v\/|^)([0-9A-Za-z_-]{11})(?:[\?&/]|$)', url) if match: video_id = match.group(1) return f"https://www.youtube.com/watch?v={video_id}" return url def _get_cache_key(youtube_url: str) -> str: url_hash = hashlib.md5(youtube_url.strip().encode('utf-8')).hexdigest() return f"yt_stream_url_{url_hash}" def get_youtube_stream_url(youtube_url: str) -> Optional[str]: """ Extract direct playable stream URL from a YouTube video URL using yt-dlp. Caches the extracted URL in Django cache for 3 hours to minimize calls to YouTube. Returns direct stream URL string if successful, or None if extraction fails. """ # Temporarily disabled - return None without resolving stream URL return None # if not youtube_url or not isinstance(youtube_url, str): # return None # # youtube_url = youtube_url.strip() # if not youtube_url: # return None # # cache_key = _get_cache_key(youtube_url) # cached_url = cache.get(cache_key) # if cached_url: # return cached_url # # clean_url = normalize_youtube_url(youtube_url) # # try: # import yt_dlp # # ydl_opts = { # 'format': 'best[ext=mp4]/bestvideo[ext=mp4]+bestaudio[ext=m4a]/best', # 'quiet': True, # 'no_warnings': True, # 'skip_download': True, # 'nocheckcertificate': True, # 'extractor_args': { # 'youtube': { # 'player_client': ['android', 'ios', 'web', 'mweb'], # } # } # } # # with yt_dlp.YoutubeDL(ydl_opts) as ydl: # info = ydl.extract_info(clean_url, download=False) # if not info: # return None # # if 'entries' in info: # info = info['entries'][0] # # stream_url = info.get('url') # if stream_url: # cache.set(cache_key, stream_url, YOUTUBE_STREAM_CACHE_TTL) # return stream_url # except Exception as e: # logger.error(f"Failed to extract YouTube stream URL for '{youtube_url}': {e}") # # return None def clear_youtube_stream_cache(youtube_url: str) -> None: """ Clear cached stream URL for a given YouTube URL. """ if youtube_url: cache_key = _get_cache_key(youtube_url) cache.delete(cache_key) def seconds_to_time(seconds) -> Optional[object]: """ Convert seconds into datetime.time(hours, minutes, seconds). """ if seconds is None: return None try: import datetime total_seconds = int(seconds) hours = total_seconds // 3600 minutes = (total_seconds % 3600) // 60 secs = total_seconds % 60 return datetime.time(min(hours, 23), minutes, secs) except Exception: return None def get_youtube_video_metadata(youtube_url: str) -> Optional[dict]: """ Extract metadata (duration, thumbnail URL, title, etc.) from a YouTube video URL using yt-dlp. Falls back to regex-based video ID and thumbnail URL if yt-dlp extraction encounters network/restriction issues. """ if not youtube_url or not isinstance(youtube_url, str): return None clean_url = normalize_youtube_url(youtube_url.strip()) match = re.search(r'(?:v=|\/|embed\/|v\/|^)([0-9A-Za-z_-]{11})(?:[\?&/]|$)', clean_url) video_id = match.group(1) if match else None fallback_thumbnail = f"https://img.youtube.com/vi/{video_id}/maxresdefault.jpg" if video_id else None try: import yt_dlp ydl_opts = { 'quiet': True, 'no_warnings': True, 'skip_download': True, 'nocheckcertificate': True, 'extractor_args': { 'youtube': { 'player_client': ['android', 'ios', 'web', 'mweb'], } } } with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(clean_url, download=False) if info: if 'entries' in info: info = info['entries'][0] duration_secs = info.get('duration') thumbnail = info.get('thumbnail') or fallback_thumbnail title = info.get('title') duration_time = seconds_to_time(duration_secs) if duration_secs is not None else None return { 'video_id': video_id, 'duration_seconds': duration_secs, 'duration_time': duration_time, 'thumbnail_url': thumbnail, 'title': title, } except Exception as e: logger.error(f"Failed to extract YouTube metadata for '{youtube_url}': {e}") if video_id: return { 'video_id': video_id, 'duration_seconds': None, 'duration_time': None, 'thumbnail_url': fallback_thumbnail, 'title': None, } return None def download_youtube_thumbnail(thumbnail_url: str) -> Optional[bytes]: """ Download image data from a thumbnail URL. Attempts maxresdefault first, falls back to hqdefault on 404 or failure. """ if not thumbnail_url: return None import urllib.request def _fetch_bytes(url: str) -> Optional[bytes]: try: req = urllib.request.Request( url, headers={'User-Agent': 'Mozilla/5.0'} ) with urllib.request.urlopen(req, timeout=10) as response: if response.status == 200: return response.read() except Exception: pass return None data = _fetch_bytes(thumbnail_url) if data: return data if 'maxresdefault' in thumbnail_url: hq_url = thumbnail_url.replace('maxresdefault', 'hqdefault') data = _fetch_bytes(hq_url) if data: return data return None