diff --git a/streamrip/client/downloadable.py b/streamrip/client/downloadable.py index 3b29b504..1ff564ba 100644 --- a/streamrip/client/downloadable.py +++ b/streamrip/client/downloadable.py @@ -306,6 +306,90 @@ async def _decrypt_mqa_file(in_path, encryption_key): return dec_bytes +TIDAL_DASH_CONCURRENCY = 16 + + +class TidalDashDownloadable(Downloadable): + """Tidal track served as an MPEG-DASH manifest (HI_RES_LOSSLESS / Max). + + The audio comes as an initialization segment plus numbered fragmented-MP4 + segments. They are fetched concurrently, joined in order and remuxed into + a plain FLAC file with ffmpeg (no re-encoding). + """ + + def __init__( + self, + session: aiohttp.ClientSession, + initialization: str, + media_template: str, + start_number: int, + segments_count: int, + codec: str, + ): + self.session = session + self.source = "tidal" + self.url = initialization + self.segment_urls = [initialization] + [ + media_template.replace("$Number$", str(n)) + for n in range(start_number, start_number + segments_count) + ] + self.extension = "flac" if "flac" in codec.lower() else "m4a" + self._size = len(self.segment_urls) # progress is counted in segments + + async def _download(self, path: str, callback): + semaphore = asyncio.Semaphore(TIDAL_DASH_CONCURRENCY) + + async def fetch(url: str) -> str: + tmp = generate_temp_path(url) + async with semaphore: + async with self.session.get(url) as resp: + resp.raise_for_status() + async with aiofiles.open(tmp, "wb") as f: + async for chunk in resp.content.iter_chunked(64 * 1024): + await f.write(chunk) + callback(1) + return tmp + + # gather keeps the results in segment order. + segment_paths = await asyncio.gather(*(fetch(u) for u in self.segment_urls)) + await remux_dash_segments(segment_paths, path) + + +async def remux_dash_segments(paths: list[str], out: str): + """Join fragmented-MP4 segments (init segment first) and remux to `out`. + + The pieces are only a valid MP4 once concatenated byte for byte, so + ffmpeg's concat protocol (which demuxes every input on its own) can't be + used here. + """ + if shutil.which("ffmpeg") is None: + raise Exception("FFmpeg must be installed to download Tidal Max quality.") + + joined = generate_temp_path(out) + ".mp4" + try: + with open(joined, "wb") as dst: + for p in paths: + with open(p, "rb") as src: + shutil.copyfileobj(src, dst) + + command = [ + "ffmpeg", "-y", "-loglevel", "error", + "-i", joined, "-map", "0:a", "-c:a", "copy", out, + ] + proc = await converter.run_ffmpeg(command) + if proc.returncode != 0: + raise Exception( + "FFmpeg could not remux the Tidal DASH stream: " + + proc.stderr.decode(errors="replace") + ) + finally: + for p in (joined, *paths): + try: + os.remove(p) + except FileNotFoundError: + pass + + class SoundcloudDownloadable(Downloadable): def __init__(self, session, info: dict): self.session = session @@ -409,7 +493,7 @@ async def concat_audio_files(paths: list[str], out: str, ext: str, max_files_ope proc_futures = [] for i in range(num_batches): - command = ( + command = [ "ffmpeg", "-i", f"concat:{'|'.join(itertools.islice(it, max_files_open))}", @@ -418,15 +502,11 @@ async def concat_audio_files(paths: list[str], out: str, ext: str, max_files_ope "-loglevel", "warning", outpaths[i], - ) - fut = asyncio.create_subprocess_exec(*command, stderr=asyncio.subprocess.PIPE) - proc_futures.append(fut) + ] + proc_futures.append(converter.run_ffmpeg(command)) - # Create all processes concurrently + # Run all processes concurrently and wait for them to finish processes = await asyncio.gather(*proc_futures) - - # wait for all of them to finish - await asyncio.gather(*[p.communicate() for p in processes]) for proc in processes: if proc.returncode != 0: raise Exception( diff --git a/streamrip/client/tidal.py b/streamrip/client/tidal.py index 92455d20..bf37b62f 100644 --- a/streamrip/client/tidal.py +++ b/streamrip/client/tidal.py @@ -4,6 +4,7 @@ import logging import re import time +import xml.etree.ElementTree as ET from json import JSONDecodeError import aiohttp @@ -11,7 +12,7 @@ from ..config import Config from ..exceptions import NonStreamableError from .client import Client -from .downloadable import TidalDownloadable +from .downloadable import TidalDashDownloadable, TidalDownloadable logger = logging.getLogger("streamrip") @@ -162,9 +163,16 @@ async def get_downloadable(self, track_id: str, quality: int): ) logger.debug(resp) try: - manifest = json.loads(base64.b64decode(resp["manifest"]).decode("utf-8")) + raw_manifest = base64.b64decode(resp["manifest"]).decode("utf-8") except KeyError: raise Exception(resp["userMessage"]) + + # Max (HI_RES_LOSSLESS) is served as an MPEG-DASH manifest, not JSON. + if "dash+xml" in resp.get("manifestMimeType", ""): + return TidalDashDownloadable(self.session, **self._parse_mpd(raw_manifest)) + + try: + manifest = json.loads(raw_manifest) except JSONDecodeError: logger.warning( f"Failed to get manifest for {track_id}. Retrying with lower quality." @@ -183,6 +191,38 @@ async def get_downloadable(self, track_id: str, quality: int): restrictions=manifest.get("restrictions"), ) + @staticmethod + def _parse_mpd(mpd: str) -> dict: + """Read the segment layout from a Tidal DASH manifest. + + Tidal sends a single audio Representation with a $Number$-based + SegmentTemplate. Anything else fails loudly instead of silently + downloading the wrong thing. + """ + ns = "{urn:mpeg:dash:schema:mpd:2011}" + root = ET.fromstring(mpd) + rep = root.find(f".//{ns}Representation") + template = root.find(f".//{ns}SegmentTemplate") + timeline = root.find(f".//{ns}SegmentTimeline") + if rep is None or template is None or timeline is None: + raise NonStreamableError("Unrecognized Tidal DASH manifest") + + initialization = template.get("initialization") + media = template.get("media") + if not initialization or not media or "$Number$" not in media: + raise NonStreamableError("Unsupported Tidal DASH segment template") + + segments_count = sum( + int(s.get("r", 0)) + 1 for s in timeline.findall(f"{ns}S") + ) + return { + "initialization": initialization, + "media_template": media, + "start_number": int(template.get("startNumber", 1)), + "segments_count": segments_count, + "codec": rep.get("codecs", "flac"), + } + async def get_video_file_url(self, video_id: str) -> str: """Get the HLS video stream url. diff --git a/streamrip/config.py b/streamrip/config.py index bafa5687..bbb3aed1 100644 --- a/streamrip/config.py +++ b/streamrip/config.py @@ -54,6 +54,9 @@ class TidalConfig: quality: int # This will download videos included in Video Albums. download_videos: bool + # Convert tracks that Tidal only delivers as AAC (.m4a) to 16-bit FLAC. + # Defaults to False so configs without this key keep working. + convert_aac_to_flac: bool = False @dataclass(slots=True) diff --git a/streamrip/config.toml b/streamrip/config.toml index 029115bd..ac86a751 100644 --- a/streamrip/config.toml +++ b/streamrip/config.toml @@ -43,6 +43,9 @@ secrets = [] quality = 3 # This will download videos included in Video Albums. download_videos = true +# Convert tracks that Tidal only delivers as AAC (.m4a) to 16-bit FLAC with +# ffmpeg. This does not restore quality lost to AAC; it only gives a FLAC file. +convert_aac_to_flac = false # Do not change any of the fields below user_id = "" diff --git a/streamrip/converter.py b/streamrip/converter.py index aa4aa2aa..13f41c77 100644 --- a/streamrip/converter.py +++ b/streamrip/converter.py @@ -4,6 +4,7 @@ import logging import os import shutil +import subprocess from tempfile import gettempdir from typing import Final, Optional @@ -14,6 +15,21 @@ SAMPLING_RATES = {44100, 48000, 88200, 96000, 176400, 192000} +async def run_ffmpeg(command: list[str]) -> subprocess.CompletedProcess: + """Run an ffmpeg command without blocking the event loop. + + asyncio subprocesses are not supported by the WindowsSelectorEventLoop + that streamrip uses on Windows (they raise NotImplementedError), so the + process is run in a worker thread instead. + """ + return await asyncio.to_thread( + subprocess.run, + command, + stdin=subprocess.DEVNULL, + capture_output=True, + ) + + class Converter: """Base class for audio codecs.""" @@ -84,11 +100,8 @@ async def convert(self, custom_fn: Optional[str] = None): self.command = self._gen_command() logger.debug("Generated conversion command: %s", self.command) - process = await asyncio.create_subprocess_exec( - *self.command, - stderr=asyncio.subprocess.PIPE, - ) - out, err = await process.communicate() + process = await run_ffmpeg(self.command) + out, err = process.stdout, process.stderr if process.returncode == 0 and os.path.isfile(self.tempfile): if self.remove_source: os.remove(self.filename) diff --git a/streamrip/media/track.py b/streamrip/media/track.py index bc3207d5..c7b1d798 100644 --- a/streamrip/media/track.py +++ b/streamrip/media/track.py @@ -76,12 +76,33 @@ async def postprocess(self): if self.is_single: remove_title(self.meta.title) + if self._should_convert_aac_to_flac(): + await self._convert_aac_to_flac() + await tag_file(self.download_path, self.meta, self.cover_path) if self.config.session.conversion.enabled: await self._convert() self.db.set_downloaded(self.meta.info.id) + def _should_convert_aac_to_flac(self) -> bool: + # The general [conversion] setting takes precedence when it is enabled. + return ( + self.downloadable.source == "tidal" + and self.downloadable.extension == "m4a" + and self.config.session.tidal.convert_aac_to_flac + and not self.config.session.conversion.enabled + ) + + async def _convert_aac_to_flac(self): + # Runs before tagging so the tags and cover are written natively to + # the FLAC file. AAC is decoded to 16-bit, there is nothing above that. + engine = converter.FLAC( + filename=self.download_path, bit_depth=16, remove_source=True + ) + await engine.convert() + self.download_path = engine.final_fn + async def _convert(self): c = self.config.session.conversion engine_class = converter.get(c.codec) diff --git a/streamrip/rip/cli.py b/streamrip/rip/cli.py index 2b3f532e..6c748222 100644 --- a/streamrip/rip/cli.py +++ b/streamrip/rip/cli.py @@ -67,6 +67,12 @@ def wrapper(*args, **kwargs): "--codec", help="Convert the downloaded files to an audio codec (ALAC, FLAC, MP3, AAC, or OGG)", ) +@click.option( + "--aac-to-flac", + help="Convert Tidal tracks that are only available as AAC to FLAC", + is_flag=True, + default=False, +) @click.option( "--no-progress", help="Do not show progress bars", @@ -87,7 +93,16 @@ def wrapper(*args, **kwargs): ) @click.pass_context def rip( - ctx, config_path, folder, no_db, quality, codec, no_progress, no_ssl_verify, verbose + ctx, + config_path, + folder, + no_db, + quality, + codec, + aac_to_flac, + no_progress, + no_ssl_verify, + verbose, ): """Streamrip: the all in one music downloader.""" global logger @@ -155,6 +170,9 @@ def rip( assert codec.upper() in ("ALAC", "FLAC", "OGG", "MP3", "AAC") c.session.conversion.codec = codec.upper() + if aac_to_flac: + c.session.tidal.convert_aac_to_flac = True + if no_progress: c.session.cli.progress_bars = False diff --git a/tests/test_config.toml b/tests/test_config.toml index a33a1864..751eb354 100644 --- a/tests/test_config.toml +++ b/tests/test_config.toml @@ -43,6 +43,9 @@ secrets = ['secret1', 'secret2'] quality = 3 # This will download videos included in Video Albums. download_videos = true +# Convert tracks that Tidal only delivers as AAC (.m4a) to 16-bit FLAC with +# ffmpeg. This does not restore quality lost to AAC; it only gives a FLAC file. +convert_aac_to_flac = false # Do not change any of the fields below user_id = "userid" diff --git a/tests/test_tidal_dash.py b/tests/test_tidal_dash.py new file mode 100644 index 00000000..c67530b9 --- /dev/null +++ b/tests/test_tidal_dash.py @@ -0,0 +1,46 @@ +import pytest + +from streamrip.client.downloadable import TidalDashDownloadable +from streamrip.client.tidal import TidalClient +from streamrip.exceptions import NonStreamableError + +# Same shape as the manifest Tidal returns for a 24-bit/192 kHz track. +MPD = """ + + + + + + + + + + +""" + + +def test_parse_mpd(): + info = TidalClient._parse_mpd(MPD) + assert info == { + "initialization": "https://cdn.test/0.mp4", + "media_template": "https://cdn.test/$Number$.mp4", + "start_number": 1, + "segments_count": 70, + "codec": "flac", + } + + +def test_dash_segment_urls(): + d = TidalDashDownloadable(None, **TidalClient._parse_mpd(MPD)) + assert d.extension == "flac" + assert d.segment_urls[0] == "https://cdn.test/0.mp4" + assert d.segment_urls[1] == "https://cdn.test/1.mp4" + assert d.segment_urls[-1] == "https://cdn.test/70.mp4" + assert len(d.segment_urls) == 71 + + +def test_parse_mpd_rejects_unknown_layout(): + with pytest.raises(NonStreamableError): + TidalClient._parse_mpd(MPD.replace("$Number$", "$Time$")) + with pytest.raises(NonStreamableError): + TidalClient._parse_mpd(MPD.replace("SegmentTimeline", "SegmentList"))