Dovodi_Backend/utils/youtube.py

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