import logging import yt_dlp logger = logging.getLogger(__name__) def extract_audio_stream(video_id: str) -> dict | None: """Extract playable audio stream URL from a YouTube video using yt-dlp.""" url = f"https://www.youtube.com/watch?v={video_id}" ydl_opts = { "format": "bestaudio/best", "quiet": True, "no_warnings": True, "extract_flat": False, "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", } try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=False) if not info: logger.error("No info extracted for video %s", video_id) return None formats = info.get("formats", []) if not formats: logger.error("No formats available for video %s", video_id) return None audio_url = None stream_type = None for fmt in formats: protocol = fmt.get("protocol", "") has_audio = fmt.get("acodec", "none") != "none" if not has_audio: continue if protocol.startswith("m3u8"): audio_url = fmt.get("url") stream_type = "hls" break if not audio_url: for fmt in formats: protocol = fmt.get("protocol", "") fmt_note = fmt.get("format_note", "") has_audio = fmt.get("acodec", "none") != "none" if not has_audio: continue if protocol.startswith("https") and "audio" in fmt_note.lower(): audio_url = fmt.get("url") stream_type = "direct" break elif protocol.startswith("https") and has_audio: audio_url = fmt.get("url") stream_type = "direct" if not audio_url: logger.error("No audio stream found for video %s", video_id) return None return { "videoId": video_id, "url": audio_url, "streamType": stream_type, "title": info.get("title", "Unknown"), "channel": info.get("channel", "Unknown"), "duration": info.get("duration"), "isLive": info.get("live_status") == "live", } except yt_dlp.utils.DownloadError as e: logger.error("yt-dlp error for video %s: %s", video_id, e) return None except Exception as e: logger.error("Unexpected error for video %s: %s", video_id, e) return None