Tidal: download Max (HI_RES_LOSSLESS) via DASH, fix ffmpeg on Windows, optional AAC→FLAC

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(

streamrip/client/tidal.py

@@ -4,14 +4,15 @@
 import logging
 import re
 import time
+import xml.etree.ElementTree as ET
 from json import JSONDecodeError
 
 import aiohttp
 
 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.
 

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)

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 = ""

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)

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)

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
 

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"

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 = """<?xml version='1.0' encoding='UTF-8'?>
+<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" mediaPresentationDuration="PT4M36.213S">
+  <Period id="0">
+    <AdaptationSet id="0" contentType="audio" mimeType="audio/mp4">
+      <Representation id="FLAC_HIRES,192000,24" codecs="flac" bandwidth="5611927" audioSamplingRate="192000">
+        <SegmentTemplate timescale="192000" initialization="https://cdn.test/0.mp4" media="https://cdn.test/$Number$.mp4" startNumber="1">
+          <SegmentTimeline><S d="765952" r="68"/><S d="182261"/></SegmentTimeline>
+        </SegmentTemplate>
+      </Representation>
+    </AdaptationSet>
+  </Period>
+</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"))