kenzuya/unshackle
1
0
Fork 2
mirror of https://github.com/unshackle-dl/unshackle.git synced 2026-11-15 05:25:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(music): pick the song container from the codec in the downloaded file

Browse Source 
FFprobe sets the codec; DTS goes to .mka, PCM to .flac or .wav, AC-4/MPEG-H/DTS-UHD to MP4Box. The remux also cuts --sample, --no-mux no longer applies to a Song, and the tagger adds cover fallback, WebP, ARTISTS and ReplayGain. Song tag data now reaches a --remote client.
This commit is contained in:
imSp4rky
2026-11-14 19:58:21 -07:00
parent 703bcf872c
commit 7a5ad5024a
24 changed files with 785 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/dev/creating-a-service.md
+3 -3
View File
@@ -370,9 +370,9 @@ Common constructor fields:
From `get_tracks()`, give one `Audio` track for each codec and bitrate you offer
for the song. The framework then selects one with its usual audio options.
A `Song` skips the muxer, so restore the real container suffix on the track path
in `on_track_downloaded`. The **MUSIC_EXAMPLE** service shows the full music
surface.
A `Song` skips the muxer: the framework reads the codec from the downloaded file
and gives the file its container, so a service does not rename the track path.
The **MUSIC_EXAMPLE** service shows the full music surface.
!!! note "The ID must be unique and stable"
`id_` must be truthy and, if it has a length, at least 4 characters, to
docs/dev/service-architecture.md
+6 -8
View File
@@ -202,7 +202,9 @@ Each subclass adds its own required fields on top of `id_` and `service`:
The tagger also falls back to `Song.data` for fields the constructor did not get, such
as the composer or a full release date. It reads the top level of that dict and a
nested `metadata` sub-dict, with the nested keys winning, so either shape works.
nested `metadata` sub-dict, with the nested keys winning, so either shape works. The keys
that a remote client gets back are `MUSIC_DATA_KEYS` in `unshackle/core/titles/music.py`,
for example `artists` (a list) and `replaygain_track_gain`.
From `get_tracks()`, a music service gives **one `Audio` track for each codec and
bitrate it offers for a `Song`**. The framework then selects one with its usual audio
@@ -216,13 +218,9 @@ Each subclass adds its own required fields on top of `id_` and `service`:
`GROUP_AUDIO_DOWNLOADS` flag, `get_music_track_options` and `get_music_collection_label`
no longer exist, and a music service does not need them.
!!! warning "Restore the container in `on_track_downloaded`"
Every track downloads to a temp file named `Audio_<id>.mp4` whatever it holds, and
the final move keeps that suffix. A `Song` skips the muxer, so its download **is**
the delivered file, and the tagger picks its metadata format from the suffix.
Rename the track path to the true container in `on_track_downloaded`, or a FLAC
lands as an untaggable `.mp4`. The event fires twice, once before decryption and
once at the end, so wait until `track.drm` is clear before you rename.
Core picks the container of a `Song` from the codec in the downloaded file, so a music
service does not rename the track path. Read
[Music output files](../guide/output-and-naming.md#music-output-files).
The **MUSIC_EXAMPLE** reference service, at `unshackle/services/MUSIC_EXAMPLE/`, shows
the full music surface.
docs/getting-started/installation.md
+2 -2
View File
@@ -268,7 +268,7 @@ download. For example, `dovi_tool` only matters if you download Dolby Vision vid
| Tool | Binary looked up | Why it is needed |
| --- | --- | --- |
| Bento4 (`mp4decrypt`) | `mp4decrypt` | Alternative DRM decryptor to shaka-packager |
| MP4Box (GPAC) | `MP4Box` | Muxing DTS:X Profile 2 (DTS-UHD) and MPEG-H audio, and 3D video |
| MP4Box (GPAC) | `MP4Box` | Muxing DTS:X Profile 2 (DTS-UHD) and MPEG-H audio, and 3D video; remuxing AC-4, MPEG-H and DTS-UHD music |
| dovi_tool | `dovi_tool` | Dolby Vision metadata handling |
| HDR10Plus_tool | `hdr10plus_tool` | HDR10+ metadata handling |
| SubtitleEdit | `seconv`, `SubtitleEdit` | Subtitle conversion (the `SeConv` CLI from 5.x) |
@@ -280,7 +280,7 @@ download. For example, `dovi_tool` only matters if you download Dolby Vision vid
| Docker | `docker` | Gluetun VPN proxy support |
| Git | `git` | Fetching and updating remote service repositories |
!!! info "MP4Box is only used for DTS:X Profile 2, MPEG-H and 3D video"
!!! info "MP4Box is only used for DTS:X Profile 2, MPEG-H, AC-4 music and 3D video"
Titles with DTS:X Profile 2 (DTS-UHD) or MPEG-H audio, or a stereoscopic 3D video track, mux to
MP4; everything else muxes to MKV with `mkvmerge`. Install with `apt install gpac`.
docs/guide/downloading.md
+4 -3
View File
@@ -865,8 +865,8 @@ By default, `mkvmerge` muxes the tracks of a movie or an episode into a single M
window, or the whole file when the track is one file.
!!! note "Music is never muxed"
A song skips the muxer. unshackle keeps the container that the service
delivered, such as `.flac`, `.m4a`, or `.mp3`, and writes the metadata into that file.
A song skips the muxer. unshackle remuxes it into the container of its codec,
such as `.flac`, `.m4a`, or `.mp3`, and writes the metadata into that file.
Read [Music output files](output-and-naming.md#music-output-files).
### After the download
@@ -874,7 +874,8 @@ By default, `mkvmerge` muxes the tracks of a movie or an episode into a single M
`--postscript "<command>"` runs your own command once per output file, with unshackle's
metadata substituted into `{variable}` placeholders. It is repeatable, and it replaces the
`post_scripts` config for that run. No `success` post-script operates under `--no-mux`, because
that run writes no muxed output. A `failure` post-script still operates if the download fails.
that run writes no muxed output; a song is the exception, because `--no-mux` does not apply to it.
A `failure` post-script still operates if the download fails.
```shell title="Hand each finished file to an uploader"
unshackle dl --postscript "python /opt/upload.py {filepath} --service={service}" EXAMPLE 81234567
docs/guide/output-and-naming.md
+36 -16
View File
@@ -262,22 +262,37 @@ The per-kind folder config keys are `movies`, `series`, `sports`, `songs`, and `
## Music output files
A song does not go through the muxer. Music services deliver audio in a fragmented MP4, which is
not a music file. Its header states a length of zero, so a player stops after the first fragment.
FLAC inside an MP4 is also not a FLAC stream. A rename cannot correct either fault, so unshackle
remuxes each downloaded audio track into the container of its codec, then moves the file to its
final path.
A song does not go through the muxer. Music services deliver audio as a fragmented MP4, a WebM
file or a plain audio file, and every download has a `.mp4` name. A fragmented MP4 header states a
length of zero, so a player stops after the first fragment. A rename cannot correct this, so
unshackle remuxes each downloaded audio track into the container of its codec, then moves the file
to its final path.
FFprobe reads the codec from the downloaded file. The codec that the service declares is used only
when FFprobe cannot read the file.
| Codec | Extension |
|---|---|
| FLAC | `.flac` |
| PCM, integer, 24-bit or less | `.flac` |
| PCM, more than 24-bit or floating point | `.wav` |
| Opus | `.opus` |
| Vorbis | `.ogg` |
| MP3 | `.mp3` |
| AAC, ALAC, and all other codecs | `.m4a` |
| DTS, DTS-HD | `.mka` |
| AAC, ALAC, Dolby Digital, Dolby Digital Plus, and all other codecs | `.m4a` |
unshackle re-encodes FLAC in this step. The audio does not change, but the new file states its
true length. A plain copy cannot, because it keeps the source header and the zero in it.
unshackle encodes FLAC and PCM to FLAC in this step. The audio does not change, but the new file
states its true length. A plain copy cannot, because it keeps the source header and the zero in it.
FLAC cannot hold floating-point PCM, and FFmpeg writes 32-bit FLAC only as an experimental feature,
so that audio goes to `.wav`.
FFmpeg writes DTS into MP4 with the wrong codec entry, so DTS goes to Matroska. AC-4, MPEG-H and
DTS:X Profile 2 (DTS-UHD) go through MP4Box, because FFmpeg cannot write them into MP4. MP4Box
needs [GPAC](../getting-started/installation.md#recommended-and-optional-tools).
With `--sample`, the same step cuts the song to the sample window. If the window starts after the
end of the song, unshackle shows a warning and keeps the full download.
unshackle writes no `.mkv` file for music, and the [muxing options](#muxing-options) below do not
apply.
@@ -295,13 +310,18 @@ around it.
### Music metadata
After the file moves, unshackle writes the metadata into the audio file. It uses Vorbis
comments for FLAC and Ogg, ID3 frames for MP3, and MP4 atoms for M4A. The tags hold the
track title, artist, album artist, track and disc numbers, date, genre, ISRC, UPC,
copyright, and record label. When `tag` and `tag_group_name` are set, unshackle also writes the group
name into a `GROUP` tag.
comments for FLAC and Ogg, ID3 frames for MP3 and WAV, MP4 atoms for M4A, and Matroska tags for
MKA. The tags hold the track title, artist, album artist, track and disc numbers, date, genre,
ISRC, UPC, copyright, and record label. When the service gives them, unshackle also writes each
artist of a song with several artists into `ARTISTS`, and the ReplayGain values into
`REPLAYGAIN_*` tags. An Opus file gets no `REPLAYGAIN_*` tags, because RFC 7845 advises against
them in Opus files. When `tag` and `tag_group_name` are set, unshackle also writes the group name
into a `GROUP` tag.
unshackle embeds cover art when the service gives an artwork URL. It downloads the image
and puts it in the file.
and puts it in the file. With no artwork URL, it uses the first image attachment of the song. It
converts a WebP image to JPEG. It skips an image larger than 16 MiB, the FLAC limit, with a
warning. An MKA file gets no cover art.
unshackle writes lyrics when the service supplies them: a `LYRICS` Vorbis comment for FLAC, a
`USLT` frame for MP3, and a `©lyr` atom for M4A. Ogg uses the same `LYRICS` field as
@@ -314,8 +334,8 @@ the timed lines from that file. The file also comes back to the client from a re
A metadata fault does not stop the run. The file is already in place, so unshackle writes a
warning and continues.
!!! warning "`--no-mux` leaves music files untagged"
[`--no-mux`](cli-reference.md#output-muxing-files) moves each downloaded track file to the output path without the tagging step. A music file from such a run holds only the metadata that the service put in it.
!!! note "`--no-mux` does not change music output"
A song has no mux step, so [`--no-mux`](cli-reference.md#output-muxing-files) does not apply to it. The song gets the same file name, metadata, cover art and lyrics file as in a run without the flag.
## Muxing options
@@ -486,7 +506,7 @@ A few muxing behaviors are automatic and not configurable:
- Descriptive audio gets the visually-impaired flag. SDH subtitles get the hearing-impaired flag. Forced subtitles get the forced flag.
- unshackle writes the chapters from the title's chapter list (see [Chapters](#chapters-naming) below).
The `--no-mux` flag skips muxing entirely and writes the individual track files, each with a track-type suffix, into the same output/folder structure. A track that shaka-packager decrypts has its date set to 1970-01-01; other tracks keep the date of the source file.
The `--no-mux` flag skips muxing entirely and writes the individual track files, each with a track-type suffix, into the same output/folder structure. An audio track in an MP4 container gets the `.m4a` extension. A track that shaka-packager decrypts has its date set to 1970-01-01; other tracks keep the date of the source file.
## Sample files
docs/reference/configuration/post-scripts.md
+2 -1
View File
@@ -328,7 +328,8 @@ credentials and `token=` / `api_key=` query parameters are also masked.
## Limitations
- No `success` post-script operates under `--no-mux` or `--skip-dl`, because neither writes a
muxed output. A `failure` post-script still runs under `--no-mux` if the download itself fails.
muxed output. A song is the exception: `--no-mux` does not apply to it, so its post-scripts
operate. A `failure` post-script still runs under `--no-mux` if the download itself fails.
- `failure` post-scripts only operate in `file` mode.
- There is no webhook sender. Call `curl` from a script instead.
- Script output is not captured. Redirect it inside your own script if you want a log.
tests/core/test_music_container.py
+130 -5
View File
@@ -13,11 +13,13 @@ from __future__ import annotations
import shutil
import subprocess
from pathlib import Path
from typing import Optional
import mutagen
import pytest
from tests.core.test_music_tagger_roundtrip import make_song
from unshackle.core import binaries
from unshackle.core.music.tagger import write_music_metadata
from unshackle.core.tracks import Audio
@@ -79,7 +81,7 @@ def test_remux_states_the_full_length_in_the_codec_container(
track = Audio(id_="a1", url="https://example.test/a1.mp4", language="en", codec=codec)
track.path = source
assert track.to_music_container() is True
track.to_music_container()
assert track.path.suffix == extension
assert not source.exists()
assert header_length(track.path) == pytest.approx(6.0, abs=0.5)
@@ -118,14 +120,14 @@ def test_flac_is_re_encoded_because_a_copy_states_no_length(tmp_path: Path) -> N
assert mutagen.File(track.path).info.bits_per_sample == 16, "the re-encode must not change the sample format"
def test_dolby_digital_plus_falls_back_to_the_mp4_muxer(tmp_path: Path) -> None:
"""A .m4a name picks FFmpeg's ipod muxer, which refuses Dolby Digital Plus; the remux must not fail on it."""
def test_dolby_digital_plus_takes_the_mp4_muxer(tmp_path: Path) -> None:
"""FFmpeg's ipod muxer refuses Dolby Digital Plus, so it goes straight to the plain MP4 muxer."""
source = encode(tmp_path / "track.mp4", "eac3", "-ar", "48000")
track = Audio(id_="a1", url="https://example.test/a1.mp4", language="en", codec=Audio.Codec.EC3)
track.path = source
assert track.to_music_container() is True
track.to_music_container()
assert track.path.suffix == ".m4a"
assert header_length(track.path) == pytest.approx(6.0, abs=0.5)
assert write_music_metadata(track.path, make_song()).written
@@ -139,8 +141,131 @@ def test_a_failed_remux_keeps_the_source_and_removes_its_partial_file(tmp_path:
track = Audio(id_="a1", url="https://example.test/a1.mp4", language="en", codec=None)
track.path = source
with pytest.raises(subprocess.CalledProcessError):
with pytest.raises(RuntimeError):
track.to_music_container()
assert track.path == source and source.exists(), "a failed remux must keep the downloaded file"
assert not list(tmp_path.glob("track_music*")), "a failed remux must not leave a partial file"
def make_track(path: Path, codec: Optional[Audio.Codec]) -> Audio:
track = Audio(id_="a1", url="https://example.test/a1.mp4", language="en", codec=codec)
track.path = path
return track
@pytest.mark.parametrize(
("encoder", "fmt", "declared", "extension", "probed"),
[
("libopus", "webm", None, ".opus", Audio.Codec.OPUS),
("libvorbis", "webm", Audio.Codec.AAC, ".ogg", Audio.Codec.OGG),
("flac", "flac", Audio.Codec.AAC, ".flac", Audio.Codec.FLAC),
("libmp3lame", "mp3", None, ".mp3", Audio.Codec.MP3),
("pcm_s16le", "wav", None, ".flac", Audio.Codec.PCM),
("pcm_s24le", "wav", None, ".flac", Audio.Codec.PCM),
("pcm_f32le", "wav", None, ".wav", Audio.Codec.PCM),
],
)
def test_the_file_codec_decides_the_container(
tmp_path: Path, encoder: str, fmt: str, declared: Optional[Audio.Codec], extension: str, probed: Audio.Codec
) -> None:
"""Every track downloads under a .mp4 name, and a service can declare no codec or the wrong one."""
track = make_track(encode(tmp_path / "track.mp4", encoder, "-f", fmt), declared)
track.to_music_container()
assert track.path.suffix == extension
assert track.codec == probed
assert header_length(track.path) == pytest.approx(6.0, abs=0.5)
assert write_music_metadata(track.path, make_song()).written
def test_24_bit_pcm_stays_24_bit_in_flac(tmp_path: Path) -> None:
track = make_track(encode(tmp_path / "track.mp4", "pcm_s24le", "-f", "wav"), Audio.Codec.PCM)
track.to_music_container()
assert mutagen.File(track.path).info.bits_per_sample == 24
def test_dts_goes_to_matroska(tmp_path: Path) -> None:
"""FFmpeg writes DTS into MP4 with the wrong sample entry, so DTS takes Matroska."""
track = make_track(encode(tmp_path / "track.mp4", "dca", "-ar", "48000", "-f", "dts"), None)
track.to_music_container()
assert track.path.suffix == ".mka"
assert track.codec == Audio.Codec.DTS
def test_without_ffprobe_the_declared_codec_decides(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(binaries, "FFProbe", None)
track = make_track(fragmented_mp4(tmp_path / "track.mp4", "aac"), Audio.Codec.AAC)
track.to_music_container()
assert track.path.suffix == ".m4a"
@pytest.mark.parametrize(("encoder", "extra"), [("flac", ()), ("eac3", ("-ar", "48000"))])
def test_a_sample_window_is_cut_in_the_remux(tmp_path: Path, encoder: str, extra: tuple[str, ...]) -> None:
"""The cut runs in the same step, so a FLAC header states the cut length and EC-3 takes the MP4 muxer."""
source = encode(tmp_path / "track.mp4", encoder, *extra)
track = make_track(source, None)
track.to_music_container(start=1.0, duration=2.0)
assert header_length(track.path) == pytest.approx(2.0, abs=0.3)
def test_the_remux_writes_no_encoder_tag(tmp_path: Path) -> None:
track = make_track(encode(tmp_path / "track.mp4", "libmp3lame", "-f", "mp3"), Audio.Codec.MP3)
track.to_music_container()
tags = mutagen.File(track.path).tags
assert not tags or "TSSE" not in tags
def test_float_pcm_goes_to_wav_whatever_its_bit_depth(tmp_path: Path) -> None:
"""16- and 24-bit float PCM report a bit depth FLAC can hold, but FLAC has no float samples."""
track = make_track(encode(tmp_path / "track.mp4", "pcm_f32le", "-f", "wav"), Audio.Codec.PCM)
stream = {"codec_name": "pcm_f24le", "sample_fmt": "flt", "bits_per_raw_sample": "24"}
output = track.ffmpeg_music_container(track.path, stream, 0.0, None)
assert output.suffix == ".wav"
assert mutagen.File(output).info.bits_per_sample == 32
def test_a_sample_window_after_the_end_keeps_the_download(tmp_path: Path) -> None:
source = encode(tmp_path / "track.mp4", "flac")
track = make_track(source, None)
with pytest.raises(ValueError):
track.to_music_container(start=60.0, duration=30.0)
assert track.path == source and source.exists()
assert not list(tmp_path.glob("track_music*"))
@pytest.mark.skipif(binaries.MP4Box is None, reason="requires MP4Box")
def test_mp4box_writes_a_cut_m4a_and_leaves_no_temp_file(tmp_path: Path) -> None:
track = make_track(fragmented_mp4(tmp_path / "track.mp4", "aac"), Audio.Codec.AAC)
output = track.mp4box_music_container(track.path, 1.0, 2.0)
assert output.suffix == ".m4a"
assert header_length(output) == pytest.approx(2.0, abs=0.3)
assert sorted(p.name for p in tmp_path.iterdir()) == ["track.mp4", "track_music.m4a"]
def test_mp4box_route_removes_its_temp_file_when_preparation_fails(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def strip(source: Path, output: Path) -> bool:
output.write_bytes(b"init")
return True
def fail(path: Path) -> None:
raise OSError("disk full")
monkeypatch.setattr(binaries, "MP4Box", Path("MP4Box"))
monkeypatch.setattr("unshackle.core.tracks.track.strip_duplicate_init_boxes", strip)
monkeypatch.setattr("unshackle.core.tracks.track.fix_data_reference_indexes", fail)
track = make_track(fragmented_mp4(tmp_path / "track.mp4", "aac"), Audio.Codec.AC4)
with pytest.raises(OSError):
track.mp4box_music_container(track.path, 0.0, None)
assert sorted(p.name for p in tmp_path.iterdir()) == ["track.mp4"]
tests/core/test_music_tagger_roundtrip.py
+79
View File
@@ -16,10 +16,12 @@ import pytest
from mutagen.flac import FLAC
from mutagen.mp3 import MP3
from mutagen.mp4 import MP4
from mutagen.oggopus import OggOpus
from mutagen.oggvorbis import OggVorbis
from unshackle.core.music.tagger import write_music_metadata
from unshackle.core.titles.music import Song
from unshackle.core.tracks import Attachment
pytestmark = pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="requires ffmpeg")
@@ -196,3 +198,80 @@ def test_a_nested_metadata_dict_reaches_the_tags(flac_path: Path) -> None:
assert audio["COMPOSER"] == ["Example Composer"]
assert audio["RELEASEDATE"] == ["2020-01-02"]
assert audio["DATE"] == ["2020-01-02"], "a full release date must beat the bare year"
def test_an_image_attachment_is_the_cover_when_the_song_gives_no_artwork_url(flac_path: Path, tmp_path: Path) -> None:
image = tmp_path / "thumbnail.png"
image.write_bytes(PNG_STUB)
song = make_song()
song.artwork_url = None
song.tracks.attachments.append(Attachment(path=image, name="thumbnail", mime_type="image/png"))
assert write_music_metadata(flac_path, song).artwork_embedded
assert FLAC(flac_path).pictures[0].data == PNG_STUB
def test_a_webp_cover_is_embedded_as_jpeg(tmp_path: Path) -> None:
webp = tmp_path / "cover.webp"
subprocess.run(
["ffmpeg", "-nostdin", "-v", "error", "-f", "lavfi", "-i", "color=red:s=16x16", "-frames:v", "1", str(webp)],
check=True,
)
m4a = tmp_path / "track.m4a"
subprocess.run(
["ffmpeg", "-nostdin", "-v", "error", "-f", "lavfi", "-i", "anullsrc", "-t", "0.2", "-c:a", "aac", str(m4a)],
check=True,
)
write_music_metadata(m4a, make_song(), session=FakeSession(webp.read_bytes()))
cover = MP4(m4a)["covr"][0]
assert bytes(cover).startswith(b"\xff\xd8\xff")
def test_a_cover_over_16_mib_is_skipped(flac_path: Path, caplog: pytest.LogCaptureFixture) -> None:
result = write_music_metadata(flac_path, make_song(), session=FakeSession(PNG_STUB + b"\x00" * 2**24))
assert not result.artwork_embedded
assert not FLAC(flac_path).pictures
assert "16 MiB" in caplog.text
def test_artists_and_replaygain_read_back(flac_path: Path) -> None:
song = make_song()
song.data = {"artists": ["Example Artist", "Guest Artist"], "replaygain_track_gain": -7.25, "quality": "Hi-Res"}
write_music_metadata(flac_path, song)
audio = FLAC(flac_path)
assert audio["ARTISTS"] == ["Example Artist", "Guest Artist"]
assert audio["REPLAYGAIN_TRACK_GAIN"] == ["-7.25 dB"]
assert "COMMENT" not in audio, "a quality label is display text, not a comment"
def test_a_replaygain_value_that_is_not_a_number_is_skipped(flac_path: Path) -> None:
song = make_song()
song.data = {"replaygain_track_gain": "nan", "replaygain_track_peak": "inf", "replaygain_album_gain": -3}
write_music_metadata(flac_path, song)
audio = FLAC(flac_path)
assert "REPLAYGAIN_TRACK_GAIN" not in audio and "REPLAYGAIN_TRACK_PEAK" not in audio
assert audio["REPLAYGAIN_ALBUM_GAIN"] == ["-3.00 dB"]
def test_opus_gets_no_replaygain(tmp_path: Path) -> None:
"""RFC 7845 section 5.2.1: an Opus file should not carry REPLAYGAIN_* tags."""
path = tmp_path / "04. Example Track.opus"
subprocess.run(
["ffmpeg", "-nostdin", "-v", "error", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=mono", "-t", "0.2", str(path)],
check=True,
)
song = make_song()
song.data = {"replaygain_track_gain": -7.25}
write_music_metadata(path, song)
audio = OggOpus(path)
assert audio["TITLE"] == ["Example Track"]
assert not [key for key in audio.keys() if key.upper().startswith("REPLAYGAIN_")]
tests/core/test_tag_file_dispatch.py
+21
View File
@@ -118,3 +118,24 @@ def test_a_movie_or_episode_still_takes_the_mkv_path(calls: Calls, title: Any) -
assert calls.music == [], "write_music_metadata must not see a Matroska file"
assert [call[0] for call in calls.mkv] == [path]
def test_a_dts_song_in_matroska_takes_mkvpropedit(calls: Calls) -> None:
path = Path("04. Example Track.mka")
tags_module.tag_file(path, make_song(), session=SESSION)
assert calls.music == []
assert calls.mkv and calls.mkv[0][1]["TITLE"] == "Example Track"
def test_a_matroska_tagger_fault_warns_and_never_raises(
calls: Calls, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
def broken_apply_tags(path: Path, tags: dict[str, str]) -> None:
raise OSError("mkvpropedit not found")
monkeypatch.setattr(tags_module, "apply_tags", broken_apply_tags)
with caplog.at_level("WARNING", logger="TAGS"):
tags_module.tag_file(Path("04. Example Track.mka"), make_song(), session=SESSION)
assert "mkvpropedit not found" in caplog.text
tests/orchestration/test_music_remux_flow.py
+44 -2
View File
@@ -20,9 +20,10 @@ from pymediainfo import MediaInfo
from tests.core.test_music_container import fragmented_mp4, header_length
from tests.core.test_music_tagger_roundtrip import make_song
from unshackle.commands.dl import remux_music_tracks
from unshackle.commands.dl import mux_skipped, no_mux_extension, remux_music_tracks
from unshackle.core.config import config
from unshackle.core.tracks import Audio, Tracks
from unshackle.core.titles.movie import Movie
from unshackle.core.tracks import Audio, Tracks, Video
from unshackle.core.utils import tags
pytestmark = pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="requires ffmpeg")
@@ -120,3 +121,44 @@ def test_a_fragmented_flac_song_ends_playable_named_and_tagged(tmp_path: Path, d
audio = FLAC(final_path)
assert audio["TITLE"] == ["Example Track"]
assert audio["ARTIST"] == ["Example Artist"]
def test_the_hook_cuts_a_song_to_the_sample_window(tmp_path: Path) -> None:
song = make_song()
track = make_audio(fragmented_mp4(tmp_path / "Audio_a1.mp4", "flac"), Audio.Codec.FLAC, "a1")
track.sample_start = 10.0
song.tracks = Tracks([track])
remux_music_tracks(song, log, sample=(11, 13))
assert header_length(track.path) == pytest.approx(2.0, abs=0.3)
def test_an_mp3_song_is_named_mp3(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""MediaInfo calls MP3 'MPEG Audio', which named a file …MPEG.Audio2.0.mp3."""
monkeypatch.setattr(config, "output_template", {"songs": "{track_number}.{title}.{audio}"})
song = make_song()
track = make_audio(fragmented_mp4(tmp_path / "Audio_a1.mp4", "libmp3lame"), Audio.Codec.MP3, "a1")
song.tracks = Tracks([track])
remux_music_tracks(song, log)
assert song.get_filename(MediaInfo.parse(track.path), show_service=False).endswith(".MP3")
def test_no_mux_never_skips_the_song_path() -> None:
"""A --no-mux Song must still be renamed, tagged and get its .lrc, like any other Song."""
movie = Movie(id_="movie-0001", service=type("EXAMPLE", (), {}), name="Some Movie", year=2024)
assert mux_skipped(True, make_song()) is False
assert mux_skipped(True, movie) is True
assert mux_skipped(False, movie) is False
def test_a_no_mux_audio_track_in_mpeg_4_is_saved_as_m4a(tmp_path: Path) -> None:
path = fragmented_mp4(tmp_path / "Audio_a1.mp4", "aac")
media_info = MediaInfo.parse(path)
video = Video(id_="v1", url="https://example.test/v1.mp4", language="en", codec=Video.Codec.AVC)
assert no_mux_extension(make_audio(path, Audio.Codec.AAC, "a1"), path, media_info) == ".m4a"
assert no_mux_extension(video, path, media_info) == ".mp4"
tests/remote/unit/test_api_title_fields.py
+44
View File
@@ -20,6 +20,7 @@ from unshackle.core.api.handlers import serialize_title, stamp_service_flags
from unshackle.core.remote_service import build_title
from unshackle.core.titles.episode import Episode
from unshackle.core.titles.movie import Movie
from unshackle.core.titles.music import Song
pytestmark = pytest.mark.unit
@@ -213,3 +214,46 @@ def test_title_to_meta_round_trip_keeps_the_movie_anime_flag() -> None:
movie.anime = True
rebuilt = build_title(dl.title_to_meta(movie), "FAKE", "fallback-id")
assert rebuilt.anime is True
def test_song_tag_data_survives_the_remote_round_trip() -> None:
song = Song(
id_="song-0001",
service=_FakeSvc,
name="Track 1",
artist="A",
album="B",
track=1,
data={
"release_date": "2020-01-02",
"composer": "C",
"track_id": "123",
"hires": True,
"atmos": True,
"unrelated": object(),
},
)
rebuilt = build_title(json.loads(json.dumps(serialize_title(song))), "EXAMPLE", "song-0001")
assert rebuilt.data["release_date"] == "2020-01-02"
assert rebuilt.data["composer"] == "C"
assert rebuilt.data["track_id"] == "123"
assert rebuilt.data["hires"] is True and rebuilt.data["atmos"] is True
assert "unrelated" not in rebuilt.data
@pytest.mark.parametrize(
"music_data",
[["not", "a", "dict"], {"metadata": {"title": "Replaced"}, "composer": "C"}],
ids=["not-a-dict", "unknown-key"],
)
def test_the_client_keeps_only_known_music_data_keys(music_data: Any) -> None:
song = Song(id_="song-0001", service=_FakeSvc, name="Track 1", artist="A", album="B", track=1)
info = json.loads(json.dumps(serialize_title(song)))
info["music_data"] = music_data
rebuilt = build_title(info, "EXAMPLE", "song-0001")
assert "metadata" not in (rebuilt.data or {})
assert rebuilt.name == "Track 1"
unshackle/commands/dl.py
+22 -45
View File
@@ -153,7 +153,7 @@ from unshackle.core.utils.hdr10plus import probed_tracks
from unshackle.core.utils.post_scripts import NO_POST_SCRIPTS, build_context, dispatch, season_context
from unshackle.core.utils.redact import mask_proxy, redact_secrets
from unshackle.core.utils.selector import select_multiple
from unshackle.core.utils.subprocess import ffmpeg_decodes, ffprobe, run_step
from unshackle.core.utils.subprocess import ffmpeg_decodes, ffprobe
from unshackle.core.vault import Vault
from unshackle.core.vaults import Vaults
@@ -501,8 +501,8 @@ def post_script_group(candidate: Any) -> Any:
return ("title", id(candidate))
def remux_music_tracks(title: Any, log: logging.Logger) -> None:
"""Remux each audio track of a Song into the container of its codec.
def remux_music_tracks(title: Any, log: logging.Logger, sample: Optional[tuple[int, int]] = None) -> None:
"""Remux each audio track of a Song into the container of its codec, cut to ``sample`` when given.
A Song never reaches the muxer, so each downloaded audio track is itself a delivered
file. A failure warns, keeps that downloaded track, and moves on to the next one: the
@@ -511,45 +511,22 @@ def remux_music_tracks(title: Any, log: logging.Logger) -> None:
if not isinstance(title, Song):
return
for track in title.tracks.audio:
window = (max(sample[0] - track.sample_start, 0), sample[1] - sample[0]) if sample else (0.0, None)
try:
track.to_music_container()
track.to_music_container(*window)
except Exception as e: # noqa: BLE001
log.warning(f"Could not remux audio track {track.id} for music: {e}")
def cut_music_sample(title: Any, window: tuple[int, int], log: logging.Logger) -> None:
"""Cut each audio track of a Song to the sample window. A Song has no mux step that cuts it."""
if not isinstance(title, Song) or not binaries.FFMPEG:
return
for track in title.tracks.audio:
if not track.path or not track.path.exists():
continue
cut_path = track.path.with_stem(f"{track.path.stem}_cut")
try:
run_step(
[
binaries.FFMPEG,
"-v",
"error",
"-y",
"-ss",
f"{max(window[0] - track.sample_start, 0):.3f}",
"-i",
track.path,
"-t",
str(window[1] - window[0]),
"-map",
"0",
"-c",
"copy",
cut_path,
],
output=cut_path,
label="music sample cut",
)
os.replace(cut_path, track.path)
except RuntimeError as e:
log.warning(f"Could not cut audio track {track.id} to the sample window, so it holds whole segments: {e}")
def mux_skipped(no_mux: bool, title: Any) -> bool:
"""A Song has no mux step, so --no-mux never applies to it."""
return no_mux and not isinstance(title, Song)
def no_mux_extension(track: Any, path: Path, media_info: MediaInfo) -> str:
"""The extension of a --no-mux track file: an audio track in an MPEG-4 file is saved as .m4a."""
container = media_info.general_tracks[0].format if media_info.general_tracks else None
return ".m4a" if isinstance(track, Audio) and path.suffix == ".mp4" and container == "MPEG-4" else path.suffix
def apply_service_dl_overrides(ctx: click.Context, service_dl_config: dict[str, Any], log: logging.Logger) -> None:
@@ -2382,7 +2359,7 @@ class dl:
if progress_sink:
progress_sink({"skipped_titles": list(self.skipped_titles)})
self.log.warning(f"Skipping {title}: {reason}")
if not no_mux:
if not mux_skipped(no_mux, title):
settle_post_script_group(title)
for i, title in enumerate(titles):
@@ -3605,16 +3582,15 @@ class dl:
except Exception as e: # noqa: BLE001
self.log.warning(f"VUI normalization skipped for {track.id}: {e}")
remux_music_tracks(title, self.log)
if sample:
cut_music_sample(title, sample, self.log)
remux_music_tracks(title, self.log, sample)
muxed_paths = []
muxed_audio_codecs: dict[Path, Optional[Audio.Codec]] = {}
tagged_paths: set[Path] = set()
append_audio_codec_suffix = True
skip_mux = mux_skipped(no_mux, title)
if no_mux:
if skip_mux:
for track in title.tracks:
if track.path and track.path.exists():
muxed_paths.append(track.path)
@@ -3941,7 +3917,7 @@ class dl:
else:
muxed_paths.extend(track.path for track in title.tracks.audio if track.path and track.path.exists())
if no_mux:
if skip_mux:
final_dir = self.output_dir or config.directories.downloads
if not no_folder and (
isinstance(title, (Episode, Song))
@@ -3987,7 +3963,8 @@ class dl:
else:
track_suffix = ""
final_path = final_dir / f"{base_filename}{track_suffix}{track_path.suffix}"
ext = no_mux_extension(track, track_path, media_info)
final_path = final_dir / f"{base_filename}{track_suffix}{ext}"
else:
final_path = final_dir / f"{base_filename}{track_path.suffix}"
@@ -4068,7 +4045,7 @@ class dl:
if final_path.parent not in post_script_folders:
post_script_folders.append(final_path.parent)
if not no_mux:
if not skip_mux:
settle_post_script_group(title)
title_dl_time = time_elapsed_since(dl_start_time)
unshackle/core/api/handlers.py
+3 -1
View File
@@ -43,7 +43,7 @@ from unshackle.core.proxies.resolve import initialize_proxy_providers, is_loopba
from unshackle.core.services import Services
from unshackle.core.session import RnetSession, utf8_text
from unshackle.core.titles import Episode, Movie, Song, Title_T
from unshackle.core.titles.music import SONG_FIELDS
from unshackle.core.titles.music import MUSIC_DATA_KEYS, SONG_FIELDS, song_metadata
from unshackle.core.titles.title import rename
from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
from unshackle.core.tunnel import (
@@ -1577,6 +1577,8 @@ def serialize_title(title: Title_T) -> Dict[str, Any]:
if is_song:
for field in SONG_FIELDS:
result[field] = getattr(title, field, None)
metadata = song_metadata(title)
result["music_data"] = {key: metadata[key] for key in MUSIC_DATA_KEYS if metadata.get(key) not in (None, "")}
if isinstance(title, (Episode, Movie)) and getattr(title, "anime", None) is not None:
result["anime"] = title.anime
unshackle/core/constants.py
+1
View File
@@ -38,6 +38,7 @@ AUDIO_CODEC_MAP = {
"DTS-UHD": "DTS-X",
"DTS LBR": "DTS-LBR",
"MPEG-H 3D Audio": "MPEG-H",
"MPEG Audio Layer 3": "MP3",
}
SPACED_AUDIO_CODECS = {"DTS-X"}
unshackle/core/music/tagger.py
+111 -33
View File
@@ -1,13 +1,20 @@
from __future__ import annotations
import logging
import math
import subprocess
import time
from base64 import b64encode
from dataclasses import dataclass
from io import BytesIO
from pathlib import Path
from typing import Any, Optional
from typing import Any, Optional, Union
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.titles.music import song_metadata
from unshackle.core.utils.collections import as_list
from unshackle.core.utils.subprocess import log_tool_run
try:
from PIL import Image
@@ -38,9 +45,10 @@ try:
from mutagen.mp4 import MP4, MP4Cover
from mutagen.oggopus import OggOpus
from mutagen.oggvorbis import OggVorbis
from mutagen.wave import WAVE
except ImportError: # pragma: no cover - optional tagging dependency
FLAC = Picture = None # type: ignore[assignment, misc]
MP3 = MP4 = MP4Cover = OggOpus = OggVorbis = None # type: ignore[assignment, misc]
MP3 = MP4 = MP4Cover = OggOpus = OggVorbis = WAVE = None # type: ignore[assignment, misc]
ID3NoHeaderError = Exception # type: ignore[assignment, misc]
APIC = COMM = TALB = TCOM = TCON = TCOP = TDRC = TIT2 = TPE1 = TPE2 = TPOS = TRCK = TSRC = TXXX = None # type: ignore[assignment, misc]
USLT = None # type: ignore[assignment, misc]
@@ -48,6 +56,9 @@ except ImportError: # pragma: no cover - optional tagging dependency
log = logging.getLogger("MUSIC_TAGGER")
TagValues = dict[str, Union[str, list[str]]]
MAX_COVER_BYTES = 2**24 - 1 # a FLAC metadata block length is 24 bits
@dataclass
class MusicMetadataResult:
@@ -66,15 +77,18 @@ class MusicMetadataResult:
def write_music_metadata(path: Path, song: Any, *, session: Any = None, source_md5: str = "") -> MusicMetadataResult:
"""Write normalized music metadata for FLAC, Ogg, Opus, MP3, and M4A/MP4 audio files."""
"""Write normalized music metadata for FLAC, Ogg, Opus, MP3, WAV, and M4A/MP4 audio files.
Opus files get no ``REPLAYGAIN_*`` tags, because RFC 7845 section 5.2.1 says an Opus file should not carry them.
"""
path = Path(path)
extension = path.suffix.lower()
if extension not in {".flac", ".mp3", ".m4a", ".mp4", ".ogg", ".opus"}:
if extension not in {".flac", ".mp3", ".wav", ".m4a", ".mp4", ".ogg", ".opus"}:
return MusicMetadataResult(skipped=True, reason=f"Unsupported music metadata container: {extension}")
if extension == ".flac" and FLAC is None:
return MusicMetadataResult(skipped=True, reason="install mutagen to write FLAC tags")
if extension == ".mp3" and MP3 is None:
if extension in {".mp3", ".wav"} and MP3 is None:
return MusicMetadataResult(skipped=True, reason="install mutagen to write MP3 tags")
if extension in {".m4a", ".mp4"} and MP4 is None:
return MusicMetadataResult(skipped=True, reason="install mutagen to write MP4/M4A tags")
@@ -82,30 +96,35 @@ def write_music_metadata(path: Path, song: Any, *, session: Any = None, source_m
return MusicMetadataResult(skipped=True, reason="install mutagen to write Ogg tags")
tags = build_tag_values(song, source_md5=source_md5)
artwork_url = first_text(getattr(song, "artwork_url", None), _metadata(song).get("artwork_url"))
artwork_url = first_text(getattr(song, "artwork_url", None), song_metadata(song).get("artwork_url"))
cover_data, mime_type = download_cover(session, artwork_url)
if not cover_data:
cover_data, mime_type = attachment_cover(song)
cover_data, mime_type = usable_cover(cover_data, mime_type, path.name)
if extension == ".flac":
write_flac_tags(path, tags, cover_data, mime_type)
elif extension in {".ogg", ".opus"}:
if extension == ".opus":
tags = {key: value for key, value in tags.items() if not key.startswith("REPLAYGAIN_")}
write_ogg_tags(path, tags, cover_data, mime_type, OggOpus if extension == ".opus" else OggVorbis)
elif extension == ".mp3":
write_mp3_tags(path, tags, cover_data, mime_type)
elif extension in {".mp3", ".wav"}:
write_mp3_tags(path, tags, cover_data, mime_type, WAVE if extension == ".wav" else MP3)
else:
write_mp4_tags(path, tags, cover_data, mime_type)
return MusicMetadataResult(written=True, artwork_embedded=bool(cover_data))
def build_tag_values(song: Any, *, source_md5: str = "") -> dict[str, str]:
metadata = _metadata(song)
def build_tag_values(song: Any, *, source_md5: str = "") -> TagValues:
metadata = song_metadata(song)
track_number = string_tag(getattr(song, "track", None) or metadata.get("track_number"))
track_total = string_tag(getattr(song, "total_tracks", None) or metadata.get("total_tracks"))
disc_number = string_tag(getattr(song, "disc", None) or metadata.get("disc_number"))
disc_total = string_tag(getattr(song, "total_discs", None) or metadata.get("total_discs"))
explicit = as_bool(getattr(song, "explicit", None), metadata.get("explicit"), metadata.get("parental_warning"))
tags = {
tags: TagValues = {
"TITLE": first_text(getattr(song, "name", None), metadata.get("title")),
"ARTIST": first_text(getattr(song, "artist", None), metadata.get("artist"), metadata.get("performer")),
"ALBUM": first_text(getattr(song, "album", None), metadata.get("album")),
@@ -124,12 +143,27 @@ def build_tag_values(song: Any, *, source_md5: str = "") -> dict[str, str]:
"UPC": string_tag(getattr(song, "upc", None) or metadata.get("upc") or metadata.get("barcode")),
"COPYRIGHT": string_tag(getattr(song, "copyright", None) or metadata.get("copyright")),
"LABEL": first_text(getattr(song, "label", None), metadata.get("label")),
"COMMENT": first_text(metadata.get("comment"), metadata.get("quality")),
"COMMENT": first_text(metadata.get("comment")),
"LYRICS": first_text(getattr(song, "lyrics", None), metadata.get("lyrics")),
"SOURCE": first_text(metadata.get("source"), metadata.get("service")),
"ENCODEDBY": "Unshackle",
"UNSHACKLE_SOURCE_MD5": source_md5,
}
artists = metadata.get("artists")
if isinstance(artists, list) and len(artists) > 1:
tags["ARTISTS"] = [text for text in map(first_text, artists) if text]
for key, fmt in (
("track_gain", "{:.2f} dB"),
("track_peak", "{:.6f}"),
("album_gain", "{:.2f} dB"),
("album_peak", "{:.6f}"),
):
try:
value = float(metadata[f"replaygain_{key}"])
except (KeyError, TypeError, ValueError):
continue
if math.isfinite(value):
tags[f"REPLAYGAIN_{key.upper()}"] = fmt.format(value)
if explicit:
tags["EXPLICIT"] = "1"
tags["ITUNESADVISORY"] = "1"
@@ -151,10 +185,10 @@ def build_tag_values(song: Any, *, source_md5: str = "") -> dict[str, str]:
return {key: value for key, value in tags.items() if value}
def write_flac_tags(path: Path, tags: dict[str, str], cover_data: Optional[bytes], mime_type: str) -> None:
def write_flac_tags(path: Path, tags: TagValues, cover_data: Optional[bytes], mime_type: str) -> None:
audio = FLAC(path)
for key, value in tags.items():
audio[key] = [value]
audio[key] = as_list(value)
picture = build_picture(cover_data, mime_type)
if picture is not None:
audio.clear_pictures()
@@ -182,11 +216,11 @@ def build_picture(cover_data: Optional[bytes], mime_type: str) -> Optional[Any]:
def write_ogg_tags(
path: Path, tags: dict[str, str], cover_data: Optional[bytes], mime_type: str, reader: Any = None
path: Path, tags: TagValues, cover_data: Optional[bytes], mime_type: str, reader: Any = None
) -> None:
audio = (reader or OggVorbis)(path)
for key, value in tags.items():
audio[key] = [value]
audio[key] = as_list(value)
picture = build_picture(cover_data, mime_type)
if picture is not None:
# Ogg carries no picture block of its own, so the FLAC block goes in base64
@@ -194,11 +228,12 @@ def write_ogg_tags(
audio.save()
def write_mp3_tags(path: Path, tags: dict[str, str], cover_data: Optional[bytes], mime_type: str) -> None:
def write_mp3_tags(path: Path, tags: TagValues, cover_data: Optional[bytes], mime_type: str, reader: Any) -> None:
"""Write ID3 frames; ``reader`` is mutagen's MP3 or WAVE, both of which carry ID3."""
try:
audio = MP3(path)
audio = reader(path)
except ID3NoHeaderError:
audio = MP3(path)
audio = reader(path)
audio.add_tags()
if audio.tags is None:
audio.add_tags()
@@ -220,7 +255,7 @@ def write_mp3_tags(path: Path, tags: dict[str, str], cover_data: Optional[bytes]
for key, (frame_id, frame_class) in frame_map.items():
value = tags.get(key)
if value and frame_class is not None:
audio.tags.setall(frame_id, [frame_class(encoding=3, text=[value])])
audio.tags.setall(frame_id, [frame_class(encoding=3, text=as_list(value))])
if tags.get("COMMENT") and COMM is not None:
audio.tags.setall("COMM", [COMM(encoding=3, lang="eng", desc="", text=[tags["COMMENT"]])])
if tags.get("LYRICS") and USLT is not None:
@@ -233,11 +268,11 @@ def write_mp3_tags(path: Path, tags: dict[str, str], cover_data: Optional[bytes]
for key in sorted(custom_keys):
if TXXX is not None and tags[key]:
audio.tags.delall(f"TXXX:{key}")
audio.tags.add(TXXX(encoding=3, desc=key, text=[tags[key]]))
audio.tags.add(TXXX(encoding=3, desc=key, text=as_list(tags[key])))
audio.save()
def write_mp4_tags(path: Path, tags: dict[str, str], cover_data: Optional[bytes], mime_type: str) -> None:
def write_mp4_tags(path: Path, tags: TagValues, cover_data: Optional[bytes], mime_type: str) -> None:
audio = MP4(path)
if tags.get("TITLE"):
audio["\xa9nam"] = [tags["TITLE"]]
@@ -260,10 +295,10 @@ def write_mp4_tags(path: Path, tags: dict[str, str], cover_data: Optional[bytes]
if tags.get("LYRICS"):
audio["\xa9lyr"] = [tags["LYRICS"]]
track_number, track_total = split_number_pair(tags.get("TRACKNUMBER", ""))
track_number, track_total = split_number_pair(str(tags.get("TRACKNUMBER", "")))
if track_number:
audio["trkn"] = [(track_number, track_total or 0)]
disc_number, disc_total = split_number_pair(tags.get("DISCNUMBER", ""))
disc_number, disc_total = split_number_pair(str(tags.get("DISCNUMBER", "")))
if disc_number:
audio["disk"] = [(disc_number, disc_total or 0)]
@@ -287,7 +322,7 @@ def write_mp4_tags(path: Path, tags: dict[str, str], cover_data: Optional[bytes]
for key in sorted(set(tags) - mapped_keys):
value = tags.get(key)
if value:
audio[f"----:com.apple.iTunes:{key}"] = [str(value).encode("utf-8")]
audio[f"----:com.apple.iTunes:{key}"] = [str(item).encode("utf-8") for item in as_list(value)]
audio.save()
@@ -307,14 +342,57 @@ def download_cover(session: Any, artwork_url: str) -> tuple[Optional[bytes], str
return None, ""
def _metadata(song: Any) -> dict[str, Any]:
data = getattr(song, "data", None)
if not isinstance(data, dict):
return {}
# the renderer reads a nested "metadata" sub-dict as well as the top level, so a
# service that follows it would otherwise lose every tag read through this helper
nested = data.get("metadata")
return {**data, **nested} if isinstance(nested, dict) else data
def attachment_cover(song: Any) -> tuple[Optional[bytes], str]:
"""Return the first downloaded image attachment of the Song, for a service that gives no artwork_url."""
tracks = getattr(song, "tracks", None)
for attachment in getattr(tracks, "attachments", None) or []:
path = getattr(attachment, "path", None)
if path and Path(path).exists() and str(getattr(attachment, "mime_type", "")).startswith("image/"):
data = Path(path).read_bytes()
return data, mime_from_image(data, attachment.mime_type)
return None, ""
def usable_cover(cover_data: Optional[bytes], mime_type: str, name: str) -> tuple[Optional[bytes], str]:
"""Convert a WebP cover to JPEG, which every container takes, and drop one too large to embed."""
if cover_data and mime_type == "image/webp":
try:
if not binaries.FFMPEG:
raise OSError("FFmpeg not found")
convert_start = time.monotonic()
converted = subprocess.run(
[
binaries.FFMPEG,
"-v",
"error",
"-i",
"pipe:",
"-frames:v",
"1",
"-f",
"image2pipe",
"-c:v",
"mjpeg",
"pipe:",
],
input=cover_data,
capture_output=True,
)
log_tool_run(
"cover webp to jpeg",
"ffmpeg",
converted.returncode,
duration_ms=round((time.monotonic() - convert_start) * 1000, 1),
)
converted.check_returncode()
cover_data, mime_type = converted.stdout, "image/jpeg"
except (OSError, subprocess.CalledProcessError):
log.warning(f"Skipped cover art for {name}: the WebP image could not be converted to JPEG")
return None, ""
if cover_data and len(cover_data) > MAX_COVER_BYTES:
log.warning(f"Skipped cover art for {name}: {len(cover_data)} bytes is over the 16 MiB limit")
return None, ""
return cover_data, mime_type
def first_text(*values: Any) -> str:
unshackle/core/remote_service.py
+4 -1
View File
@@ -50,7 +50,7 @@ from unshackle.core.session import MAX_BACKOFF
from unshackle.core.titles import Title_T, Titles_T, remap_titles
from unshackle.core.titles.episode import Episode, Series
from unshackle.core.titles.movie import Movie, Movies
from unshackle.core.titles.music import SONG_FIELDS, Album, Song
from unshackle.core.titles.music import MUSIC_DATA_KEYS, SONG_FIELDS, Album, Song
from unshackle.core.tracks import Audio, Chapter, Chapters, Subtitle, Tracks, Video
from unshackle.core.tracks.attachment import Attachment
from unshackle.core.tracks.track import Track
@@ -606,6 +606,9 @@ def apply_title_fields(title: Title_T, info: Dict[str, Any]) -> None:
setattr(title, key, info[key])
extra = {k: info[k] for k in ("date", "cover_url") if info.get(k) is not None}
music_data = info.get("music_data")
if isinstance(music_data, dict):
extra.update({key: music_data[key] for key in MUSIC_DATA_KEYS if key in music_data})
if extra:
if isinstance(title.data, dict):
title.data.update(extra)
unshackle/core/titles/music.py
+40
View File
@@ -34,6 +34,46 @@ SONG_FIELDS = (
)
"""The `Song` attributes a remote session sends, beyond the fields every title has."""
MUSIC_DATA_KEYS = (
"release_date",
"composer",
"performer",
"comment",
"source",
"service",
"track_id",
"album_id",
"track_url",
"album_url",
"artists",
"replaygain_track_gain",
"replaygain_track_peak",
"replaygain_album_gain",
"replaygain_album_peak",
"quality",
"quality_label",
"lossless",
"hires",
"atmos",
"duration",
"released_at",
"year",
"barcode",
"parental_warning",
"unavailable_reason",
"skip_reason",
)
"""The `Song.data` keys the music tagger and renderer read, which a remote session sends as `music_data`."""
def song_metadata(song: Any) -> dict[str, Any]:
"""``Song.data`` with its nested ``metadata`` dict merged over it, as the renderer also reads both."""
data = getattr(song, "data", None)
if not isinstance(data, dict):
return {}
nested = data.get("metadata")
return {**data, **nested} if isinstance(nested, dict) else data
class Song(Title):
def __init__(
unshackle/core/titles/title.py
+2 -1
View File
@@ -186,7 +186,8 @@ class Title:
channels_str = f"{len(positions) - heights - lfe}.{lfe}.{heights}"
else:
channels_str = f"{channels:.1f}"
builtin = AUDIO_CODEC_MAP.get(codec, codec)
profile = getattr(primary_audio_track, "format_profile", None)
builtin = AUDIO_CODEC_MAP.get(f"{codec} {profile}") or AUDIO_CODEC_MAP.get(codec, codec)
audio = rename(builtin)
spacer = " " if builtin in SPACED_AUDIO_CODECS or audio[-1:].isdigit() else ""
unshackle/core/tracks/audio.py
+196 -63
View File
@@ -1,13 +1,26 @@
from __future__ import annotations
import json
import logging
import math
import os
import re
import subprocess
import time
from enum import Enum
from pathlib import Path
from typing import Any, Optional, Union
from unshackle.core import binaries
from unshackle.core.tracks.track import MPEGH_SAMPLE_ENTRIES, Track
from unshackle.core.tracks.track import (
MPEGH_SAMPLE_ENTRIES,
Track,
has_mp4_only_audio_entry,
prepare_for_mp4box,
)
from unshackle.core.utils.subprocess import log_tool_run, run_step
log = logging.getLogger("Audio")
# the Dolby DASH channel configuration is the E-AC-3 chanmap (ETSI TS 102 366 Table E.1.4), read from the MSB
DOLBY_CHANNEL_BITS = (
@@ -78,8 +91,21 @@ class Audio(Track):
return Audio.Codec.ALAC
if mime == "flac":
return Audio.Codec.FLAC
if mime == "vorbis":
return Audio.Codec.OGG
if mime == "mp3":
return Audio.Codec.MP3
raise ValueError(f"The MIME '{mime}' is not a supported Audio Codec")
@staticmethod
def from_ffprobe(codec_name: str, profile: str = "") -> Optional[Audio.Codec]:
"""Map an FFprobe ``codec_name`` to a codec, or None for one core does not name."""
if codec_name.startswith("pcm_"):
return Audio.Codec.PCM
if codec_name == "dts":
return Audio.Codec.DTSHD if "DTS-HD" in profile or "Express" in profile else Audio.Codec.DTS
return FFPROBE_CODECS.get(codec_name)
@staticmethod
def from_codecs(codecs: str) -> Audio.Codec:
for codec in codecs.lower().split(","):
@@ -104,9 +130,8 @@ class Audio(Track):
Make a new Audio track object.
Parameters:
codec: An Audio.Codec enum representing the audio codec.
If not specified, unshackle uses MediaInfo to get the codec
after it downloads the track.
codec: An Audio.Codec enum representing the audio codec. The music remux
replaces it with the codec FFprobe reads from the downloaded file.
bitrate: A number or float representing the average bandwidth in bits/s.
unshackle rounds float values up to the nearest integer.
channels: A number, float, or string representing the number of audio channels.
@@ -269,77 +294,161 @@ class Audio(Track):
return float(sum(int(part) for part in channels.split(".")))
return float(channels)
def to_music_container(self) -> bool:
"""Remux the track into the container of its codec, as a standalone audio file.
def probe_stream(self) -> dict[str, Any]:
"""FFprobe codec fields of the first audio track plus the file ``duration``; {} when it cannot probe."""
if not binaries.FFProbe or not self.path or not self.path.exists():
return {}
probe_start = time.monotonic()
probe = subprocess.run(
[
str(binaries.FFProbe),
"-v",
"error",
"-select_streams",
"a:0",
"-show_entries",
"stream=codec_name,profile,sample_fmt,bits_per_raw_sample,sample_rate:format=duration",
"-of",
"json",
str(self.path),
],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
)
log_tool_run(
"ffprobe audio", "ffprobe", probe.returncode, duration_ms=round((time.monotonic() - probe_start) * 1000, 1)
)
try:
data = json.loads(probe.stdout)
stream: dict[str, Any] = data["streams"][0]
stream["duration"] = float((data.get("format") or {}).get("duration") or 0)
except (ValueError, KeyError, IndexError, TypeError):
return {}
return stream
Music titles never reach the muxer, so the downloaded file is the delivered file, and
after decryption that is still a fragmented MP4. Its header states a length of zero, so
a player stops after the first fragment. FLAC inside an MP4 is also not a FLAC stream.
A rename cannot correct either fault. Returns True once the new file replaces the
downloaded one; every failure raises.
def to_music_container(self, start: float = 0.0, duration: Optional[float] = None) -> None:
"""Remux the track into the container of the codec FFprobe reads from the file, cut to the sample window.
The declared codec is used only when FFprobe cannot read the file. Every failure raises and keeps the
downloaded file.
"""
if not self.path or not self.path.exists():
raise ValueError("Cannot remux a Track that has not been downloaded.")
stream = self.probe_stream()
probed = Audio.Codec.from_ffprobe(stream.get("codec_name", ""), stream.get("profile", ""))
if probed and probed != self.codec:
log.debug(f"Audio track {self.id} declares {getattr(self.codec, 'value', None)} but holds {probed.value}")
self.codec = probed
if start > 0 and 0 < stream.get("duration", 0) <= start:
raise ValueError(f"Sample window starts at {start:.0f}s, after the {stream['duration']:.0f}s track ends")
source = self.path
if self.codec in MP4BOX_CODECS or has_mp4_only_audio_entry(source):
output = self.mp4box_music_container(source, start, duration)
else:
output = self.ffmpeg_music_container(source, stream, start, duration)
source.unlink()
self.path = output
def ffmpeg_music_container(
self, source: Path, stream: dict[str, Any], start: float, duration: Optional[float]
) -> Path:
"""Remux with FFmpeg. FLAC is encoded, never copied: FFmpeg's FLAC muxer writes a real sample count
only from an encoder. Float PCM and PCM above 24 bits go to WAV: FLAC has no float samples, and FFmpeg
writes 32-bit FLAC only as an experimental feature. 16- and 24-bit float, which FFmpeg cannot write,
widen to 32-bit float. DTS goes to Matroska, because FFmpeg writes it into
MP4 with an mp4a sample entry. Only AAC, ALAC and AC-3 take the ipod muxer (the M4A brand)."""
if not binaries.FFMPEG:
raise EnvironmentError('FFmpeg executable "ffmpeg" was not found but is required for this call.')
containers: dict[Optional[Audio.Codec], str] = {
Audio.Codec.FLAC: ".flac",
Audio.Codec.OPUS: ".opus",
Audio.Codec.OGG: ".ogg",
Audio.Codec.MP3: ".mp3",
}
extension = containers.get(self.codec, ".m4a")
original_path = self.path
output_path = original_path.with_name(f"{original_path.stem}_music{extension}")
sample_fmt = stream.get("sample_fmt")
lossless_pcm = (
self.codec == Audio.Codec.PCM
and sample_fmt not in ("flt", "fltp", "dbl", "dblp")
and (sample_fmt in ("u8", "s16", "s16p") or 4 <= int(stream.get("bits_per_raw_sample") or 0) <= 24)
)
if self.codec == Audio.Codec.FLAC or lossless_pcm:
extension, codec_args = ".flac", ["-c:a", "flac"]
if int(stream.get("sample_rate") or 0) > 384000:
codec_args += ["-frame_size", "4608"]
elif self.codec == Audio.Codec.PCM:
pcm = (stream.get("codec_name") or "pcm_s32le").replace("be", "le")
if pcm in ("pcm_f16le", "pcm_f24le"):
pcm = "pcm_f32le"
extension, codec_args = ".wav", ["-c:a", pcm]
elif self.codec in COPY_CONTAINERS:
extension, codec_args = COPY_CONTAINERS[self.codec], ["-c", "copy"]
else:
ipod = self.codec in (Audio.Codec.AAC, Audio.Codec.ALAC, Audio.Codec.AC3)
extension, codec_args = ".m4a", ["-c", "copy", "-movflags", "+faststart", *([] if ipod else ["-f", "mp4"])]
def ffmpeg(*extra_args: str) -> None:
subprocess.run(
[
str(binaries.FFMPEG),
"-nostdin",
"-hide_banner",
"-loglevel",
"error",
"-y",
"-i",
str(original_path),
"-map",
"0:a:0",
"-map_metadata",
"-1",
# a copy keeps the source STREAMINFO, whose sample count a fragmenting writer
# leaves at zero, so the FLAC states no length; the encoder writes a real one
*(["-c:a", "flac"] if extension == ".flac" else ["-c", "copy"]),
# the fragmented moov is what cuts playback short, so ask for one contiguous header
*(["-movflags", "+faststart"] if extension == ".m4a" else []),
*extra_args,
str(output_path),
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
output = source.with_name(f"{source.stem}_music{extension}")
run_step(
[
binaries.FFMPEG,
"-nostdin",
"-hide_banner",
"-loglevel",
"error",
"-y",
*(["-ss", f"{start:.3f}"] if start > 0 else []),
"-i",
source,
*(["-t", str(duration)] if duration else []),
"-map",
"0:a:0",
"-map_metadata",
"-1",
"-fflags",
"+bitexact",
"-flags:a",
"+bitexact",
*codec_args,
output,
],
output=output,
label="music remux",
)
return output
def mp4box_music_container(self, source: Path, start: float, duration: Optional[float]) -> Path:
"""Rewrite the track as an unfragmented .m4a with MP4Box, for the codecs FFmpeg cannot write into MP4."""
if not binaries.MP4Box:
raise EnvironmentError(f"MP4Box (GPAC) not found, the {getattr(self.codec, 'value', None)} remux needs it")
candidate = source.with_name(f"{source.stem}_init.mp4")
output = source.with_name(f"{source.stem}_music.m4a")
try:
try:
ffmpeg()
except subprocess.CalledProcessError as e:
if b"not currently supported in container" not in e.stderr:
raise
# a .m4a name picks FFmpeg's ipod muxer, which refuses Dolby Digital Plus and the
# other codecs Apple never put in an M4A; the plain mp4 muxer accepts them
ffmpeg("-f", "mp4")
except subprocess.CalledProcessError:
# FFmpeg creates the output before it fails, and nothing sweeps the shared temp
# directory for it, so a failed run must remove its own partial file
output_path.unlink(missing_ok=True)
prepared = prepare_for_mp4box(source, candidate, self.id)
run_step(
[binaries.MP4Box, "-for-test", "-add", f"{prepared}#audio", "-new", output],
output=output,
label="music MP4Box remux",
)
if duration:
cut = source.with_name(f"{source.stem}_cut.m4a")
run_step(
[
binaries.MP4Box,
"-for-test",
"-splitx",
f"{start:.3f}:{start + duration:.3f}",
output,
"-out",
cut,
],
output=cut,
label="music MP4Box sample cut",
)
os.replace(cut, output)
except Exception:
output.unlink(missing_ok=True)
raise
original_path.unlink()
self.path = output_path
return True
finally:
candidate.unlink(missing_ok=True)
return output
def get_track_name(self) -> Optional[str]:
"""Return the base Track Name."""
@@ -352,4 +461,28 @@ class Audio(Track):
return track_name or None
FFPROBE_CODECS: dict[str, Audio.Codec] = {
"aac": Audio.Codec.AAC,
"alac": Audio.Codec.ALAC,
"flac": Audio.Codec.FLAC,
"opus": Audio.Codec.OPUS,
"vorbis": Audio.Codec.OGG,
"mp3": Audio.Codec.MP3,
"ac3": Audio.Codec.AC3,
"eac3": Audio.Codec.EC3,
"ac4": Audio.Codec.AC4,
"mpegh_3d_audio": Audio.Codec.MPEGH,
}
COPY_CONTAINERS: dict[Optional[Audio.Codec], str] = {
Audio.Codec.OPUS: ".opus",
Audio.Codec.OGG: ".ogg",
Audio.Codec.MP3: ".mp3",
Audio.Codec.DTS: ".mka",
Audio.Codec.DTSHD: ".mka",
}
MP4BOX_CODECS = (Audio.Codec.AC4, Audio.Codec.MPEGH, Audio.Codec.DTSX)
__all__ = ("Audio",)
unshackle/core/tracks/track.py
+27
View File
@@ -203,6 +203,33 @@ def fix_data_reference_indexes(path: Path) -> int:
MPEGH_SAMPLE_ENTRIES = ("mha1", "mha2", "mhm1", "mhm2")
def prepare_for_mp4box(source: Path, candidate: Path, track_id: Any) -> Path:
"""Return the file MP4Box reads: ``candidate`` when repeated init boxes were dropped into it, else ``source``.
The returned file also has its sample entries pointed at data reference 1.
"""
dropped = strip_duplicate_init_boxes(source, candidate)
if dropped:
log_event(
"mux_mp4_init_deduplicated",
level="DEBUG",
message=f"Dropped {dropped} repeated init box(es) from {source.name} for MP4Box",
context={"track_id": track_id, "dropped": dropped},
)
else:
candidate.unlink(missing_ok=True)
prepared = candidate if dropped else source
repointed = fix_data_reference_indexes(prepared)
if repointed:
log_event(
"mux_mp4_dref_fixed",
level="DEBUG",
message=f"Pointed {repointed} sample entry(s) of {source.name} at data reference 1 for MP4Box",
context={"track_id": track_id, "entries": repointed},
)
return prepared
def has_mp4_only_audio_entry(path: Path) -> bool:
"""True if the MP4 holds a DTS-UHD (DTS:X Profile 2) or MPEG-H audio sample entry.
unshackle/core/tracks/tracks.py
+3 -22
View File
@@ -29,9 +29,8 @@ from unshackle.core.tracks.chapters import Chapter, Chapters
from unshackle.core.tracks.subtitle import Subtitle
from unshackle.core.tracks.track import (
Track,
fix_data_reference_indexes,
has_mp4_only_audio_entry,
strip_duplicate_init_boxes,
prepare_for_mp4box,
)
from unshackle.core.tracks.video import Video
from unshackle.core.utilities import (
@@ -713,28 +712,10 @@ class Tracks:
continue
events.emit(events.Types.TRACK_MULTIPLEX, track=track)
source = track.path
candidate = config.directories.temp / f"mp4mux_{track.id}_{os.getpid()}_{threading.get_ident()}.mp4"
dropped = strip_duplicate_init_boxes(source, candidate)
if dropped:
source = prepare_for_mp4box(track.path, candidate, track.id)
if source == candidate:
cleaned.append(candidate)
source = candidate
log_event(
"mux_mp4_init_deduplicated",
level="DEBUG",
message=f"Dropped {dropped} repeated init box(es) from {track.path.name} for MP4Box",
context={"track_id": track.id, "dropped": dropped},
)
elif candidate.exists():
candidate.unlink()
repointed = fix_data_reference_indexes(source)
if repointed:
log_event(
"mux_mp4_dref_fixed",
level="DEBUG",
message=f"Pointed {repointed} sample entry(s) of {track.path.name} at data reference 1 for MP4Box",
context={"track_id": track.id, "entries": repointed},
)
delay = round((track.sample_start - sample_base) * 1000) if config.sample else 0
moov = read_top_level_box(source, b"moov")
unshackle/core/utils/tags.py
+5 -1
View File
@@ -10,7 +10,7 @@ from xml.sax.saxutils import escape
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.music.tagger import write_music_metadata
from unshackle.core.music.tagger import build_tag_values, write_music_metadata
from unshackle.core.providers import (
ExternalIds,
fuzzy_match,
@@ -98,6 +98,10 @@ def tag_file(
if isinstance(title, Song):
try:
if Path(path).suffix.lower() == ".mka":
values = build_tag_values(title).items()
apply_tags(path, {key: "; ".join(value) if isinstance(value, list) else value for key, value in values})
return
music_result = write_music_metadata(path, title, session=session)
except Exception as e: # the file already moved; a tagging fault must warn, not abort the run
log.warning("Music tagging failed for %s: %s", path.name, e)
unshackle/services/MUSIC_EXAMPLE/__init__.py
-20
View File
@@ -9,7 +9,6 @@ from typing import Any, Optional
import click
from unshackle.core.console import console
from unshackle.core.constants import AnyTrack
from unshackle.core.credential import Credential
from unshackle.core.music import (
build_music_from_songs,
@@ -62,7 +61,6 @@ class MUSIC_EXAMPLE(Service):
build_song the full Song field set, and what `data` feeds
get_tracks ONE Audio track per codec+bitrate the source offers
get_chapters empty, because a song has no chapters
on_track_downloaded restore the real container suffix before tagging
There is no music-specific Service API. A `Song` is an ordinary `Title` on the
one generic download path, so everything else a music service needs (auth,
@@ -460,24 +458,6 @@ class MUSIC_EXAMPLE(Service):
# base class requires the method and the pipeline reads it for every title.
return Chapters()
def on_track_downloaded(self, track: AnyTrack) -> None:
# Every track downloads to a temp file named `.mp4` whatever it holds, and a
# Song skips the muxer entirely, so its download IS the delivered file, suffix
# and all. The tagger picks its metadata format off that suffix, so restore the
# real container here or a FLAC ships as a `.mp4` the tagger then fails to open
# as one.
if not isinstance(track, Audio) or not track.path or not track.codec:
return
# This event fires twice, once before decryption and once at the end of the
# download. Decryption still reads the original path, so a track that is still
# encrypted has to wait for the second call, by which time drm is cleared.
if track.drm:
return
suffix = self.config["containers"].get(track.codec.name)
if not suffix or track.path.suffix == suffix:
return
track.path = track.path.rename(track.path.with_suffix(suffix))
# DRM is identical to any other service: implement get_widevine_license (and the
# PlayReady/ClearKey callbacks) and mark the Audio tracks with the DRM the API
# reports. See the EXAMPLE service for those; nothing about them is music-specific.
unshackle/services/MUSIC_EXAMPLE/config.yaml
-8
View File
@@ -24,11 +24,3 @@ client:
mobile:
user_agent: USER_AGENT
formats: [aac]
# Audio.Codec name -> the container the API delivers that codec in. A song skips
# the muxer, so on_track_downloaded restores these suffixes on the downloaded file
# and the tagger then writes the matching metadata format.
containers:
FLAC: .flac
ALAC: .m4a
AAC: .m4a