From 5f0e2a97688c79f1435e311239074c67b5b7b826 Mon Sep 17 00:00:00 2001 From: Grok Bot Date: Sat, 17 Oct 2026 15:27:14 -0500 Subject: [PATCH] Add pipeline code --- _pipeline/.gitignore | 1 + _pipeline/Dockerfile | 15 +++ _pipeline/acx_check.py | 106 ++++++++++++++++ _pipeline/assemble.py | 138 +++++++++++++++++++++ _pipeline/audiomeasure.py | 51 ++++++++ _pipeline/bookio.py | 44 +++++++ _pipeline/cleanup.py | 200 +++++++++++++++++++++++++++++++ _pipeline/cli.py | 105 ++++++++++++++++ _pipeline/figures.py | 71 +++++++++++ _pipeline/ingest.py | 103 ++++++++++++++++ _pipeline/license.py | 40 +++++++ _pipeline/llm.py | 18 +++ _pipeline/pageparts.py | 65 ++++++++++ _pipeline/pdftext.py | 76 ++++++++++++ _pipeline/requirements.txt | 3 + _pipeline/script.py | 104 ++++++++++++++++ _pipeline/settings.py | 30 +++++ _pipeline/settings.yaml | 75 ++++++++++++ _pipeline/structure.py | 140 ++++++++++++++++++++++ _pipeline/tests/test_pipeline.py | 137 +++++++++++++++++++++ _pipeline/voice.py | 90 ++++++++++++++ 21 files changed, 1612 insertions(+) create mode 100644 _pipeline/.gitignore create mode 100644 _pipeline/Dockerfile create mode 100644 _pipeline/acx_check.py create mode 100644 _pipeline/assemble.py create mode 100644 _pipeline/audiomeasure.py create mode 100644 _pipeline/bookio.py create mode 100644 _pipeline/cleanup.py create mode 100644 _pipeline/cli.py create mode 100644 _pipeline/figures.py create mode 100644 _pipeline/ingest.py create mode 100644 _pipeline/license.py create mode 100644 _pipeline/llm.py create mode 100644 _pipeline/pageparts.py create mode 100644 _pipeline/pdftext.py create mode 100644 _pipeline/requirements.txt create mode 100644 _pipeline/script.py create mode 100644 _pipeline/settings.py create mode 100644 _pipeline/settings.yaml create mode 100644 _pipeline/structure.py create mode 100644 _pipeline/tests/test_pipeline.py create mode 100644 _pipeline/voice.py diff --git a/_pipeline/.gitignore b/_pipeline/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/_pipeline/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/_pipeline/Dockerfile b/_pipeline/Dockerfile new file mode 100644 index 0000000..c3d9e1e --- /dev/null +++ b/_pipeline/Dockerfile @@ -0,0 +1,15 @@ +# Pipeline container. Build from the repo root: docker build -f _pipeline/Dockerfile -t old-books-pipeline . +# Run: docker run --rm -v /path/to/old-books/books:/repo/books old-books-pipeline run "/repo/books/x.pdf" --chapters 1 +# Not built yet. Voice and speech-to-text are separate off-the-shelf containers reached through the URLs in settings.yaml. +FROM python:3.13-slim + +RUN apt-get update \ + && apt-get install -y --no-install-recommends ffmpeg poppler-utils tesseract-ocr tesseract-ocr-eng ghostscript \ + && rm -rf /var/lib/apt/lists/* + +COPY _pipeline/requirements.txt /tmp/requirements.txt +RUN pip install --no-cache-dir -r /tmp/requirements.txt && rm /tmp/requirements.txt + +COPY _pipeline /repo/_pipeline +WORKDIR /repo +ENTRYPOINT ["python3", "/repo/_pipeline/cli.py"] diff --git a/_pipeline/acx_check.py b/_pipeline/acx_check.py new file mode 100644 index 0000000..42c5363 --- /dev/null +++ b/_pipeline/acx_check.py @@ -0,0 +1,106 @@ +"""Deterministic ACX checker (R13). Measures each MP3 and prints a pass/fail table. Exit code 1 if anything fails. +Checks per file: MP3 CBR bitrate, sample rate, channels, RMS, peak, noise floor, room tone head and tail, duration. +Checks per set: all files mono or all stereo; each script has an MP3; chapter scripts start with a spoken header.""" +import glob +import json +import math +import os +import subprocess + +import audiomeasure as am + + +def packet_sizes(path): + out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "packet=size", "-of", "default=nw=1:nk=1", path], + capture_output=True, text=True, check=True).stdout.split() + return [int(x) for x in out] + + +def measure(cfg, path): + """All numbers for one file, as a dict.""" + a = cfg["acx"] + info = am.probe(path) + w = am.windows(path) + rms = [r for r, _ in w] + win_s = 0.05 + loud = [i for i, r in enumerate(rms) if r > a["speech_window_db"]] + sizes = packet_sizes(path) + blocks = [am.db_mean_power(rms[i:i + 10]) for i in range(0, len(rms) - 9, 10)] + return { + "codec": info["codec_name"], + "sample_rate": int(info["sample_rate"]), + "channels": int(info["channels"]), + "bitrate_kbps": round(int(info["bit_rate"]) / 1000), + "cbr": max(sizes) - min(sizes) <= 1, + "duration_s": float(info["duration"]), + "rms_db": am.db_mean_power(rms), + "peak_db": max(p for _, p in w), + "noise_floor_db": min(blocks) if blocks else float("nan"), + "head_s": loud[0] * win_s if loud else float("nan"), + "tail_s": (len(rms) - 1 - loud[-1]) * win_s if loud else float("nan"), + } + + +def judge(cfg, name, m): + """List of (check name, value text, passed).""" + a = cfg["acx"] + retail = os.path.basename(name).startswith("retail-sample") + dur_ok = a["retail_min_s"] <= m["duration_s"] <= a["retail_max_s"] if retail else m["duration_s"] < a["max_minutes"] * 60 + ok = lambda v, lo, hi: (not math.isnan(v)) and lo <= v <= hi + return [ + ("format", "%s %s %dk %d Hz" % (m["codec"], "CBR" if m["cbr"] else "VBR", m["bitrate_kbps"], m["sample_rate"]), + m["codec"] == "mp3" and m["cbr"] and m["bitrate_kbps"] >= a["min_bitrate_kbps"] and m["sample_rate"] == a["sample_rate"]), + ("channels", str(m["channels"]), m["channels"] in (1, 2)), + ("rms dB", "%.1f" % m["rms_db"], ok(m["rms_db"], a["rms_min_db"], a["rms_max_db"])), + ("peak dB", "%.1f" % m["peak_db"], m["peak_db"] <= a["peak_max_db"]), + ("noise floor dB", "%.1f" % m["noise_floor_db"], m["noise_floor_db"] <= a["noise_floor_max_db"]), + ("head s", "%.2f" % m["head_s"], ok(m["head_s"], a["head_min_s"], a["head_max_s"])), + ("tail s", "%.2f" % m["tail_s"], ok(m["tail_s"], a["tail_min_s"], a["tail_max_s"])), + ("length", "%.1f s" % m["duration_s"], dur_ok), + ] + + +def set_checks(cfg, folder, measured): + """Checks across files; folder is an audio/ folder next to script/ (or None).""" + rows = [] + chans = {m["channels"] for m in measured.values()} + rows.append(("all files mono or all stereo", ",".join(map(str, sorted(chans))), len(chans) <= 1)) + scripts = os.path.join(os.path.dirname(folder), "script") if folder else None + if scripts and os.path.isdir(scripts): + for sp in sorted(glob.glob(os.path.join(scripts, "*.json"))): + stem = os.path.splitext(os.path.basename(sp))[0] + rows.append(("mp3 exists for script " + stem, "yes" if stem in {os.path.splitext(os.path.basename(n))[0] for n in measured} else "no", + stem in {os.path.splitext(os.path.basename(n))[0] for n in measured})) + if stem.startswith("chapter-"): + with open(sp, encoding="utf-8") as f: + first = json.load(f)[0]["kind"] + rows.append(("spoken section header in " + stem, first, first == "header")) + return rows + + +def run(cfg, paths): + files, folder = [], None + for p in paths: + if os.path.isdir(p): + folder = os.path.abspath(p) + files += sorted(glob.glob(os.path.join(p, "*.mp3"))) + else: + files.append(p) + if not files: + print("no MP3 files found") + return 1 + measured, failed = {}, False + for f in files: + m = measure(cfg, f) + measured[f] = m + print("\n%s" % f) + print(" %-16s %-26s %s" % ("check", "value", "result")) + for name, value, passed in judge(cfg, f, m): + failed |= not passed + print(" %-16s %-26s %s" % (name, value, "PASS" if passed else "FAIL")) + print("\nSet checks") + for name, value, passed in set_checks(cfg, folder, measured): + failed |= not passed + print(" %-40s %-10s %s" % (name, value, "PASS" if passed else "FAIL")) + print("\nOVERALL: %s" % ("FAIL" if failed else "PASS")) + return 1 if failed else 0 diff --git a/_pipeline/assemble.py b/_pipeline/assemble.py new file mode 100644 index 0000000..7aadf6e --- /dev/null +++ b/_pipeline/assemble.py @@ -0,0 +1,138 @@ +"""Assembly with ffmpeg: wav pieces + chimes + pauses -> one MP3 per script file, to ACX levels. +1. Speech track: head, pieces, chimes, pauses (digital silence), tail. +2. Room tone: quiet noise (fixed seed) as long as the speech track, mixed underneath everywhere, so there is no digital silence. +3. Gain on the speech track so the finished mix lands on the RMS target; a limiter holds the peak ceiling. +4. MP3, CBR, 44.1 kHz, mono.""" +import json +import math +import os +import shutil +import tempfile + +import audiomeasure as am +import settings +import voice + +FMT = ["-ar", "{sr}", "-ac", "{ch}", "-c:a", "pcm_f32le"] + + +def fmt(a): + return [x.format(sr=a["sample_rate"], ch=a["channels"]) for x in FMT] + + +def make_silence(a, seconds, path): + am.ffmpeg(["-f", "lavfi", "-i", "anullsrc=r=%d:cl=%s" % (a["sample_rate"], "mono" if a["channels"] == 1 else "stereo"), + "-t", "%.3f" % seconds] + fmt(a) + [path]) + + +def make_room_tone(a, seconds, path): + """Very quiet noise (ACX wants room tone, not digital silence). Fixed seed, so output is repeatable.""" + amp = 10 ** (a["room_tone_db"] / 20) * math.sqrt(3) # uniform white noise: RMS = amplitude / sqrt(3) + am.ffmpeg(["-f", "lavfi", "-i", "anoisesrc=d=%.3f:c=white:r=%d:a=%.6f:seed=1" % (seconds, a["sample_rate"], amp)] + fmt(a) + [path]) + + +def make_chime(a, level_db, path): + """A soft sine tone with a smooth rise and fall (sin^2 envelope), peak at level_db.""" + amp = 10 ** (level_db / 20) + dur = a["chime_s"] + expr = "%.6f*sin(2*PI*%g*t)*pow(sin(PI*t/%g),2)" % (amp, a["chime_hz"], dur) + am.ffmpeg(["-f", "lavfi", "-i", "aevalsrc='%s':s=%d:d=%g" % (expr, a["sample_rate"], dur)] + fmt(a) + [path]) + + +def concat(paths, out, tmp): + lst = os.path.join(tmp, "list_%s.txt" % os.path.basename(out)) + with open(lst, "w") as f: + for p in paths: + f.write("file '%s'\n" % p.replace("'", "'\\''")) + am.ffmpeg(["-f", "concat", "-safe", "0", "-i", lst, "-c", "copy", out]) + + +def build_speech(a, items, chime_db, tmp, tag): + """Speech track. chime_db None leaves the cue slots silent (used to find the gain before the chime is made).""" + cache = {} + + def silence(seconds): + key = "sil%.3f" % seconds + if key not in cache: + cache[key] = os.path.join(tmp, key + ".wav") + make_silence(a, seconds, cache[key]) + return cache[key] + + parts = [silence(a["head_room_tone_s"])] + for n, it in enumerate(items): + if "cue" in it: + if chime_db is None: + parts.append(silence(a["chime_s"])) + else: + if "chime" not in cache: + cache["chime"] = os.path.join(tmp, "chime_%s.wav" % tag) + make_chime(a, chime_db, cache["chime"]) + parts.append(cache["chime"]) + else: + p = os.path.join(tmp, "p%05d.wav" % n) + if not os.path.exists(p): + am.ffmpeg(["-i", it["wav"]] + fmt(a) + [p]) + parts.append(p) + if n < len(items) - 1: + parts.append(silence(it["gap_after_s"])) + parts.append(silence(a["tail_room_tone_s"])) + out = os.path.join(tmp, "speech_%s.wav" % tag) + concat(parts, out, tmp) + return out + + +def mix(a, speech, gain, noise, out): + limit = 10 ** (a["peak_ceiling_db"] / 20) + am.ffmpeg(["-i", speech, "-i", noise, "-filter_complex", + "[0:a]volume=%.3fdB,alimiter=limit=%.5f:level=0[s];[s][1:a]amix=inputs=2:normalize=0:duration=first" % (gain, limit)] + + fmt(a) + [out]) + + +def set_level(a, speech, tmp): + """Find the gain that puts the finished mix on the RMS target. Returns (mix path, gain dB).""" + noise = os.path.join(tmp, "noise.wav") + make_room_tone(a, float(am.probe(speech)["duration"]), noise) + rms, _ = am.overall(speech) + gain = a["target_rms_db"] - rms + for i in range(3): + out = os.path.join(tmp, "mix_%d.wav" % i) + mix(a, speech, gain, noise, out) + got, _ = am.overall(out) + if abs(got - a["target_rms_db"]) <= 0.3: + break + gain += a["target_rms_db"] - got + return out, gain, noise + + +def assemble_items(cfg, items, out_mp3): + """items as written by voice.py. Writes out_mp3. Returns the gain applied to speech in dB.""" + a = dict(cfg["audio"]) + work = settings.resolve(cfg, "work_dir") + os.makedirs(work, exist_ok=True) + tmp = tempfile.mkdtemp(prefix="asm_", dir=work) + try: + speech = build_speech(a, items, None, tmp, "measure") + final, gain, noise = set_level(a, speech, tmp) + if any("cue" in it for it in items): + # build again with the chime in; it is made so it lands at chime_db after the gain + speech = build_speech(a, items, a["chime_db"] - gain, tmp, "final") + final = os.path.join(tmp, "final.wav") + mix(a, speech, gain, noise, final) + os.makedirs(os.path.dirname(out_mp3), exist_ok=True) + am.ffmpeg(["-i", final, "-c:a", "libmp3lame", "-b:a", a["mp3_bitrate"], "-ar", str(a["sample_rate"]), "-ac", str(a["channels"]), out_mp3]) + return gain + finally: + shutil.rmtree(tmp, ignore_errors=True) + + +def run(cfg, book_dir, only=None): + import bookio + book = bookio.read_book(book_dir) + out_dir = settings.audio_dir(cfg, book_dir) + for name in book["audio_files"]: + if only and name != only: + continue + with open(os.path.join(voice.work_dir_for(cfg, book_dir, name), "manifest.json"), encoding="utf-8") as f: + items = json.load(f) + gain = assemble_items(cfg, items, os.path.join(out_dir, name + ".mp3")) + print("%s.mp3 written (speech gain %.1f dB)" % (name, gain)) diff --git a/_pipeline/audiomeasure.py b/_pipeline/audiomeasure.py new file mode 100644 index 0000000..2744a8f --- /dev/null +++ b/_pipeline/audiomeasure.py @@ -0,0 +1,51 @@ +"""Measure audio with ffmpeg: RMS and peak per 50 ms window. The checker and the assembler both use this.""" +import math +import os +import re +import subprocess +import tempfile + + +def ffmpeg(args): + r = subprocess.run(["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y"] + args, capture_output=True, text=True) + if r.returncode != 0: + raise RuntimeError("ffmpeg failed: " + r.stderr.strip()[-500:]) + + +def probe(path): + out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", + "stream=codec_name,sample_rate,channels,bit_rate:format=duration", "-of", "default=nw=1", path], + capture_output=True, text=True, check=True).stdout + return dict(line.split("=", 1) for line in out.strip().splitlines()) + + +def windows(path): + """[(rms_db, peak_db)] for consecutive 50 ms windows of the decoded file.""" + sr = int(probe(path)["sample_rate"]) + n = sr // 20 + with tempfile.TemporaryDirectory() as tmp: + rms_file, peak_file = os.path.join(tmp, "rms.txt"), os.path.join(tmp, "peak.txt") + af = ("asetnsamples=n=%d:p=0,astats=metadata=1:reset=1," + "ametadata=print:key=lavfi.astats.Overall.RMS_level:file=%s," + "ametadata=print:key=lavfi.astats.Overall.Peak_level:file=%s" % (n, rms_file, peak_file)) + r = subprocess.run(["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-i", path, "-af", af, "-f", "null", "-"], + capture_output=True, text=True) + if r.returncode != 0: + raise RuntimeError("ffmpeg failed: " + r.stderr.strip()[-500:]) + read = lambda p, key: [float(v) for v in re.findall(key + r"=(-?[\d.]+|-inf|inf)", open(p).read())] + rms, peak = read(rms_file, "RMS_level"), read(peak_file, "Peak_level") + return list(zip(rms, peak)) + + +def db_mean_power(values): + """Overall level in dB of equal-length windows given as dB RMS.""" + if not values: + return float("-inf") + p = sum(10 ** (v / 10) for v in values) / len(values) + return 10 * math.log10(p) if p > 0 else float("-inf") + + +def overall(path): + """(rms_db, peak_db) of the whole file.""" + w = windows(path) + return db_mean_power([r for r, _ in w]), max(p for _, p in w) diff --git a/_pipeline/bookio.py b/_pipeline/bookio.py new file mode 100644 index 0000000..39a64f9 --- /dev/null +++ b/_pipeline/bookio.py @@ -0,0 +1,44 @@ +"""Small helpers shared by the stages: book.json, warnings.log, slugs.""" +import json +import os +import re + +WARN_NAME = "warnings.log" + + +def slug(text): + text = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") + return text + + +def read_book(book_dir): + with open(os.path.join(book_dir, "book.json"), encoding="utf-8") as f: + return json.load(f) + + +def write_book(book_dir, book): + with open(os.path.join(book_dir, "book.json"), "w", encoding="utf-8") as f: + json.dump(book, f, indent=2, ensure_ascii=False) + f.write("\n") + + +def write_warnings(book_dir, stage, entries): + """Replace this stage's lines in warnings.log. entry = (kind, where, message). + Line format: stage | kind | where | message""" + path = os.path.join(book_dir, WARN_NAME) + keep = [] + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + keep = [ln for ln in f.read().splitlines() if ln and not ln.startswith(stage + " |")] + lines = keep + ["%s | %s | %s | %s" % (stage, k, w, m) for k, w, m in entries] + with open(path, "w", encoding="utf-8") as f: + f.write("\n".join(lines) + ("\n" if lines else "")) + + +def title_case(text): + small = {"a", "an", "and", "as", "at", "but", "by", "for", "in", "of", "on", "or", "the", "to"} + words = text.lower().split() + out = [] + for i, w in enumerate(words): + out.append(w if (w in small and 0 < i < len(words) - 1) else w[:1].upper() + w[1:]) + return " ".join(out) diff --git a/_pipeline/cleanup.py b/_pipeline/cleanup.py new file mode 100644 index 0000000..c26a5d0 --- /dev/null +++ b/_pipeline/cleanup.py @@ -0,0 +1,200 @@ +"""Stage 2, clean-up: turn the text layer of a chapter into clean paragraphs. +Every rule is deterministic. Every change and every word the dictionary does not know goes to warnings.log. +Author words are never rewritten, only layout artifacts are removed or repaired.""" +import os +import re + +from spellchecker import SpellChecker + +import bookio +import pageparts +import pdftext +import structure + +WORD_RE = re.compile(r"[A-Za-z]+(?:'[A-Za-z]+)?") +_dict = SpellChecker() + + +def known(word): + return bool(_dict.known([word.lower()])) + + +def vocabulary(units): + """Words and hyphenated words that occur inline (not split by a line break) anywhere in the book.""" + inline = set() + for u in units: + for ln in u["text"].split("\n"): + toks = re.findall(r"[A-Za-z]+(?:-[A-Za-z]+)*", ln) + if ln.rstrip().endswith("-") and toks: + toks = toks[:-1] + inline.update(t.lower() for t in toks) + return inline + + +def parts_known(word, inline): + return all(known(p) or p.lower() in inline for p in word.split("-") if p) + + +def join_hyphen(left, frag, inline): + """Decide how to join 'left-' + 'frag'. Returns (joined_word, reason, resolved).""" + joined, hyph = left + frag, left + "-" + frag + if hyph.lower() in inline: + return hyph, "hyphenated form occurs elsewhere in the book", True + if parts_known(joined, inline) and (known(joined.split("-")[-1]) or joined.lower() in inline): + return joined, "joined word is a known word", True + if known(left.split("-")[-1]) and known(frag): + return hyph, "both fragments are words and the joined form is unknown, hyphen kept", True + return joined, "neither form is known, joined by default", False + + +def split_dropcap(lines, i): + """A lone capital letter followed by a line that starts with the rest of the first word.""" + if re.fullmatch(r"[A-Z]", lines[i]["text"].strip()) and i + 1 < len(lines): + nxt = lines[i + 1]["text"].strip() + if re.match(r"[A-Z]{2,}\b", nxt) or re.match(r"[A-Z]{2,}$", nxt): + return True + return False + + +def smallcaps_lead(text): + """'ALL MY LIFE I have' -> 'All my life I have' (leading run of capitals after a drop cap).""" + toks = text.split(" ") + n = 0 + while n < len(toks) and toks[n].strip(".,;:!?\"'").isupper(): + n += 1 + out = [] + for k, t in enumerate(toks): + if k < n and len(t.strip(".,;:!?\"'")) > 1: + t = t.lower() + if k == 0: + t = t[:1].upper() + t[1:] + out.append(t) + return " ".join(out) + + +def unit_lines(unit, title, heading, first_unit, notes): + """Body lines of one unit as dicts {text, indent, where}; removed parts are noted.""" + raw = [l.rstrip() for l in unit["text"].split("\n")] + where = "pdf p.%d%s" % (unit["pdf_page"], unit["half"] or "") + drop = set() + z = pageparts.header_zone(raw, title) + if z: + drop.update(range(z[0], z[1])) + what = "page number %d" % z[2] if z[2] is not None else "no page number" + notes.append(("header-removed", where, "running header '%s' (%s)" % (" ".join(l.strip() for l in raw[z[0]:z[1]]), what))) + f = pageparts.footer_number(raw) + if f: + drop.add(f[0]) + notes.append(("page-number-removed", where, "footer page number %d" % f[1])) + if first_unit and heading: + drop.update(range(0, heading["title_end"])) + notes.append(("heading-removed", where, "chapter heading 'CHAPTER %d / %s' (spoken header is added in the script)" % (heading["number"], heading["title"]))) + out = [] + for i, ln in enumerate(raw): + if i in drop or not ln.strip(): + continue + if pageparts.CAPTION_RE.match(ln): + notes.append(("caption-removed", where, "figure caption '%s' (listed in figures/skipped_figures.md)" % ln.strip())) + continue + out.append({"text": ln.strip(), "indent": len(ln) - len(ln.lstrip()), "where": where}) + return out + + +def paragraphs_from(lines, inline, notes): + """Group lines into paragraphs by first-line indent, repair drop caps and hyphenation.""" + base = min(l["indent"] for l in lines) + i, paras, cur, cap_indent = 0, [], None, None + while i < len(lines): + ln = dict(lines[i]) + is_cap = split_dropcap(lines, i) + if is_cap: + letter, frag = ln["text"], lines[i + 1] + first = re.match(r"[A-Z]+", frag["text"]).group(0) + if known(letter + first) or not (letter in "AI" and known(first)): + merged = letter + frag["text"] + how = "joined" + else: + merged = letter + " " + frag["text"] + how = "kept as its own word" + notes.append(("dropcap", ln["where"], "drop cap '%s' + '%s': %s" % (letter, frag["text"][:12], how))) + ln = {"text": merged, "indent": base + 1, "where": ln["where"], "smallcaps": True} + cap_indent = frag["indent"] + i += 2 + cur = [ln] + paras.append(cur) + continue + starts = ln["indent"] - base >= 1 and not (cap_indent is not None and ln["indent"] == cap_indent) + if ln["indent"] != cap_indent: + cap_indent = None + if starts or cur is None: + cur = [ln] + paras.append(cur) + else: + cur.append(ln) + i += 1 + return paras + + +def join_lines(para, inline, notes): + text = para[0]["text"] + for ln in para[1:]: + nxt = ln["text"] + m = re.search(r"([A-Za-z]+(?:-[A-Za-z]+)*)-$", text) + f = re.match(r"([a-z]+)(.*)$", nxt) + if m and f: + joined, reason, ok = join_hyphen(m.group(1), f.group(1), inline) + notes.append(("hyphen-joined" if ok else "hyphen-unresolved", ln["where"], + "'%s-' + '%s' -> '%s' (%s)" % (m.group(1), f.group(1), joined, reason))) + text = text[:m.start()] + joined + f.group(2) + else: + text = text + " " + nxt + return re.sub(r"\s+", " ", text).strip() + + +def fix_quotes(text, where, notes): + """A paragraph with an unclosed double quote that ends in punctuation + ' : the ' is a misread closing quote.""" + if text.count('"') % 2 == 1 and re.search(r"[.?!]'$", text): + notes.append(("quote-fixed", where, "closing quote misread as apostrophe at '...%s'" % text[-12:])) + return text[:-1] + '"' + return text + + +def odd_words(text, where, notes): + for w in WORD_RE.findall(text): + core = re.sub(r"'s$", "", w) + if len(core) > 1 and not known(core): + notes.append(("odd-word", where, "'%s' is not in the dictionary%s" % (w, " (capitalised: probably a name)" if w[0].isupper() else ""))) + + +def run(book_dir, book, pdf, chapter_numbers): + """Writes text/chapter-NN.txt for the chosen chapters. Returns warnings.""" + units = pdftext.units(pdf) + heads = {h["number"]: h for h in structure.find_headings(units)} + inline = vocabulary(units) + os.makedirs(os.path.join(book_dir, "text"), exist_ok=True) + notes = [] + for ch in book["chapters"]: + if ch["number"] not in chapter_numbers: + continue + lines = [] + for ui in range(ch["first_unit"], ch["last_unit"] + 1): + lines += unit_lines(units[ui], book["title"], heads[ch["number"]], ui == ch["first_unit"], notes) + paras = paragraphs_from(lines, inline, notes) + texts = [] + for n, p in enumerate(paras, start=1): + t = join_lines(p, inline, notes) + if p[0].get("smallcaps"): + before, t = t, smallcaps_lead(t) + if before != t: + notes.append(("smallcaps", p[0]["where"], "'%s...' -> '%s...'" % (before[:14], t[:14]))) + where = "ch%d para %d" % (ch["number"], n) + t = fix_quotes(t, where, notes) + odd_words(t, where, notes) + texts.append(t) + name = "chapter-%02d.txt" % ch["number"] + with open(os.path.join(book_dir, "text", name), "w", encoding="utf-8") as f: + f.write("\n\n".join(texts) + "\n") + ch["text_file"] = "text/" + name + ch["words"] = sum(len(t.split()) for t in texts) + ch["paragraphs"] = len(texts) + return notes diff --git a/_pipeline/cli.py b/_pipeline/cli.py new file mode 100644 index 0000000..bcb51b6 --- /dev/null +++ b/_pipeline/cli.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""One command line for the whole pipeline. + + python3 _pipeline/cli.py run --chapters 1 ingest, structure, cleanup, figures, script, license + python3 _pipeline/cli.py ingest|structure|cleanup|figures|script|license ... + python3 _pipeline/cli.py voice [--only NAME] script segments -> wav pieces (needs the voice endpoint) + python3 _pipeline/cli.py assemble wav pieces -> MP3 per script file in audio/ + python3 _pipeline/cli.py check ... ACX checker, pass/fail table +""" +import argparse +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import bookio +import cleanup +import figures +import ingest +import license as license_stage +import script +import settings +import structure + + +def chapters_arg(text): + return [int(x) for x in text.split(",")] + + +def stage(name, fn, book_dir): + warnings = fn() + bookio.write_warnings(book_dir, name, warnings) + print("%-10s done, %d warnings" % (name, len(warnings))) + + +def run_stage(name, cfg, book_dir, chapters=None): + book = bookio.read_book(book_dir) + pdf = ingest.work_pdf(cfg, book_dir, book) + if name == "structure": + stage(name, lambda: structure.run(book_dir, book, pdf), book_dir) + for c in book["chapters"]: + print(" chapter %d %s (pdf pp. %d-%d, book pp. %s-%s)" % ( + c["number"], c["title"], c["pdf_pages"][0], c["pdf_pages"][1], c["book_pages"][0], c["book_pages"][1])) + elif name == "cleanup": + stage(name, lambda: cleanup.run(book_dir, book, pdf, chapters), book_dir) + elif name == "figures": + stage(name, lambda: figures.run(book_dir, book, pdf, chapters), book_dir) + elif name == "script": + stage(name, lambda: script.run(cfg, book_dir, book, chapters), book_dir) + elif name == "license": + stage(name, lambda: license_stage.run(cfg, book_dir, book, pdf) or [], book_dir) + bookio.write_book(book_dir, book) + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--settings", help="path to settings.yaml (default: _pipeline/settings.yaml)") + sub = ap.add_subparsers(dest="cmd", required=True) + p = sub.add_parser("run") + p.add_argument("pdf") + p.add_argument("--chapters", required=True, type=chapters_arg, help="chapter numbers, e.g. 1 or 1,2") + p.add_argument("--title") + p.add_argument("--author") + p = sub.add_parser("ingest") + p.add_argument("pdf") + p.add_argument("--title") + p.add_argument("--author") + sub.add_parser("structure").add_argument("book_dir") + for name in ("cleanup", "figures", "script"): + p = sub.add_parser(name) + p.add_argument("book_dir") + p.add_argument("--chapters", required=True, type=chapters_arg) + sub.add_parser("license").add_argument("book_dir") + p = sub.add_parser("voice") + p.add_argument("book_dir") + p.add_argument("--only", help="one script name, e.g. chapter-01") + p = sub.add_parser("assemble") + p.add_argument("book_dir") + p.add_argument("--only") + p = sub.add_parser("check") + p.add_argument("paths", nargs="+") + a = ap.parse_args(argv) + cfg = settings.load(a.settings) + + if a.cmd in ("run", "ingest"): + book_dir = ingest.run(cfg, os.path.abspath(a.pdf), a.title, a.author) + print("ingest done ->", book_dir) + if a.cmd == "run": + for name in ("structure", "cleanup", "figures", "script", "license"): + run_stage(name, cfg, book_dir, a.chapters) + elif a.cmd in ("structure", "cleanup", "figures", "script", "license"): + run_stage(a.cmd, cfg, os.path.abspath(a.book_dir), getattr(a, "chapters", None)) + elif a.cmd == "voice": + import voice + voice.run(cfg, os.path.abspath(a.book_dir), a.only) + elif a.cmd == "assemble": + import assemble + assemble.run(cfg, os.path.abspath(a.book_dir), a.only) + elif a.cmd == "check": + import acx_check + sys.exit(acx_check.run(cfg, a.paths)) + + +if __name__ == "__main__": + main() diff --git a/_pipeline/figures.py b/_pipeline/figures.py new file mode 100644 index 0000000..0fcea94 --- /dev/null +++ b/_pipeline/figures.py @@ -0,0 +1,71 @@ +"""Figures: find every image in the chapters being processed and list it in figures/skipped_figures.md. +Figures support the concept only (R7). Nothing is spoken unless a spoken sentence is approved, and none is yet, +so every figure is listed as skipped.""" +import os +import re + +import pageparts +import pdftext +import structure + + +def toc_pictures(units, before_unit): + """Picture list in the table of contents: [(caption, book_page)].""" + lines = [] + for u in units[:before_unit]: + lines += u["text"].split("\n") + out = [] + for ln in lines: + m = re.match(r"^\s*([A-Z][A-Z .,'\-—]+?)\s*\.{3,}\s*(\d+)(?:\s*-\s*\d+)?\s*$", ln) + if m: + out.append((m.group(1).strip(), int(m.group(2)))) + return out + + +def run(book_dir, book, pdf, chapter_numbers): + """Writes figures/skipped_figures.md. Returns warnings.""" + units = pdftext.units(pdf) + heads = structure.find_headings(units) + pages, _ = structure.book_pages(units, book["title"]) + toc = toc_pictures(units, heads[0]["unit"]) + imgs = pdftext.images(pdf) + rows, warnings = [], [] + for ch in book["chapters"]: + if ch["number"] not in chapter_numbers: + continue + lo, hi = ch["pdf_pages"] + for im in imgs: + if not lo <= im["pdf_page"] <= hi: + continue + ui = next(i for i, u in enumerate(units) if u["pdf_page"] == im["pdf_page"] and u["half"] == im["half"]) + caption, source = None, None + for u in units: + if u["pdf_page"] == im["pdf_page"]: + for ln in u["text"].split("\n"): + m = pageparts.CAPTION_RE.match(ln) + if m: + caption, source = m.group(2).title(), "caption printed on the page (%s)" % ln.strip() + if caption is None: + hit = [c for c, p in toc if p == pages[ui]] + if hit: + caption, source = hit[0].title(), "table of contents picture list, page %d" % pages[ui] + if caption is None: + warnings.append(("figure-no-caption", "pdf p.%d%s" % (im["pdf_page"], im["half"] or ""), "no caption found")) + rows.append((ch["number"], im, pages[ui], caption, source)) + os.makedirs(os.path.join(book_dir, "figures"), exist_ok=True) + out = ["# Skipped figures", "", + "Figures support the concept only, and nothing in a figure is read aloud unless a spoken sentence is approved.", + "No spoken figure sentence is approved for this book yet, so every figure below is skipped.", "", + "Scope: chapters %s." % ", ".join(str(n) for n in sorted(chapter_numbers)), ""] + for n, im, bp, caption, source in rows: + out += ["- Chapter %d, pdf page %d (%s half), book page %s: image %d x %d pt on the page." % ( + n, im["pdf_page"], {"L": "left", "R": "right", None: "whole"}[im["half"]], bp if bp and bp >= 1 else "unnumbered", + im["width"], im["height"]), + " Caption: %s." % (caption or "none found"), + " Caption source: %s." % (source or "none"), + " Reason skipped: no approved spoken sentence."] + if not rows: + out.append("No figures found in these chapters.") + with open(os.path.join(book_dir, "figures", "skipped_figures.md"), "w", encoding="utf-8") as f: + f.write("\n".join(out) + "\n") + return warnings diff --git a/_pipeline/ingest.py b/_pipeline/ingest.py new file mode 100644 index 0000000..aafee05 --- /dev/null +++ b/_pipeline/ingest.py @@ -0,0 +1,103 @@ +"""Stage 1, ingest: copy the original PDF into its book folder (never modify it), record its sha256, +find pages without a text layer and OCR only if there are any.""" +import hashlib +import os +import re +import shutil +import subprocess + +import bookio +import pdftext +import settings + + +def sha256(path): + h = hashlib.sha256() + with open(path, "rb") as f: + for block in iter(lambda: f.read(1 << 20), b""): + h.update(block) + return h.hexdigest() + + +def find_title(pdf): + out = subprocess.run(["pdfinfo", pdf], capture_output=True, text=True).stdout + m = re.search(r"^Title:\s+(.+)$", out, re.M) + return bookio.title_case(m.group(1).strip()) if m else "" + + +def find_author(unit_texts): + """The author is the line 'By ' on the first pages.""" + for text in unit_texts: + m = re.search(r"^\s*By\s+([A-Z][A-Za-z.\- ]+?)\s*$", text, re.M) + if m: + return m.group(1).strip() + return "" + + +def author_slug(author): + """'Glenn Clark' -> 'clark-glenn' (surname first, last word taken as surname).""" + parts = author.split() + return bookio.slug(" ".join([parts[-1]] + parts[:-1])) if parts else "unknown" + + +def run(cfg, pdf, title=None, author=None): + """Returns the book folder path.""" + pdf_units = pdftext.units(pdf) + author = author or find_author([u["text"] for u in pdf_units[:6]]) + title = title or find_title(pdf) + if not author or not title: + raise SystemExit("could not find title/author in the PDF; pass --title and --author") + book_dir = os.path.join(settings.resolve(cfg, "books_dir"), author_slug(author), bookio.slug(title)) + os.makedirs(book_dir, exist_ok=True) + copy = os.path.join(book_dir, os.path.basename(pdf)) + if os.path.exists(copy) and sha256(copy) != sha256(pdf): + raise SystemExit("a different file with this name is already in the book folder: " + copy) + if not os.path.exists(copy): + shutil.copyfile(pdf, copy) + book = {} + if os.path.exists(os.path.join(book_dir, "book.json")): + book = bookio.read_book(book_dir) + counts = pdftext.chars_per_page(copy) + book.update({ + "title": title, + "author": author, + "source_pdf": os.path.basename(pdf), + "sha256": sha256(copy), + "pdf_pages": len(counts), + "pages_without_text_layer": [i + 1 for i, c in enumerate(counts) if c < cfg["ocr"]["min_chars_per_page"]], + }) + book.setdefault("coauthors", []) + book.setdefault("publisher", "") + book.setdefault("edition", "") + book.setdefault("series", "") + book.setdefault("summaries", {}) + bookio.write_book(book_dir, book) + warnings = ocr_if_needed(cfg, book_dir, book) + bookio.write_warnings(book_dir, "ingest", warnings) + return book_dir + + +def work_pdf(cfg, book_dir, book): + """The PDF text is read from: the OCR copy if one exists, else the untouched original.""" + ocr_pdf = os.path.join(settings.resolve(cfg, "work_dir"), os.path.basename(book_dir), "ocr.pdf") + return ocr_pdf if os.path.exists(ocr_pdf) else os.path.join(book_dir, book["source_pdf"]) + + +def ocr_if_needed(cfg, book_dir, book): + """OCR a scratch copy (never the original) when pages lack a text layer. Returns warnings.""" + missing = book["pages_without_text_layer"] + if not missing: + return [] + where = "pdf pages " + ",".join(map(str, missing)) + if not cfg["ocr"]["enabled"]: + return [("no-text-layer", where, "no text layer and OCR is disabled in settings")] + if shutil.which("ocrmypdf") is None: + return [("no-text-layer", where, "no text layer and ocrmypdf is not installed on this host; those pages have no text")] + out_dir = os.path.join(settings.resolve(cfg, "work_dir"), os.path.basename(book_dir)) + os.makedirs(out_dir, exist_ok=True) + out = os.path.join(out_dir, "ocr.pdf") + src = os.path.join(book_dir, book["source_pdf"]) + r = subprocess.run(["ocrmypdf", "--skip-text", "-l", cfg["ocr"]["language"], src, out], capture_output=True, text=True) + if r.returncode != 0: + return [("ocr-failed", where, "ocrmypdf exit %d: %s" % (r.returncode, r.stderr.strip()[-300:]))] + return [("ocr", where, "OCR applied to a scratch copy; original untouched")] diff --git a/_pipeline/license.py b/_pipeline/license.py new file mode 100644 index 0000000..39d0820 --- /dev/null +++ b/_pipeline/license.py @@ -0,0 +1,40 @@ +"""License status by code. Published 1930 or earlier = US public domain as of 2026. Anything else is unknown.""" +import datetime +import os +import re + +import pdftext + +PATTERNS = [r"first\s+published\s+in\s+(\d{4})", r"(?:©|copyright|\(c\))\s*(\d{4})"] + + +def find_year(units): + """Earliest year the book's own front pages give as first publication or copyright: (year, where, text) or None.""" + for pat in PATTERNS: + for u in units[:8]: + m = re.search(pat, u["text"], re.I) + if m: + return int(m.group(1)), "pdf p.%d" % u["pdf_page"], m.group(0) + return None + + +def run(cfg, book_dir, book, pdf): + found = find_year(pdftext.units(pdf)) + limit = cfg["license"]["public_domain_max_year"] + if found is None: + status, reason, year = "unknown", "no publication year found in the book's own text", None + else: + year, where, quote = found + if year <= limit: + status = "public domain" + reason = "published %d (book text, %s: \"%s\"), %d or earlier, so US public domain as of 2026" % (year, where, quote, limit) + else: + status = "unknown" + reason = "published %d (book text, %s: \"%s\"), later than %d, so the date rule does not clear it; confirm before sharing" % (year, where, quote, limit) + today = datetime.date.today().isoformat() + book["published_year"] = year + book["license"] = {"status": status, "reason": reason, "checked": today} + with open(os.path.join(book_dir, "LICENSE_STATUS.md"), "w", encoding="utf-8") as f: + f.write("# License status\n\nStatus: %s\n\nReason: %s\n\nDate checked: %s\n\n" + "Rule: published %d or earlier is public domain in the US as of 2026; anything else is unknown until confirmed. " + "Status never blocks audio production. Nothing is published.\n" % (status, reason, today, limit)) diff --git a/_pipeline/llm.py b/_pipeline/llm.py new file mode 100644 index 0000000..91931ea --- /dev/null +++ b/_pipeline/llm.py @@ -0,0 +1,18 @@ +"""One call to an OpenAI-compatible chat endpoint. Used only for summaries, and only when a switch is on.""" +import json +import os +import urllib.request + + +def complete(cfg, system, user): + c = cfg["llm"] + if not c["endpoint"]: + raise SystemExit("a summary switch is on but llm.endpoint is empty in settings.yaml") + headers = {"Content-Type": "application/json"} + if c.get("api_key_env"): + headers["Authorization"] = "Bearer " + os.environ[c["api_key_env"]] + body = {"model": c["model"], "temperature": 0, + "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}]} + req = urllib.request.Request(c["endpoint"], json.dumps(body).encode(), headers) + with urllib.request.urlopen(req, timeout=c["timeout_s"]) as r: + return json.load(r)["choices"][0]["message"]["content"].strip() diff --git a/_pipeline/pageparts.py b/_pipeline/pageparts.py new file mode 100644 index 0000000..835efa5 --- /dev/null +++ b/_pipeline/pageparts.py @@ -0,0 +1,65 @@ +"""Recognize the non-text parts of a page: running header, page number, figure caption.""" +import re +from difflib import SequenceMatcher + +CAPTION_RE = re.compile(r"^\s*(LEFT|RIGHT|ABOVE|BELOW|OPPOSITE|FACING)\s*:\s*([A-Z][A-Z .,'\-]+?)\s*$") +_NUM_LEAD = re.compile(r"^\s*(\d{1,3})\s{2,}(\S.*)$") +_NUM_TRAIL = re.compile(r"^(.*\S)\s{2,}(\d{1,3})\s*$") +_NUM_ONLY = re.compile(r"^\s*(\d{1,3})\s*$") + + +def norm(text): + return re.sub(r"[^a-z]", "", text.lower()) + + +def similar(a, b): + return SequenceMatcher(None, norm(a), norm(b)).ratio() + + +def header_zone(lines, title): + """The running header at the top of a unit. + Returns (first_line, last_line_exclusive, page_number or None) or None. + A header is the first block of lines (up to a blank line) that either carries a page number + next to uppercase text, or reads as the book title.""" + i = 0 + while i < len(lines) and not lines[i].strip(): + i += 1 + j = i + while j < len(lines) and lines[j].strip(): + j += 1 + zone = lines[i:j] + if not zone: + return None + number, rest = None, [] + for ln in zone: + m = _NUM_LEAD.match(ln) or None + if m and number is None: + number, rest = int(m.group(1)), rest + [m.group(2)] + continue + m = _NUM_TRAIL.match(ln) + if m and number is None: + number, rest = int(m.group(2)), rest + [m.group(1)] + continue + m = _NUM_ONLY.match(ln) + if m and number is None: + number = int(m.group(1)) + continue + rest.append(ln) + text = " ".join(rest) + letters = [c for c in text if c.isalpha()] + mostly_upper = (not letters) or sum(c.isupper() for c in letters) / len(letters) > 0.7 + if number is not None and mostly_upper and len(zone) <= 3: + return i, j, number + if title and similar(text, title) >= 0.9: + return i, j, None + return None + + +def footer_number(lines): + """A lone page number as the last non-empty line. Returns (line_index, number) or None.""" + k = len(lines) - 1 + while k >= 0 and not lines[k].strip(): + k -= 1 + if k >= 0 and _NUM_ONLY.match(lines[k]): + return k, int(lines[k]) + return None diff --git a/_pipeline/pdftext.py b/_pipeline/pdftext.py new file mode 100644 index 0000000..794a0f8 --- /dev/null +++ b/_pipeline/pdftext.py @@ -0,0 +1,76 @@ +"""Read a PDF's text layer and image positions with poppler. No guessing: coordinates in, text out.""" +import os +import re +import subprocess +import tempfile +import xml.etree.ElementTree as ET + + +def _run(cmd): + return subprocess.run(cmd, check=True, capture_output=True, text=True).stdout + + +def page_geometry(pdf): + """[(width, height)] per page as displayed (rotation applied).""" + n = int(re.search(r"^Pages:\s+(\d+)", _run(["pdfinfo", pdf]), re.M).group(1)) + out = _run(["pdfinfo", "-f", "1", "-l", str(n), pdf]) + sizes = {int(m[0]): (float(m[1]), float(m[2])) for m in re.findall(r"^Page\s+(\d+) size:\s+([\d.]+) x ([\d.]+)", out, re.M)} + rots = {int(m[0]): int(m[1]) for m in re.findall(r"^Page\s+(\d+) rot:\s+(\d+)", out, re.M)} + geo = [] + for p in range(1, n + 1): + w, h = sizes[p] + if rots.get(p, 0) in (90, 270): + w, h = h, w + geo.append((w, h)) + return geo + + +def halves(width, height): + """A landscape page that holds two book pages is read as two units, left then right.""" + if width > height * 1.2: + return [("L", 0, width / 2), ("R", width / 2, width / 2)] + return [(None, 0, width)] + + +def chars_per_page(pdf): + """Number of non-space characters in the text layer of each page.""" + n = len(page_geometry(pdf)) + out = [] + for p in range(1, n + 1): + txt = _run(["pdftotext", "-f", str(p), "-l", str(p), pdf, "-"]) + out.append(len(re.sub(r"\s", "", txt))) + return out + + +def units(pdf): + """Reading units in order: [{'pdf_page', 'half', 'text'}]. Text keeps the layout (indentation, hyphens).""" + result = [] + for p, (w, h) in enumerate(page_geometry(pdf), start=1): + for half, x, width in halves(w, h): + text = _run(["pdftotext", "-layout", "-f", str(p), "-l", str(p), + "-x", str(int(x)), "-y", "0", "-W", str(int(width)), "-H", str(int(h)), pdf, "-"]) + result.append({"pdf_page": p, "half": half, "text": text.replace("\f", "")}) + return result + + +def images(pdf): + """Images with their page, half and pixel size: [{'pdf_page', 'half', 'width', 'height'}].""" + geo = page_geometry(pdf) + found = [] + with tempfile.TemporaryDirectory() as tmp: + subprocess.run(["pdftohtml", "-xml", "-q", pdf, os.path.join(tmp, "o")], check=True, capture_output=True) + root = ET.parse(os.path.join(tmp, "o.xml")).getroot() + for page in root.iter("page"): + p = int(page.get("number")) + w, h = geo[p - 1] + scale = float(page.get("width")) / w + for img in page.iter("image"): + cx = (float(img.get("left")) + float(img.get("width")) / 2) / scale + half = None + for name, x, width in halves(w, h): + if x <= cx < x + width: + half = name + found.append({"pdf_page": p, "half": half, + "width": round(float(img.get("width")) / scale), + "height": round(float(img.get("height")) / scale)}) + return found diff --git a/_pipeline/requirements.txt b/_pipeline/requirements.txt new file mode 100644 index 0000000..c77f202 --- /dev/null +++ b/_pipeline/requirements.txt @@ -0,0 +1,3 @@ +pyyaml +pyspellchecker +ocrmypdf diff --git a/_pipeline/script.py b/_pipeline/script.py new file mode 100644 index 0000000..0d51463 --- /dev/null +++ b/_pipeline/script.py @@ -0,0 +1,104 @@ +"""Stage 4, narration script. One JSON file per audio file, Audible style: +opening-credits, chapter-NN (one chapter per file), closing-credits, retail-sample. +A script is a list of segments {kind, added, text}. Author text is verbatim (added=false). +Everything else is added=true. Each inserted part (preview, summary) is wrapped: +cue chime-start, label, text, cue chime-end.""" +import json +import os + +import bookio +import llm +import structure + +LABELS = {"chapter_preview": "Chapter preview", "chapter_summary": "Chapter summary", "concept_summary": "Concept summary"} +PROMPTS = { + "chapter_preview": "In two sentences of plain spoken English, say what this chapter is about. Use only what the text says.", + "chapter_summary": "In three sentences of plain spoken English, summarize this chapter. Use only what the text says.", + "concept_summary": "In two sentences of plain spoken English, restate the main concept in this passage. Use only what the text says.", +} + + +def switch(cfg, book, chapter, name): + """Chapter switch beats book switch beats settings default.""" + for source in (chapter.get("switches", {}), book.get("summaries", {})): + if source.get(name) is not None: + return bool(source[name]) + return bool(cfg["summaries"][name]) + + +def seg(kind, text, added=True): + return {"kind": kind, "added": added, "text": text} + + +def inserted(kind, text): + return [seg("cue", "chime-start"), seg("label", LABELS[kind]), seg(kind, text), seg("cue", "chime-end")] + + +def spoken_header(ch): + n = ch["number"] + word = structure.WORDS[n - 1].title() if n <= len(structure.WORDS) else str(n) + return "Chapter %s. %s." % (word, ch["title"]) + + +def credits(book, cfg, closing): + by = "written by %s" % book["author"] + if cfg.get("narrator"): + by += ", narrated by %s" % cfg["narrator"] + return [seg("credits", ("This is the end of %s, %s." if closing else "This is %s, %s.") % (book["title"], by))] + + +def chapter_script(cfg, book, ch, paragraphs): + segs = [seg("header", spoken_header(ch))] + full = "\n\n".join(paragraphs) + if switch(cfg, book, ch, "chapter_preview"): + segs += inserted("chapter_preview", llm.complete(cfg, PROMPTS["chapter_preview"], full)) + mid = None + if switch(cfg, book, ch, "concept_summary"): + total, running = sum(len(p.split()) for p in paragraphs), 0 + for i, p in enumerate(paragraphs): + running += len(p.split()) + if running * 2 >= total: + mid = i + break + for i, p in enumerate(paragraphs): + segs.append(seg("author", p, added=False)) + if i == mid: + segs += inserted("concept_summary", llm.complete(cfg, PROMPTS["concept_summary"], "\n\n".join(paragraphs[:i + 1]))) + if switch(cfg, book, ch, "chapter_summary"): + segs += inserted("chapter_summary", llm.complete(cfg, PROMPTS["chapter_summary"], full)) + return segs + + +def retail_sample(cfg, ch, paragraphs): + budget, segs, words = cfg["script"]["retail_sample_words"], [seg("header", spoken_header(ch))], 0 + for p in paragraphs: + n = len(p.split()) + if segs[1:] and words + n > budget: + break + segs.append(seg("author", p, added=False)) + words += n + return segs + + +def write(book_dir, name, segs): + os.makedirs(os.path.join(book_dir, "script"), exist_ok=True) + with open(os.path.join(book_dir, "script", name + ".json"), "w", encoding="utf-8") as f: + json.dump(segs, f, indent=2, ensure_ascii=False) + f.write("\n") + + +def run(cfg, book_dir, book, chapter_numbers): + """Writes script/*.json and the audio file list in book.json. Returns warnings.""" + files = [("opening-credits", credits(book, cfg, False))] + chosen = [c for c in book["chapters"] if c["number"] in chapter_numbers] + for ch in chosen: + with open(os.path.join(book_dir, ch["text_file"]), encoding="utf-8") as f: + paragraphs = [p for p in f.read().split("\n\n") if p.strip()] + files.append(("chapter-%02d" % ch["number"], chapter_script(cfg, book, ch, [p.strip() for p in paragraphs]))) + if ch is chosen[0]: + sample = ("retail-sample", retail_sample(cfg, ch, [p.strip() for p in paragraphs])) + files += [("closing-credits", credits(book, cfg, True)), sample] + for name, segs in files: + write(book_dir, name, segs) + book["audio_files"] = [n for n, _ in files] + return [] diff --git a/_pipeline/settings.py b/_pipeline/settings.py new file mode 100644 index 0000000..d545563 --- /dev/null +++ b/_pipeline/settings.py @@ -0,0 +1,30 @@ +"""Load settings.yaml. One place for every tunable value.""" +import os +import yaml + +PIPELINE_DIR = os.path.dirname(os.path.abspath(__file__)) +REPO_ROOT = os.path.dirname(PIPELINE_DIR) + + +def load(path=None): + path = path or os.environ.get("AUDIOBOOK_SETTINGS") or os.path.join(PIPELINE_DIR, "settings.yaml") + with open(path, encoding="utf-8") as f: + cfg = yaml.safe_load(f) + cfg["_path"] = path + return cfg + + +def resolve(cfg, key): + """A path setting as an absolute path (relative values are relative to the repo root).""" + value = cfg[key] + return value if os.path.isabs(value) else os.path.join(REPO_ROOT, value) + + +def audio_dir(cfg, book_dir): + """Where audio for a book goes: the book folder by default, else under archive_dir.""" + if not cfg.get("archive_dir"): + return os.path.join(book_dir, "audio") + base = cfg["archive_dir"] + base = base if os.path.isabs(base) else os.path.join(REPO_ROOT, base) + parts = os.path.normpath(book_dir).split(os.sep)[-2:] # / + return os.path.join(base, *parts, "audio") diff --git a/_pipeline/settings.yaml b/_pipeline/settings.yaml new file mode 100644 index 0000000..7ea7439 --- /dev/null +++ b/_pipeline/settings.yaml @@ -0,0 +1,75 @@ +# One settings file for the whole pipeline. Relative paths are relative to the repo root. + +books_dir: books # where book folders live: books/// +work_dir: /tmp/audiobook-work # scratch only (OCR output, wav pieces). Point at local NVMe in production. + +# Where finished audio goes. Empty = the book folder in the repo (the agreed default). +# Set to a folder to write ///audio there instead. +archive_dir: "" + +narrator: "" # spoken in the credits only if filled in (leave empty for a TTS voice) + +# Inserted parts. All OFF by default. Per-book and per-chapter switches in book.json override these. +summaries: + chapter_preview: false + chapter_summary: false + concept_summary: false # one mid-chapter concept summary + +# OpenAI-compatible chat endpoint, used only when a summary switch is on. +llm: + endpoint: "" # e.g. http://host:8081/v1/chat/completions + model: "" + api_key_env: "" # name of an environment variable that holds the key (never put a key here) + timeout_s: 300 + +# OpenAI-compatible speech endpoint. +voice: + endpoint: http://127.0.0.1:8000/v1/audio/speech + model: chatterbox + voice: default + extra: {} # extra JSON fields sent with every request (for example speed) + timeout_s: 600 + max_chunk_chars: 300 # long paragraphs are cut at sentence ends into pieces this size or smaller + gap_sentence_s: 0.25 # pause between pieces inside a paragraph + gap_paragraph_s: 0.7 # pause between paragraphs and after cues and labels + +ocr: + enabled: true # run ocrmypdf only when a page has no text layer + language: eng + min_chars_per_page: 10 # fewer characters than this = no text layer + +script: + retail_sample_words: 400 # about 2.5 minutes at 155 words per minute (ACX asks for 1 to 5 minutes) + chime_label: Chime + +license: + public_domain_max_year: 1930 # US, as of 2026 + +audio: + sample_rate: 44100 + channels: 1 + mp3_bitrate: 192k + head_room_tone_s: 0.75 # ACX 0.5 to 1 + tail_room_tone_s: 2.0 # ACX 1 to 5 + room_tone_db: -66 # RMS level of the room tone (ACX noise floor limit is -60) + target_rms_db: -20.5 # ACX -23 to -18 + peak_ceiling_db: -4.0 # limiter ceiling; ACX limit is -3 and MP3 encoding can add a little + chime_hz: 660 + chime_s: 1.0 + chime_db: -18 # peak level of the chime in the final file, kept soft + +acx: # limits used by the checker (R13) + min_bitrate_kbps: 192 + sample_rate: 44100 + rms_min_db: -23 + rms_max_db: -18 + peak_max_db: -3 + noise_floor_max_db: -60 + head_min_s: 0.5 + head_max_s: 1.0 + tail_min_s: 1.0 + tail_max_s: 5.0 + max_minutes: 120 + retail_min_s: 60 + retail_max_s: 300 + speech_window_db: -50 # a 50 ms window louder than this counts as speech when measuring head and tail diff --git a/_pipeline/structure.py b/_pipeline/structure.py new file mode 100644 index 0000000..eae3f6b --- /dev/null +++ b/_pipeline/structure.py @@ -0,0 +1,140 @@ +"""Stage 3, structure: find chapter boundaries from the text layer. Audible style file list is made by the script stage.""" +import re + +import bookio +import pageparts +import pdftext + +WORDS = "ONE TWO THREE FOUR FIVE SIX SEVEN EIGHT NINE TEN ELEVEN TWELVE THIRTEEN FOURTEEN FIFTEEN SIXTEEN SEVENTEEN EIGHTEEN NINETEEN TWENTY".split() +ROMAN = {"I": 1, "V": 5, "X": 10, "L": 50} +HEADING_RE = re.compile(r"^\s*CHAPTER\s+([A-Z]+|\d+)\s*$") + + +def chapter_number(token): + if token.isdigit(): + return int(token) + if token in WORDS: + return WORDS.index(token) + 1 + if set(token) <= set(ROMAN): + total = 0 + for a, b in zip(token, token[1:] + " "): + total += -ROMAN[a] if b in ROMAN and ROMAN[b] > ROMAN[a] else ROMAN[a] + return total + return None + + +def find_headings(units): + """'CHAPTER ONE' then the title line(s) up to a blank line, within the first 12 lines of a unit.""" + found = [] + for ui, u in enumerate(units): + lines = u["text"].split("\n") + shown = 0 + for li, ln in enumerate(lines): + if ln.strip(): + shown += 1 + if shown > 12: + break + m = HEADING_RE.match(ln) + n = chapter_number(m.group(1)) if m else None + if n is None: + continue + k = li + 1 + while k < len(lines) and not lines[k].strip(): + k += 1 + title = [] + while k < len(lines) and lines[k].strip(): + title.append(lines[k].strip()) + k += 1 + found.append({"unit": ui, "line": li, "title_end": k, "number": n, + "title": bookio.title_case(" ".join(title))}) + break + return found + + +def toc_chapters(units, before_unit): + """Chapter lines of the table of contents, read from the units before the first chapter: [(roman, title, page)].""" + lines = [] + for u in units[:before_unit]: + lines += u["text"].split("\n") + out = [] + for i, ln in enumerate(lines): + m = re.match(r"^\s*([IVX]+)\.\s+(.+?)\s*\.{3,}\s*(\d+)?\s*$", ln) + if not m: + continue + page = m.group(3) + if page is None and i + 1 < len(lines) and re.match(r"^\s*\d+\s*$", lines[i + 1]): + page = lines[i + 1].strip() + out.append((chapter_number(m.group(1)), bookio.title_case(m.group(2)), int(page) if page else None)) + return out + + +def book_pages(units, title): + """Printed page number of each unit (None if unknown). Printed numbers are read from headers/footers; + a unit without one takes its neighbor's number plus or minus its distance.""" + printed = [] + for u in units: + lines = u["text"].split("\n") + z = pageparts.header_zone(lines, title) + num = z[2] if z else None + if num is None: + f = pageparts.footer_number(lines) + num = f[1] if f else None + printed.append(num) + anchors = [i for i, n in enumerate(printed) if n is not None] + pages = [] + for i in range(len(units)): + if printed[i] is not None: + pages.append(printed[i]) + continue + near = min(anchors, key=lambda a: (abs(a - i), a > i)) if anchors else None + guess = printed[near] + (i - near) if near is not None else None + pages.append(guess if guess and guess >= 1 else None) + return pages, printed + + +def run(book_dir, book, pdf): + """Fills book['chapters']. Returns warnings.""" + units = pdftext.units(pdf) + heads = find_headings(units) + warnings = [] + if not heads: + raise SystemExit("no chapter headings found") + pages, printed = book_pages(units, book["title"]) + toc = {n: (t, p) for n, t, p in toc_chapters(units, heads[0]["unit"])} + old = {c["number"]: c for c in book.get("chapters", [])} + chapters = [] + for k, h in enumerate(heads): + if k + 1 < len(heads): + nh = heads[k + 1] + last_unit = nh["unit"] - 1 + lines = units[nh["unit"]]["text"].split("\n") + z = pageparts.header_zone(lines, book["title"]) + stray = [l for i, l in enumerate(lines[:nh["line"]]) if l.strip() and not (z and z[0] <= i < z[1])] + if stray: + warnings.append(("chapter-starts-mid-page", "chapter %d" % nh["number"], + "text sits above the heading on its page; it is not assigned to the previous chapter")) + else: + last_unit = len(units) - 1 + ch = { + "number": h["number"], + "title": h["title"], + "first_unit": h["unit"], + "last_unit": last_unit, + "pdf_pages": [units[h["unit"]]["pdf_page"], units[last_unit]["pdf_page"]], + "book_pages": [pages[h["unit"]], pages[last_unit]], + "switches": old.get(h["number"], {}).get("switches", {}), + } + if h["number"] in toc: + t_title, t_page = toc[h["number"]] + ch["toc_page"] = t_page + if t_page is not None and t_page != ch["book_pages"][0]: + warnings.append(("toc-mismatch", "chapter %d" % h["number"], + "table of contents says page %s, structure found page %s" % (t_page, ch["book_pages"][0]))) + if pageparts.similar(t_title, h["title"]) < 0.9: + warnings.append(("toc-mismatch", "chapter %d" % h["number"], + "table of contents title '%s' differs from heading '%s'" % (t_title, h["title"]))) + chapters.append(ch) + warnings.append(("last-chapter-end", "chapter %d" % chapters[-1]["number"], + "no later heading, so the chapter is taken to the end of the book (back matter, if any, is included)")) + book["chapters"] = chapters + return warnings diff --git a/_pipeline/tests/test_pipeline.py b/_pipeline/tests/test_pipeline.py new file mode 100644 index 0000000..a27d23c --- /dev/null +++ b/_pipeline/tests/test_pipeline.py @@ -0,0 +1,137 @@ +"""Tests. Everything audio here is a synthetic TEST signal made locally, never pilot audio. +Run: python3 -m unittest discover -s _pipeline/tests -v (from the repo root)""" +import http.server +import json +import os +import sys +import tempfile +import threading +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) + +import acx_check +import assemble +import audiomeasure as am +import script +import settings +import voice + +CFG = settings.load() + + +def test_piece(path, hz, seconds, level): + """TEST signal: speech-like (harmonics, 4 Hz syllable rhythm, a little noise) at amplitude `level`.""" + expr = "%g*(0.6*sin(2*PI*%g*t)+0.3*sin(4*PI*%g*t)+0.1*sin(6*PI*%g*t))*(0.55+0.45*sin(2*PI*4*t))" % (level, hz, hz, hz) + am.ffmpeg(["-f", "lavfi", "-i", "aevalsrc='%s':s=24000:d=%g" % (expr, seconds), "-ar", "24000", "-ac", "1", path]) + + +class AudioTests(unittest.TestCase): + def test_assemble_then_check_passes(self): + with tempfile.TemporaryDirectory() as d: + items = [] + for i, (hz, lvl) in enumerate([(120, 0.02), (150, 0.05), (130, 0.03)]): + p = os.path.join(d, "t%d.wav" % i) + test_piece(p, hz, 4, lvl) + items.append({"wav": p, "gap_after_s": 0.7}) + if i == 0: + items.append({"cue": "chime", "gap_after_s": 0.25}) + out = os.path.join(d, "audio", "chapter-99.mp3") + gain = assemble.assemble_items(CFG, items, out) + m = acx_check.measure(CFG, out) + print("\nTEST assembled: gain %.1f dB, measured %s" % (gain, {k: (round(v, 2) if isinstance(v, float) else v) for k, v in m.items()})) + for name, value, passed in acx_check.judge(CFG, out, m): + self.assertTrue(passed, "%s %s" % (name, value)) + + def test_checker_fails_bad_files(self): + with tempfile.TemporaryDirectory() as d: + raw = os.path.join(d, "raw.wav") + test_piece(raw, 130, 4, 1.4) # hot test signal, clips at full scale + bad = os.path.join(d, "bad.mp3") # stereo, 128k, 22.05 kHz, no room tone + am.ffmpeg(["-i", raw, "-ac", "2", "-ar", "22050", "-c:a", "libmp3lame", "-b:a", "128k", bad]) + failed = {n for n, v, p in acx_check.judge(CFG, bad, acx_check.measure(CFG, bad)) if not p} + print("\nTEST bad file 1 fails on:", sorted(failed)) + self.assertTrue({"format", "rms dB", "noise floor dB", "head s", "tail s"} <= failed) + hot = os.path.join(d, "hot.mp3") # correct format, but too loud + am.ffmpeg(["-i", raw, "-ac", "1", "-ar", "44100", "-c:a", "libmp3lame", "-b:a", "192k", hot]) + m = acx_check.measure(CFG, hot) + failed = {n for n, v, p in acx_check.judge(CFG, hot, m) if not p} + print("TEST bad file 2 (peak %.1f dB) fails on:" % m["peak_db"], sorted(failed)) + self.assertIn("peak dB", failed) + self.assertNotIn("format", failed) + + +class StubHandler(http.server.BaseHTTPRequestHandler): + seen = [] + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + StubHandler.seen.append((self.path, body)) + if self.path.endswith("/audio/speech"): + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "x.wav") + test_piece(p, 130, 1, 0.1) + data = open(p, "rb").read() + self.send_response(200) + self.send_header("Content-Type", "audio/wav") + else: + data = json.dumps({"choices": [{"message": {"content": "A test summary."}}]}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *a): + pass + + +class ClientTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = http.server.HTTPServer(("127.0.0.1", 0), StubHandler) + threading.Thread(target=cls.server.serve_forever, daemon=True).start() + cls.url = "http://127.0.0.1:%d" % cls.server.server_port + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + + def cfg(self): + c = json.loads(json.dumps(CFG)) + c["voice"]["endpoint"] = self.url + "/v1/audio/speech" + c["llm"].update(endpoint=self.url + "/v1/chat/completions", model="m") + return c + + def test_voice_request_and_wav(self): + StubHandler.seen.clear() + data = voice.speak(self.cfg(), "Hello there.") + path, body = StubHandler.seen[0] + self.assertEqual(path, "/v1/audio/speech") + self.assertEqual(body["input"], "Hello there.") + self.assertEqual(body["response_format"], "wav") + self.assertEqual(data[:4], b"RIFF") + + def test_sentence_split_keeps_abbreviations(self): + s = voice.sentences('One day Dr. Alexis Carrel sent word. "The world," he said. E. Stanley Jones came.') + self.assertEqual(len(s), 3) + + def test_script_wraps_inserted_parts_when_switches_on(self): + c = self.cfg() + c["summaries"].update(chapter_preview=True, chapter_summary=True, concept_summary=True) + ch = {"number": 1, "title": "We Go Seeking", "switches": {}} + segs = script.chapter_script(c, {"summaries": {}}, ch, ["Para one is here.", "Para two is here.", "Para three is here."]) + kinds = [s["kind"] for s in segs] + self.assertEqual(kinds[:5], ["header", "cue", "label", "chapter_preview", "cue"]) + self.assertEqual(kinds[-4:], ["cue", "label", "chapter_summary", "cue"]) + self.assertEqual(kinds.count("concept_summary"), 1) + self.assertEqual([s["text"] for s in segs if s["kind"] == "author"], ["Para one is here.", "Para two is here.", "Para three is here."]) + + def test_switches_default_off_and_chapter_wins(self): + ch = {"number": 1, "title": "x", "switches": {"chapter_summary": True}} + self.assertFalse(script.switch(CFG, {"summaries": {}}, {"switches": {}}, "chapter_summary")) + self.assertTrue(script.switch(CFG, {"summaries": {"chapter_summary": False}}, ch, "chapter_summary")) + + +if __name__ == "__main__": + unittest.main() diff --git a/_pipeline/voice.py b/_pipeline/voice.py new file mode 100644 index 0000000..fb61f37 --- /dev/null +++ b/_pipeline/voice.py @@ -0,0 +1,90 @@ +"""Voice client: script segments -> wav pieces, via an OpenAI-compatible /v1/audio/speech endpoint (URL in settings). +Pieces are sentence-sized. Output goes to scratch (work_dir), with a manifest the assembler reads.""" +import json +import os +import re +import urllib.request + +import bookio +import settings + +ABBREV = re.compile(r"^(Dr|Mr|Mrs|Ms|St|Mt|Jr|Sr|vs|[A-Z])\.$") + + +def sentences(text): + """Split at . ! ? (optionally followed by a quote) and a space, but not after abbreviations like 'Dr.' or 'E.'.""" + out, start = [], 0 + for m in re.finditer(r"[.!?][\"']?\s+", text): + before = text[start:m.start() + 1].split()[-1] if text[start:m.start() + 1].split() else "" + if ABBREV.match(before): + continue + out.append(text[start:m.end()].strip()) + start = m.end() + if text[start:].strip(): + out.append(text[start:].strip()) + return out + + +def pieces(text, max_chars): + """Pack whole sentences into pieces of at most max_chars (a longer sentence stays whole).""" + out, cur = [], "" + for s in sentences(text): + if cur and len(cur) + 1 + len(s) > max_chars: + out.append(cur) + cur = s + else: + cur = (cur + " " + s).strip() + if cur: + out.append(cur) + return out + + +def plan(cfg, segs): + """Ordered items: {'text', 'gap_after_s'} for speech, {'cue': 'chime', 'gap_after_s'} for cues.""" + v = cfg["voice"] + items = [] + for s in segs: + if s["kind"] == "cue": + items.append({"cue": "chime", "gap_after_s": v["gap_sentence_s"]}) + continue + ps = pieces(s["text"], v["max_chunk_chars"]) + for i, p in enumerate(ps): + items.append({"text": p, "gap_after_s": v["gap_paragraph_s"] if i == len(ps) - 1 else v["gap_sentence_s"]}) + return items + + +def speak(cfg, text): + v = cfg["voice"] + body = {"model": v["model"], "input": text, "voice": v["voice"], "response_format": "wav"} + body.update(v.get("extra") or {}) + req = urllib.request.Request(v["endpoint"], json.dumps(body).encode(), {"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=v["timeout_s"]) as r: + data = r.read() + if data[:4] != b"RIFF": + raise RuntimeError("voice endpoint did not return a wav file") + return data + + +def work_dir_for(cfg, book_dir, name): + return os.path.join(settings.resolve(cfg, "work_dir"), os.path.basename(book_dir), "audio", name) + + +def run(cfg, book_dir, only=None): + book = bookio.read_book(book_dir) + for name in book["audio_files"]: + if only and name != only: + continue + with open(os.path.join(book_dir, "script", name + ".json"), encoding="utf-8") as f: + items = plan(cfg, json.load(f)) + out = work_dir_for(cfg, book_dir, name) + os.makedirs(out, exist_ok=True) + for i, it in enumerate(items): + if "text" in it: + it["wav"] = os.path.join(out, "%04d.wav" % i) + if not os.path.exists(it["wav"]): # resume: finished pieces are kept + data = speak(cfg, it["text"]) + with open(it["wav"], "wb") as f: + f.write(data) + with open(os.path.join(out, "manifest.json"), "w", encoding="utf-8") as f: + json.dump(items, f, indent=1, ensure_ascii=False) + print("%s: %d pieces" % (name, len(items)))