lofi-app/src/modules/stream_extractor.py
opencode 55244b5027 fix: resolve 35 open issues across backend, Android, frontend, Docker
Backend (src/main.py):
- Add shared httpx.Client singleton for connection reuse (#24)
- Add cache eviction for expired entries (#23)
- Expand CORS to allow POST and OPTIONS (#25)
- Preserve HLS tags (EXT-X-TARGETDURATION, EXT-X-MAP, etc) in rewrite (#12)
- Replace urllib with httpx.stream for direct streaming

Backend (src/modules/stream_extractor.py):
- Use extract_flat='in_playlist' for reduced latency (#35)

Tests (tests/integration/test_api.py):
- Parameterize channel count assertion against CHANNELS (#20)

Android (Channel.kt):
- Change mutable vars to immutable vals, use copy() pattern (#19)

Android (ServerApi.kt):
- Fix JSON parsing for bare array response (#21)
- Close response body on error path (#29)

Android (YouTubeExtractor.kt):
- Add ConnectionPool config (10 idle, 30s keepalive) (#28)
- Tighten findHlsUrl to require audio codec (#26)

Android (AudioPlayer.kt):
- Remove false error on STATE_READY + !isPlaying (#33)

Android (MainActivity.kt):
- Reuse fragments via findFragmentByTag + show/hide (#18)
- Move AudioPlayer.release() to Activity.onDestroy (#17)

Android (LofiViewModel.kt):
- Wrap YouTubeExtractor calls in withContext(Dispatchers.IO) (#16)
- Use immutable channel copies in discoverAllChannels (#19)
- Remove audioPlayer.release() from onCleared (#17)

Android (ChannelListFragment.kt):
- Tie swipe refresh to isDiscovering LiveData (#32)

Android (AndroidManifest.xml):
- Set allowBackup=false (#15)

Android (Preferences.kt):
- Remove hardcoded IP, default to empty string (#14)

Android (proguard-rules.pro):
- Add ExoPlayer media3 HLS ProGuard rules (#34)

Frontend (useAudioPlayer.ts):
- Add retry limit (3) with exponential backoff for NETWORK_ERROR (#27)

Docker (Dockerfile.backend):
- Add playwright install chromium step (#22)
2026-07-05 12:33:26 +00:00

85 lines
2.8 KiB
Python

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": "in_playlist",
"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