"""Category D: Utility Tools
Additional utility functions for audio extraction and other helper operations.
"""
import asyncio
import yt_dlp
from ..middleware.rate_limiter import rate_limiter
def extract_video_id(url_or_id: str) -> str:
"""Extract video ID from URL or return as-is if already an ID."""
if "youtube.com" in url_or_id or "youtu.be" in url_or_id:
if "v=" in url_or_id:
return url_or_id.split("v=")[1].split("&")[0]
elif "youtu.be/" in url_or_id:
return url_or_id.split("youtu.be/")[1].split("?")[0]
return url_or_id
@rate_limiter
async def get_audio_stream_url(video_id: str) -> str:
"""
Get the direct audio stream URL for a YouTube video.
Returns a playable audio URL (stream link) without downloading the file.
Useful for audio-only consumption or integration with media players.
Args:
video_id: YouTube video ID or URL
Returns:
Direct audio stream URL or download instruction
"""
video_id = extract_video_id(video_id)
url = f"https://www.youtube.com/watch?v={video_id}"
ydl_opts = {
'format': 'bestaudio/best',
'quiet': True,
'no_warnings': True,
}
def _fetch():
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(url, download=False)
return info
try:
info = await asyncio.to_thread(_fetch)
# Get the audio URL
audio_url = info.get('url', '')
title = info.get('title', 'N/A')
duration = info.get('duration', 0)
output = f"# Audio Stream for: {title}\n\n"
output += f"**Video URL**: https://www.youtube.com/watch?v={video_id}\n"
output += f"**Duration**: {duration} seconds\n\n"
output += "## Direct Audio Stream URL\n"
output += f"```\n{audio_url}\n```\n\n"
output += "## Download Command (yt-dlp)\n"
output += f"```bash\n"
output += f"yt-dlp -f bestaudio -o \"audio.%(ext)s\" {url}\n"
output += f"```\n\n"
output += "⚠️ **Note**: Stream URLs are temporary and expire after a short period.\n"
return output
except Exception as e:
return f"❌ Error getting audio stream for {video_id}: {str(e)}"
