You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
94 lines
2.7 KiB
94 lines
2.7 KiB
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)
|