"""Post-download media handling: FLAC-in-MP4 remux and ffprobe verification.
Tidal serves hi-res FLAC **inside an MP4/DASH container**. Saving those
bytes as ``.flac`` would be a lie; the two honest outcomes are:
- remux to true FLAC with ``ffmpeg -c copy`` (no re-encode) when the
binary is available — the DJ/CDJ default;
- otherwise keep the container and name it ``.m4a``.
Verification (when ``ffprobe`` is available) probes the finished file
and compares codec/sample-rate/bit-depth against the manifest facts.
A mismatch means the CDN delivered something other than what the
manifest promised — the item fails with a diagnostic instead of
silently shipping degraded audio.
"""
from __future__ import annotations
import json
import logging
import os
import shutil
import subprocess
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from .errors import CancelledError, DownloadError
from .models import StreamInfo
logger = logging.getLogger("tidalpy.media")
_FFPROBE_TIMEOUT = 60
_FFMPEG_TIMEOUT = 300
_PROTOCOL_WHITELIST = ["file"]
_PROBE_FORMATS = "mov,mp4,m4a,3gp,3g2,mj2,flac,aac,ac3,eac3,mp3,ogg,mpegts"
[docs]
def flac_in_mp4(stream: StreamInfo) -> bool:
"""Pre-download prediction: True when the delivered FLAC is expected
to live in an MP4/DASH container.
This is only a prediction (it is what picks the file extension
before the bytes exist). The remux decision itself is made from the
bytes on disk — see :func:`is_flac_file` (reference-client parity:
the legacy tool sniffs the same header before converting).
"""
codec = (stream.codec or "").lower()
mime = (stream.manifest_mime or "").lower()
return "flac" in codec and (
"dash+xml" in mime or any(".mp4" in url.lower() for url in stream.urls[:1])
)
[docs]
def is_flac_file(path: Path) -> bool:
"""True when the bytes at ``path`` are a genuine FLAC stream
(``fLaC`` magic). A missing or unreadable file is not FLAC.
The remux decision is made from the bytes, not the manifest: a
missed prediction (an unusual bento URL shape, a CDN that changed
its naming) must never ship MP4 bytes under a ``.flac`` name.
"""
try:
with open(path, "rb") as handle:
return handle.read(4) == b"fLaC"
except OSError:
return False
[docs]
def manifest_facts(stream: StreamInfo) -> MediaFacts:
return MediaFacts(
codec=stream.codec or None,
sample_rate=stream.sample_rate,
bit_depth=stream.bit_depth,
channels=stream.channels,
bandwidth=stream.bandwidth,
)
def _run(
cmd: list[str],
timeout: float,
stdin: int | None,
stop_event: threading.Event | None = None,
) -> subprocess.CompletedProcess:
"""Run a media tool with a timeout — and, when a ``stop_event`` is
given, cancel responsiveness: a cancelled run kills the child and
raises :class:`~tidalpy.errors.CancelledError` instead of waiting
for a multi-minute remux to finish (reference-fork pin: media
processes are stopped on cancellation)."""
try:
return subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
timeout=timeout,
stdin=stdin,
)
except subprocess.TimeoutExpired as exc:
# Cancellation during the wait: the child is dead (subprocess
# kills it on timeout) — surface it as a cancel, not a failure.
if stop_event is not None and stop_event.is_set():
raise CancelledError("media process cancelled") from exc
raise DownloadError(f"media process timed out after {timeout:.0f}s") from exc
def _run_with_cancel(
cmd: list[str],
timeout: float,
stop_event: threading.Event | None,
) -> subprocess.CompletedProcess:
"""``subprocess.run`` that polls the stop event while the child
runs, killing it promptly on cancellation.
``subprocess.run``'s timeout wait is not interruptible by an
external event, so for the cancel path we drive ``Popen``
ourselves: wait in short slices, checking the event; on cancel,
kill + reap and raise.
"""
if stop_event is None:
return _run(cmd, timeout, subprocess.DEVNULL)
try:
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.DEVNULL,
)
except OSError as exc:
raise DownloadError(f"media process failed to start: {exc}") from exc
deadline = time.monotonic() + timeout
try:
while True:
if stop_event.is_set():
process.kill()
try:
process.communicate(timeout=5)
except (subprocess.SubprocessError, OSError):
pass
raise CancelledError("media process cancelled")
try:
stdout, stderr = process.communicate(timeout=0.5)
return subprocess.CompletedProcess(
cmd,
process.returncode,
stdout.decode(errors="replace"),
stderr.decode(errors="replace"),
)
except subprocess.TimeoutExpired:
if time.monotonic() >= deadline:
process.kill()
try:
process.communicate(timeout=5)
except (subprocess.SubprocessError, OSError):
pass
raise DownloadError(f"media process timed out after {timeout:.0f}s")
finally:
if process.poll() is None:
process.kill()
try:
process.communicate(timeout=5)
except (subprocess.SubprocessError, OSError):
pass
[docs]
def export_flac_from_container(
path: Path, stop_event: threading.Event | None = None
) -> Path:
"""Remux FLAC-in-MP4 to true FLAC (``-c copy``).
Returns the final path: ``<stem>.flac`` on success, ``<stem>.m4a``
when ffmpeg is missing or the remux fails (the bytes stay valid,
just container-named). With a ``stop_event``, a cancelled run
kills the running ffmpeg and propagates the cancellation
(reference-fork pin) instead of waiting out the remux.
"""
flac_path = path.with_suffix(".flac")
fallback = path.with_suffix(".m4a")
ffmpeg = shutil.which("ffmpeg")
if ffmpeg is None:
logger.warning("ffmpeg not found; keeping FLAC-in-MP4 as %s", fallback.name)
if fallback != path:
path.replace(fallback)
return fallback
temp_path = flac_path.with_name(f"{flac_path.name}.tmp.{os.getpid()}.flac")
temp_path.unlink(missing_ok=True)
cmd = [
ffmpeg,
"-nostdin",
"-y",
"-hide_banner",
"-loglevel",
"error",
"-protocol_whitelist",
"file",
"-format_whitelist",
"mov,mp4,m4a,3gp,3g2,mj2,flac",
"-i",
str(path),
"-map",
"0:a:0",
"-c",
"copy",
"-f",
"flac",
str(temp_path),
]
try:
completed = _run_with_cancel(cmd, _FFMPEG_TIMEOUT, stop_event)
ok = (
completed.returncode == 0
and temp_path.exists()
and temp_path.stat().st_size > 0
)
if not ok:
detail = (completed.stderr or completed.stdout or "").strip()
raise DownloadError(
detail or f"ffmpeg exited with code {completed.returncode}"
)
os.replace(temp_path, flac_path)
if flac_path != path:
path.unlink(missing_ok=True)
return flac_path
except CancelledError:
raise # a cancelled run must not degrade to a container rename
except Exception as exc: # noqa: BLE001 - fall back to container
logger.warning(
"FLAC remux failed for %s: %s; keeping container as .m4a", path.name, exc
)
if fallback != path:
path.replace(fallback)
return fallback
finally:
temp_path.unlink(missing_ok=True)
def _positive_int(value: str | int | float | None) -> int | None:
if value is None:
return None
try:
number = int(value)
return number if number > 0 else None
except (TypeError, ValueError):
return None