Files
2026-10-18 19:01:04 -07:00

590 lines
22 KiB
Python

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import logging
import re
import shutil
import subprocess
import tempfile
from pathlib import Path
logger = logging.getLogger(__name__)
# Inter-chapter gap inside the m4b. Kept equal to the AR package's leading/trailing silence
# (ar_export.SILENCE_MS) so every delivery uses one audible-pause value instead of three
# drifting ones. Imported rather than re-declared so the two cannot diverge again.
try:
from umbraaudio.assembly.ar_export import SILENCE_MS as CHAPTER_GAP_MS
except Exception: # pragma: no cover - ar_export pulls numpy/soundfile
CHAPTER_GAP_MS = 2000
# The m4b and the AR MP3 package must ship the same audio, so both read the
# same spec from ar_export. Previously the m4b had no loudness handling at all
# and no sample-rate flag, which is why it came out at 24 kHz / ~96 kbps while
# the MP3 masters were 44.1 kHz / 192 kbps CBR.
from umbraaudio.assembly.ar_export import ( # noqa: E402
NORM_MAX_GAIN_DB,
NORM_TARGET_RMS_DB,
PEAK_LIMIT_LINEAR,
PEAK_MAX_DB,
RMS_HI_DB,
RMS_LO_DB,
TARGET_SAMPLE_RATE,
)
# AAC overshoots the float clip ceiling by ~3 dB. Measured across clip ceilings
# on real chapter audio (-6.0 clip -> -3.10 encoded peak), so the margin is
# applied AT THE CLIP rather than as a post-hoc whole-signal trim: clipping to
# -3.0 and then trimming cost ~3.2 dB and pushed RMS to -23.8, outside the
# band. Clipping to -6.0 lands inside spec with no trim at all, and clipping
# barely moves RMS (-20.5 -> -20.7).
AAC_OVERSHOOT_MARGIN_DB = 3.0
# Fallback only, if the ceiling still is not enough: AAC overshoot is not
# monotonic (+0.8, +0.9, +1.6 dB across passes), so trim past the exact excess.
AAC_TRIM_SAFETY_DB = 0.5
AAC_ENFORCE_PASSES = 5
class AudiobookExporter:
def export_m4b(
self,
chapter_files: list[Path],
output_path: Path,
metadata: dict | None = None,
author: str = "",
title: str = "",
) -> Path:
output_path.parent.mkdir(parents=True, exist_ok=True)
chapter_groups = self._group_by_chapter(chapter_files)
chapter_wavs: list[Path] = []
tmp_dir = Path(tempfile.mkdtemp(prefix="umbra_m4b_"))
try:
for ch_dir, utt_files in chapter_groups.items():
if len(utt_files) == 1:
chapter_wavs.append(utt_files[0])
else:
stitched = tmp_dir / f"{ch_dir.name}.wav"
self._stitch_chapter(utt_files, stitched)
chapter_wavs.append(stitched)
if len(chapter_wavs) == 1:
concat_wav = chapter_wavs[0]
else:
concat_wav = tmp_dir / "concatenated.wav"
self._concat_wavs_with_silence(chapter_wavs, concat_wav, gap_ms=CHAPTER_GAP_MS)
m4b_no_meta = tmp_dir / "no_meta.m4b"
self._encode_aac_m4b(concat_wav, m4b_no_meta)
effective_title = title or (metadata.get("title", "") if metadata else "")
effective_author = author or (metadata.get("author", "") if metadata else "")
if effective_title or effective_author:
from umbraaudio.assembly.metadata import (
BookMetadata,
ChapterMeta,
write_ffmpeg_metadata,
)
chapter_titles = self._extract_chapter_titles(chapter_files, chapter_groups)
durations = self._get_wav_durations(chapter_wavs)
ch_meta = []
offset = 0.0
for i, (ct, dur) in enumerate(zip(chapter_titles, durations, strict=False)):
ch_meta.append(
ChapterMeta(
title=ct,
start_time_sec=offset,
end_time_sec=offset + dur,
index=i,
)
)
offset += dur
book_meta = BookMetadata(
title=effective_title,
author=effective_author,
chapters=ch_meta,
)
meta_file = tmp_dir / "metadata.txt"
write_ffmpeg_metadata(book_meta, meta_file)
self._apply_metadata(m4b_no_meta, meta_file, output_path)
else:
shutil.copy2(str(m4b_no_meta), str(output_path))
finally:
shutil.rmtree(tmp_dir, ignore_errors=True)
logger.info("Exported M4B: %s", output_path)
return output_path
def export_full_book_m4b(
self,
chapter_files: list[Path],
output_path: Path,
meta: dict | None = None,
cover_path: Path | None = None,
ffmpeg_path: str = "ffmpeg",
) -> Path:
"""Build a complete book .m4b from pre-assembled chapter files.
Concatenates the ``chN_full.wav`` chapters (one file per chapter) with a
short inter-chapter gap, encodes to AAC M4B, then embeds the full
publishing metadata (title, author, narrator, description/synopsis,
publisher, copyright, ISBN, language, genre, series, chapter markers)
and an attached-pic cover from ``meta``.
``chapter_files`` order is the final book order; align each with a
``meta["chapter_titles"]`` entry (falling back to "Chapter N") for the
chapter marker titles.
"""
from umbraaudio.assembly.metadata import (
BookMetadata,
ChapterMeta,
apply_ffmpeg_cover,
apply_ffmpeg_metadata,
write_ffmpeg_metadata,
)
output_path.parent.mkdir(parents=True, exist_ok=True)
meta = meta or {}
chapter_titles = meta.get("chapter_titles") or {}
tmp_dir = Path(tempfile.mkdtemp(prefix="umbra_m4b_full_"))
try:
# --- 1) Stage chapter wavs & durations ---------------------------------
staged: list[Path] = []
durations: list[float] = []
for i, f in enumerate(chapter_files):
staged.append(Path(f))
try:
import soundfile as sf
durations.append(sf.info(str(f)).duration)
except Exception:
durations.append(0.0)
# --- 2) Concatenate chapters with a small gap --------------------------
concat_wav = tmp_dir / "concatenated.wav"
self._concat_wavs_with_silence(
staged, concat_wav, gap_ms=int(meta.get("chapter_gap_ms", CHAPTER_GAP_MS))
)
# --- 3) Encode AAC M4B ------------------------------------------------
m4b_no_meta = tmp_dir / "no_meta.m4b"
self._encode_aac_m4b(concat_wav, m4b_no_meta)
# --- 4) Build full metadata + chapter markers --------------------------
ch_meta = []
offset = 0.0
for i, dur in enumerate(durations):
title = chapter_titles.get(chapter_files[i].name) or f"Chapter {i + 1}"
ch_meta.append(
ChapterMeta(title=title, start_time_sec=offset, end_time_sec=offset + dur, index=i)
)
offset += dur
book = BookMetadata(
title=meta.get("title", ""),
subtitle=meta.get("subtitle", ""),
author=meta.get("author", ""),
narrator=meta.get("narrator", ""),
description=meta.get("synopsis", "") or meta.get("description", ""),
language=meta.get("language", "eng"),
publisher=meta.get("publisher", ""),
copyright_holder=meta.get("copyright_holder", ""),
copyright_year=meta.get("copyright_year", ""),
year=meta.get("copyright_year", ""),
release_date=meta.get("release_date", ""),
series=meta.get("series", ""),
series_number=meta.get("series_number", ""),
genre=meta.get("genre", "Audiobook"),
keywords=meta.get("keywords", ""),
isbn=meta.get("isbn", ""),
explicit=meta.get("explicit", ""),
chapters=ch_meta,
cover_art=cover_path,
)
meta_file = tmp_dir / "metadata.txt"
write_ffmpeg_metadata(book, meta_file)
m4b_tagged = tmp_dir / "tagged.m4b"
self._apply_metadata(m4b_no_meta, meta_file, m4b_tagged)
# --- 5) Embed cover art ------------------------------------------------
effective_cover = cover_path
if effective_cover is not None and not Path(effective_cover).exists():
effective_cover = None
if effective_cover is not None:
apply_ffmpeg_cover(m4b_tagged, Path(effective_cover), output_path, ffmpeg_path=ffmpeg_path)
else:
shutil.copy2(str(m4b_tagged), str(output_path))
finally:
shutil.rmtree(tmp_dir, ignore_errors=True)
logger.info("Exported full-book M4B: %s", output_path)
return output_path
def _group_by_chapter(self, files: list[Path]) -> dict[Path, list[Path]]:
groups: dict[Path, list[Path]] = {}
for f in files:
parent = f.parent
if parent not in groups:
groups[parent] = []
groups[parent].append(f)
sorted_groups: dict[Path, list[Path]] = {}
for parent in sorted(groups.keys()):
sorted_groups[parent] = sorted(groups[parent])
return sorted_groups
def _stitch_chapter(self, utterance_files: list[Path], output_path: Path) -> Path:
from umbraaudio.assembly import AudioStitcher
# Match AudioStitcher's defaults, which is what ChapterBuilder (the
# approved, reviewed path) uses. The previous 0 meant the m4b stitched a
# chapter with different silence than the one a reviewer approved --
# the last hole in the pinned-set invariant (spec 22 SS7 gap 1).
stitcher = AudioStitcher(
silence_between_utterances_ms=500,
silence_between_chapters_ms=4000,
)
return stitcher.stitch_utterances(utterance_files, output_path)
def _concat_wavs_with_silence(
self,
wav_files: list[Path],
output_path: Path,
gap_ms: int = 4000,
sample_rate: int = TARGET_SAMPLE_RATE,
) -> Path:
import numpy as np
import soundfile as sf
segments = []
gap_samples = int(sample_rate * gap_ms / 1000)
gap = np.zeros(gap_samples, dtype=np.float32)
for f in wav_files:
audio, sr = sf.read(str(f), dtype="float32")
if audio.ndim > 1:
audio = audio.mean(axis=1)
if sr != sample_rate:
import subprocess as sp
tmp_out = output_path.parent / f"resample_{f.stem}.wav"
sp.run(
[
"ffmpeg",
"-y",
"-i",
str(f),
"-ar",
str(sample_rate),
str(tmp_out),
],
capture_output=True,
check=True,
)
audio, _ = sf.read(str(tmp_out), dtype="float32")
tmp_out.unlink(missing_ok=True)
segments.append(audio)
segments.append(gap)
if segments:
segments.pop()
output_path.parent.mkdir(parents=True, exist_ok=True)
# STREAM to disk. Building one np.concatenate() array for a 7.4-hour
# book at 44.1 kHz float32 needs ~4.7 GB, which this 16 GB machine
# could not satisfy alongside the TTS server's models: the array came
# back truncated and the m4b shipped ~41 minutes short. Writing each
# segment as it is produced keeps peak memory at one chapter.
# FLOAT, not the PCM_16 default: 16-bit truncates anything above
# +/-1.0, which was clipping the book to 0.00 dBFS before
# normalisation ever saw it.
with sf.SoundFile(
str(output_path),
mode="w",
samplerate=sample_rate,
channels=1,
subtype="FLOAT",
) as sink:
if not segments:
sink.write(np.zeros(1, dtype=np.float32))
else:
for segment in segments:
sink.write(np.asarray(segment, dtype=np.float32))
return output_path
def _normalize_to_spec(self, audio):
"""Scale to the RMS target and clip at the peak ceiling.
Mirrors :meth:`ARPackageExporter._normalize` so a listener hears the same
loudness whichever artifact they buy. RMS is measured on the content
before any padding, because every distributor defines RMS "across the
entire program (not including room tone)".
"""
import numpy as np
audio = np.asarray(audio, dtype=np.float32)
if audio.ndim > 1:
audio = audio.mean(axis=1)
rms = self._measure_array_rms(audio)
if rms is None:
return audio
gain_db = min(
max(NORM_TARGET_RMS_DB - rms, -NORM_MAX_GAIN_DB), NORM_MAX_GAIN_DB
)
scaled = audio * float(10 ** (gain_db / 20))
# Clip below the ceiling by the AAC overshoot margin, so the ENCODED
# peak lands at or under the limit. Trimming the whole signal after the
# fact instead cost ~4.5 dB and pushed RMS out of the distributor band.
ceiling = PEAK_LIMIT_LINEAR * float(10 ** (-AAC_OVERSHOOT_MARGIN_DB / 20))
np.clip(scaled, -ceiling, ceiling, out=scaled)
return scaled
def _measure_file_rms(self, path: Path) -> float | None:
"""RMS in dB of a whole file, measured by ffmpeg so file size is irrelevant."""
result = subprocess.run(
["ffmpeg", "-i", str(path), "-af", "volumedetect", "-f", "null", "-"],
capture_output=True,
text=True,
)
match = re.search(r"mean_volume:\s*([-\d.]+) dB", result.stderr or "")
return float(match.group(1)) if match else None
def _measure_array_rms(self, audio):
"""RMS in dB of a float array, computed directly (no temp file needed)."""
import numpy as np
if audio.size == 0:
return None
rms = float(np.sqrt(np.mean(np.square(audio, dtype=np.float64))))
if rms <= 0:
return None
return 20.0 * float(np.log10(rms))
def _measure_encoded_peak(self, path: Path) -> float | None:
"""Peak of the ENCODED m4b. The AAC encoder overshoots the float
ceiling, so only the delivered file reveals a breach - measuring the
source WAV would certify our intent rather than the artifact."""
result = subprocess.run(
["ffmpeg", "-i", str(path), "-af", "volumedetect", "-f", "null", "-"],
capture_output=True,
text=True,
)
if result.returncode != 0:
return None
match = re.search(r"max_volume:\s*([-\d.]+) dB", result.stderr or "")
return float(match.group(1)) if match else None
def _encode_aac_m4b(
self, input_wav: Path, output_m4b: Path, work_dir: Path | None = None
) -> float | None:
"""Encode to AAC 44.1 kHz mono, trimming gain until the file complies.
Returns the delivered file's peak in dB, or None if it could not be
measured. ``-ar`` and ``-ac`` are explicit: without them the encoder
kept the TTS-native 24 kHz, which fails the 44.1 kHz requirement at
both Author's Republic and Spotify.
"""
if work_dir is None:
work_dir = input_wav.parent
work_dir.mkdir(parents=True, exist_ok=True)
# Both the measurement and the decode stay inside ffmpeg. libsndfile
# stops reading at 2**30 - 1 frames, and a 7.4-hour book at 44.1 kHz is
# ~1.18e9, so ANY read through libsndfile silently dropped ~41 minutes
# before the encoder saw them - including a chunked read loop, because
# the cap is in the library's frame counter, not in the call.
rms = self._measure_file_rms(input_wav)
gain_db = 0.0
if rms is not None:
gain_db = min(
max(NORM_TARGET_RMS_DB - rms, -NORM_MAX_GAIN_DB), NORM_MAX_GAIN_DB
)
# alimiter looks ahead, so it holds the ceiling without the flat-topped
# distortion hard clipping produces - and it survives the AAC encoder
# better, which is why the margin is applied here rather than as a trim.
ceiling_db = PEAK_MAX_DB - AAC_OVERSHOOT_MARGIN_DB
limiter = float(10 ** (ceiling_db / 20))
peak = None
for attempt in range(AAC_ENFORCE_PASSES):
cmd = [
"ffmpeg",
"-y",
"-i",
str(input_wav),
"-af",
f"volume={gain_db:.3f}dB,alimiter=limit={limiter:.6f}:level=disabled",
"-c:a",
"aac",
"-b:a",
"128k",
"-ar",
str(TARGET_SAMPLE_RATE),
"-ac",
"1",
"-movflags",
"+faststart",
str(output_m4b),
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
logger.error("AAC encoding failed: %s", result.stderr)
raise RuntimeError(f"AAC encoding failed: {result.stderr}")
peak = self._measure_encoded_peak(output_m4b)
if peak is None or peak <= PEAK_MAX_DB:
break
trim_db = (PEAK_MAX_DB - peak) - AAC_TRIM_SAFETY_DB
logger.warning(
"m4b peak %.2f dB breaches %.1f dB (pass %d); re-encoding with %.2f dB trim",
peak,
PEAK_MAX_DB,
attempt + 1,
trim_db,
)
gain_db += trim_db
ceiling_db -= abs(trim_db)
limiter = float(10 ** (ceiling_db / 20))
return peak
def m4b_loudness_report(self, m4b_path: Path) -> dict[str, float | bool | None]:
"""Measure a delivered m4b against the shared spec, for verification."""
result = subprocess.run(
["ffmpeg", "-i", str(m4b_path), "-af", "volumedetect", "-f", "null", "-"],
capture_output=True,
text=True,
)
stderr = result.stderr or ""
rms = peak = None
m = re.search(r"mean_volume:\s*([-\d.]+) dB", stderr)
if m:
rms = float(m.group(1))
m = re.search(r"max_volume:\s*([-\d.]+) dB", stderr)
if m:
peak = float(m.group(1))
return {
"rms_db": rms,
"peak_db": peak,
"loudness_ok": rms is not None and RMS_LO_DB <= rms <= RMS_HI_DB,
"peak_ok": peak is not None and peak <= PEAK_MAX_DB,
}
def _apply_metadata(
self,
input_m4b: Path,
metadata_file: Path,
output_path: Path,
) -> None:
cmd = [
"ffmpeg",
"-y",
"-i",
str(input_m4b),
"-f",
"ffmetadata",
"-i",
str(metadata_file),
"-map",
"0",
"-map_metadata",
"1",
"-codec",
"copy",
str(output_path),
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
logger.warning(
"Metadata embedding failed, using untagged M4B: %s",
result.stderr,
)
shutil.copy2(str(input_m4b), str(output_path))
def _extract_chapter_titles(
self,
all_files: list[Path],
chapter_groups: dict[Path, list[Path]],
) -> list[str]:
titles = []
for parent in chapter_groups:
dir_name = parent.name
if dir_name.startswith("ch"):
try:
ch_num = int(dir_name[2:].split("_")[0])
titles.append(f"Chapter {ch_num}")
except ValueError:
titles.append(dir_name)
else:
titles.append(dir_name)
return titles
def _get_wav_durations(self, wav_files: list[Path]) -> list[float]:
import soundfile as sf
durations = []
for f in wav_files:
try:
info = sf.info(str(f))
durations.append(info.duration)
except Exception:
durations.append(0.0)
return durations
def export_mp3(
self,
chapter_files: list[Path],
output_dir: Path,
metadata: dict | None = None,
) -> list[Path]:
output_dir.mkdir(parents=True, exist_ok=True)
exported = []
for i, ch_file in enumerate(chapter_files, 1):
out = output_dir / f"chapter_{i:03d}.mp3"
cmd = [
"ffmpeg",
"-y",
"-i",
str(ch_file),
"-codec:a",
"libmp3lame",
"-b:a",
"192k",
str(out),
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
exported.append(out)
else:
logger.error("MP3 export failed for chapter %d: %s", i, result.stderr)
return exported
def export_wav(
self,
chapter_files: list[Path],
output_dir: Path,
) -> list[Path]:
output_dir.mkdir(parents=True, exist_ok=True)
exported = []
for i, ch_file in enumerate(chapter_files, 1):
out = output_dir / f"chapter_{i:03d}.wav"
shutil.copy2(str(ch_file), str(out))
exported.append(out)
return exported