from pathlib import Path from urllib.parse import quote from app.features.streaming.library.ffprobe import ffprobe from app.features.streaming.types import StreamingError from app.library.Utils import get_file_sidecar class Playlist: def __init__(self, download_path: Path, url: str): self.url: str = url "The base URL for the playlist." self.download_path: Path = download_path "The path where files are downloaded." async def make(self, file: Path) -> str: ref: Path = Path(str(file.relative_to(self.download_path)).strip("/")) try: ff = await ffprobe(file) except UnicodeDecodeError: pass if "duration" not in ff.metadata: msg = f"Unable to get '{ref}' duration." raise StreamingError(msg) playlist: list[str] = [] playlist.append("#EXTM3U") subs: str = "" duration: float = float(ff.metadata.get("duration")) for sub_file in get_file_sidecar(file).get("subtitle", []): lang: str = sub_file["lang"] item: Path = sub_file["file"] name: str = sub_file["name"] subs = ',SUBTITLES="subs"' url = f"{self.url}api/player/m3u8/subtitle/{quote(str(ref.with_name(item.name)))}.m3u8?duration={duration}" playlist.append( f'#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="subs",NAME="{name}",DEFAULT=NO,AUTOSELECT=NO,FORCED=NO,LANGUAGE="{lang}",URI="{url}"' ) playlist.append(f"#EXT-X-STREAM-INF:PROGRAM-ID=1{subs}") playlist.append(f"{self.url}api/player/m3u8/video/{quote(str(ref))}.m3u8") return "\n".join(playlist)