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. """ 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