From 8c9ea145687ff936afde044e952a4f95139543e7 Mon Sep 17 00:00:00 2001 From: Audiobook Producer Date: Sun, 18 Oct 2026 01:34:58 -0500 Subject: [PATCH] Add 'process': one command for a whole book (all chapters), idempotent, loud failures, OCR required; synthetic test-PDF tool --- _pipeline/Dockerfile | 2 +- _pipeline/cli.py | 13 ++ _pipeline/process.py | 278 +++++++++++++++++++++++++++++++ _pipeline/requirements.txt | 2 +- _pipeline/tests/test_process.py | 200 ++++++++++++++++++++++ _pipeline/tools/make_test_pdf.py | 163 ++++++++++++++++++ 6 files changed, 656 insertions(+), 2 deletions(-) create mode 100644 _pipeline/process.py create mode 100644 _pipeline/tests/test_process.py create mode 100644 _pipeline/tools/make_test_pdf.py diff --git a/_pipeline/Dockerfile b/_pipeline/Dockerfile index c3d9e1e..b1dee9a 100644 --- a/_pipeline/Dockerfile +++ b/_pipeline/Dockerfile @@ -4,7 +4,7 @@ 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 \ + && apt-get install -y --no-install-recommends ffmpeg poppler-utils tesseract-ocr tesseract-ocr-eng ocrmypdf ghostscript \ && rm -rf /var/lib/apt/lists/* COPY _pipeline/requirements.txt /tmp/requirements.txt diff --git a/_pipeline/cli.py b/_pipeline/cli.py index bcb51b6..ff1f98c 100644 --- a/_pipeline/cli.py +++ b/_pipeline/cli.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 """One command line for the whole pipeline. + python3 _pipeline/cli.py process [--dry-run] [--force] EVERYTHING for a whole book, all chapters; exit 1 on any failure 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) @@ -61,6 +62,15 @@ def main(argv=None): 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("process", help="whole book, all stages, all chapters") + p.add_argument("pdf") + p.add_argument("--dry-run", action="store_true", help="text stages only; checks the voice server is reachable; renders nothing") + p.add_argument("--force", action="store_true", help="re-render audio even if it exists") + p.add_argument("--title") + p.add_argument("--author") + p.add_argument("--compare", help="a books folder (like the repo books/); the text-stage output is compared with the same / folder in it") + p.add_argument("--compare-strict", action="store_true", help="with --compare: fail when the output differs (default: report only)") + p.add_argument("--result-file", help="write {ok, book_dir, pdf} JSON here") p = sub.add_parser("ingest") p.add_argument("pdf") p.add_argument("--title") @@ -82,6 +92,9 @@ def main(argv=None): a = ap.parse_args(argv) cfg = settings.load(a.settings) + if a.cmd == "process": + import process + sys.exit(process.process(cfg, a.pdf, a.dry_run, a.force, a.title, a.author, a.compare, a.result_file, a.compare_strict)) 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) diff --git a/_pipeline/process.py b/_pipeline/process.py new file mode 100644 index 0000000..c6d0ee7 --- /dev/null +++ b/_pipeline/process.py @@ -0,0 +1,278 @@ +"""One command for a whole book: PDF in, finished book folder out. + + python3 _pipeline/cli.py process [--dry-run] [--force] [--title T --author A] [--compare BOOKS_ROOT [--compare-strict]] [--result-file F] + +Stages, in order, for ALL chapters the structure stage finds: + ingest (OCR only if pages lack a text layer) -> structure -> cleanup -> figures -> script -> license + -> [voice -> assemble -> ACX check] (skipped with --dry-run, which instead proves the voice server is reachable) + +Rules: + * Any failed step or failed ACX check stops the run with a clear message and exit code 1. + * Warnings are logged to warnings.log in the book folder. There is no approval gate. + * Idempotent: an audio file that already exists is skipped, unless its script or the voice settings changed + since it was built (book.json "audio_built") or --force is given. + * Output goes where the settings say: books_dir (env AUDIOBOOK_BOOKS_DIR), scratch work_dir (env AUDIOBOOK_WORK_DIR), + voice endpoint (env AUDIOBOOK_VOICE_ENDPOINT). +""" +import difflib +import hashlib +import io +import json +import os +import re +import shutil +import sys + +import acx_check +import assemble +import bookio +import ingest +import settings +import voice + + +class Failed(Exception): + """A pipeline step failed. The message is what the operator reads.""" + + +class Tee(io.TextIOBase): + def __init__(self, *streams): + self.streams = streams + + def write(self, s): + for st in self.streams: + st.write(s) + return len(s) + + def flush(self): + for st in self.streams: + st.flush() + + +def sidecar(pdf): + """Optional .json next to the PDF: {"title": "...", "author": "..."}.""" + path = os.path.splitext(pdf)[0] + ".json" + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + return json.load(f) + return {} + + +def guess_from_filename(pdf): + """'The Title - by Glenn Clark (Extra).pdf' or 'Title by Glenn Clark.pdf' -> (title, author), else (None, None).""" + stem = os.path.splitext(os.path.basename(pdf))[0] + m = re.match(r"^(?P.+?)[\s\-–—]+by\s+(?P<author>[^()]+?)\s*(\(.*\))?\s*$", stem, re.I) + if m: + return m.group("title").strip(" -–—"), m.group("author").strip() + return None, None + + +def voice_fingerprint(cfg, script_path): + h = hashlib.sha256() + with open(script_path, "rb") as f: + h.update(f.read()) + v = cfg["voice"] + h.update(json.dumps([v["api"], v["voice"], v.get("extra")], sort_keys=True).encode()) + h.update(json.dumps(cfg["audio"], sort_keys=True).encode()) + return h.hexdigest()[:16] + + +def read_if(path): + with open(path, encoding="utf-8") as f: + return f.read() + + +def book_json_for_compare(path): + book = json.loads(read_if(path)) + book.get("license", {}).pop("checked", None) + return json.dumps(book, indent=1, sort_keys=True) + + +def compare_books(new_dir, old_dir): + """Text-stage output of new_dir against an existing book folder. Returns a list of difference descriptions.""" + diffs = [] + pairs = [("book.json", book_json_for_compare), ("warnings.log", read_if), ("figures/skipped_figures.md", read_if), + ("LICENSE_STATUS.md", lambda p: re.sub(r"Date checked: .*", "", read_if(p)))] + for sub in ("text", "script"): + names = set(os.listdir(os.path.join(old_dir, sub))) | set(os.listdir(os.path.join(new_dir, sub))) + pairs += [(os.path.join(sub, n), read_if) for n in sorted(names)] + for rel, reader in pairs: + a, b = os.path.join(old_dir, rel), os.path.join(new_dir, rel) + if not os.path.exists(a) or not os.path.exists(b): + diffs.append("%s: missing in %s" % (rel, "new output" if os.path.exists(a) else "existing book")) + continue + ta, tb = reader(a), reader(b) + if ta != tb: + d = list(difflib.unified_diff(ta.splitlines(), tb.splitlines(), "existing/" + rel, "new/" + rel, lineterm="", n=0)) + diffs.append("%s differs (%d diff lines):\n %s" % (rel, len(d), "\n ".join(d[:12]))) + return diffs + + +def step(name): + print("\n=== %s ===" % name, flush=True) + + +def check_ocr(cfg, book_dir): + """Pages with no text layer must have been OCRed. Fails the run if they were not (setting ocr.required).""" + book = bookio.read_book(book_dir) + missing = book["pages_without_text_layer"] + if not missing: + print("OCR: not needed, every page has a text layer") + return + ocr_pdf = os.path.join(settings.resolve(cfg, "work_dir"), os.path.basename(book_dir), "ocr.pdf") + pages = ",".join(map(str, missing)) + if not os.path.exists(ocr_pdf): + msg = "pdf pages %s have no text layer and OCR did not produce a text copy (is ocrmypdf installed? see warnings.log)" % pages + if cfg["ocr"].get("required", True): + raise Failed("ingest: " + msg) + print("WARNING:", msg) + return + import pdftext + counts = pdftext.chars_per_page(ocr_pdf) + got = {p: counts[p - 1] for p in missing} + print("OCR applied to pdf pages %s; characters found after OCR: %s" % (pages, got)) + empty = [p for p, n in got.items() if n < cfg["ocr"]["min_chars_per_page"]] + if empty: + bookio.write_warnings(book_dir, "process-ocr", [("ocr-no-text", "pdf pages " + ",".join(map(str, empty)), + "OCR found no text (blank page or picture only)")]) + + +def run_text_stages(cfg, pdf, title, author): + """Ingest through license for every chapter. Returns (book_dir, book).""" + import cli + meta = sidecar(pdf) + title, author = title or meta.get("title"), author or meta.get("author") + step("ingest") + try: + book_dir = ingest.run(cfg, pdf, title, author) + except SystemExit as e: + t, a = guess_from_filename(pdf) + if not (t and a): + raise Failed("ingest: %s. Put a file with the same name and .json next to the PDF: " + '{"title": "...", "author": "..."}' % e) + print("ingest: title/author not found in the PDF; taking them from the file name: '%s' by '%s'" % (t, a)) + book_dir = ingest.run(cfg, pdf, title or t, author or a) + bookio.write_warnings(book_dir, "ingest-metadata", [("metadata-from-filename", os.path.basename(pdf), "title '%s', author '%s'" % (t, a))]) + print("ingest done ->", book_dir) + check_ocr(cfg, book_dir) + for name in ("structure",): + step(name) + try: + cli.run_stage(name, cfg, book_dir) + except SystemExit as e: + raise Failed("%s: %s" % (name, e)) + book = bookio.read_book(book_dir) + numbers = [c["number"] for c in book["chapters"]] + if numbers != list(range(1, len(numbers) + 1)): + bookio.write_warnings(book_dir, "process", [("chapter-numbers", "structure", "chapter numbers found: %s (expected 1..%d in order)" % (numbers, len(numbers)))]) + print("chapters found:", numbers) + for name in ("cleanup", "figures", "script", "license"): + step(name) + try: + cli.run_stage(name, cfg, book_dir, numbers) + except SystemExit as e: + raise Failed("%s: %s" % (name, e)) + return book_dir, bookio.read_book(book_dir) + + +def run_audio_stages(cfg, book_dir, book, force): + out_dir = settings.audio_dir(cfg, book_dir) + built = book.setdefault("audio_built", {}) + todo = [] + for name in book["audio_files"]: + fp = voice_fingerprint(cfg, os.path.join(book_dir, "script", name + ".json")) + have = os.path.exists(os.path.join(out_dir, name + ".mp3")) + if have and not force and built.get(name, fp) == fp: + print("skip %s: audio exists and is current" % name) + else: + todo.append((name, fp)) + if todo: + step("voice (%d of %d files)" % (len(todo), len(book["audio_files"]))) + h = voice.ensure_loaded(cfg) + print("voice server ok:", h) + names = [n for n, _ in todo] + for n, fp in todo: # finished pieces from an interrupted run are reused only if script and voice settings are the same + wd = voice.work_dir_for(cfg, book_dir, n) + stamp = os.path.join(wd, "fingerprint") + if os.path.isdir(wd) and (force or not os.path.exists(stamp) or read_if(stamp) != fp): + shutil.rmtree(wd) + os.makedirs(wd, exist_ok=True) + with open(stamp, "w") as f: + f.write(fp) + voice.run(cfg, book_dir, names=names) + step("assemble") + for n in names: + assemble.run(cfg, book_dir, only=n) + for n, fp in todo: + built[n] = fp + bookio.write_book(book_dir, book) + if cfg["voice"].get("unload_when_done"): + voice.unload(cfg) + print("voice model unloaded to free the GPU") + else: + print("nothing to render") + step("ACX check") + rc = acx_check.run(cfg, [out_dir]) + if rc != 0: + raise Failed("ACX check FAILED (see the table above); nothing should be committed") + + +def process(cfg, pdf, dry_run=False, force=False, title=None, author=None, compare=None, result_file=None, compare_strict=False): + """Returns 0 on success, 1 on failure (message printed).""" + pdf = os.path.abspath(pdf) + if not os.path.isfile(pdf): + print("PROCESS FAILED: no such file: %s" % pdf) + return 1 + log = io.StringIO() + real_out = sys.stdout + sys.stdout = Tee(real_out, log) + book_dir = None + try: + print("process %s%s" % (os.path.basename(pdf), " [DRY RUN: text stages only]" if dry_run else "")) + book_dir, book = run_text_stages(cfg, pdf, title, author) + if compare: + old = os.path.join(compare, os.path.relpath(book_dir, settings.resolve(cfg, "books_dir"))) + step("compare with the existing book folder %s" % old) + if not os.path.isdir(old): + print("no existing folder to compare with (this is a new book)") + else: + diffs = compare_books(book_dir, old) + if diffs and compare_strict: + raise Failed("output differs from the existing book folder:\n" + "\n".join(diffs)) + if diffs: + print("DIFFERENCES from the existing book folder (report only; add --compare-strict to fail on them):\n" + "\n".join(diffs)) + else: + print("identical to the existing text, script, figures list, warnings, book.json and license file") + if dry_run: + step("voice server reachability (no audio rendered)") + try: + h = voice.health(cfg) + except voice.VoiceError as e: + raise Failed(str(e)) + print("voice server reachable:", h) + else: + try: + run_audio_stages(cfg, book_dir, book, force) + except voice.VoiceError as e: + raise Failed("voice: %s" % e) + print("\nPROCESS OK: %s" % book_dir) + rc = 0 + except Failed as e: + print("\nPROCESS FAILED: %s" % e) + rc = 1 + except Exception as e: # any other error is also a loud failure, with the type named + print("\nPROCESS FAILED: unexpected %s: %s" % (type(e).__name__, e)) + import traceback + traceback.print_exc(file=sys.stdout) + rc = 1 + finally: + sys.stdout = real_out + if rc == 0 and book_dir and not dry_run: + out_dir = settings.audio_dir(cfg, book_dir) + os.makedirs(out_dir, exist_ok=True) + with open(os.path.join(out_dir, "run.log"), "w", encoding="utf-8") as f: + f.write(log.getvalue()) + if result_file: + with open(result_file, "w", encoding="utf-8") as f: + json.dump({"ok": rc == 0, "book_dir": book_dir, "pdf": pdf}, f) + return rc diff --git a/_pipeline/requirements.txt b/_pipeline/requirements.txt index c77f202..e870756 100644 --- a/_pipeline/requirements.txt +++ b/_pipeline/requirements.txt @@ -1,3 +1,3 @@ pyyaml pyspellchecker -ocrmypdf +# OCR (ocrmypdf) is installed with apt (it needs tesseract and ghostscript); see .gitea/workflows/audiobook.yml and _pipeline/Dockerfile diff --git a/_pipeline/tests/test_process.py b/_pipeline/tests/test_process.py new file mode 100644 index 0000000..78b0685 --- /dev/null +++ b/_pipeline/tests/test_process.py @@ -0,0 +1,200 @@ +"""Tests for the one-command `process`. A local stub stands in for the voice server (synthetic TEST audio only). +Run: python3 -m unittest discover -s _pipeline/tests -v (from the repo root)""" +import contextlib +import http.server +import io +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__)), "..")) +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tools")) + +import audiomeasure as am +import process +import settings +import voice + +try: + import make_test_pdf +except ImportError: + make_test_pdf = None + + +class Stub(http.server.BaseHTTPRequestHandler): + posts = [] + loaded = True + + def _json(self, obj): + data = json.dumps(obj).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 do_GET(self): + if self.path == "/api/model-info": + self._json({"loaded": Stub.loaded, "class_name": "StubModel", "device": "cpu"}) + elif self.path == "/get_predefined_voices": + self._json([{"display_name": "Michael", "filename": "Michael.wav"}]) + else: + self.send_error(404) + + def do_POST(self): + n = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(n) + Stub.posts.append(self.path) + if self.path == "/restart_server": + Stub.loaded = True + return self._json({"message": "ok"}) + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "x.wav") + expr = "0.1*(0.6*sin(2*PI*130*t)+0.3*sin(4*PI*130*t))*(0.55+0.45*sin(2*PI*4*t))" + am.ffmpeg(["-f", "lavfi", "-i", "aevalsrc='%s':s=24000:d=10" % expr, "-ar", "24000", "-ac", "1", p]) + data = open(p, "rb").read() + self.send_response(200) + self.send_header("Content-Type", "audio/wav") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *a): + pass + + +def run_process(cfg, pdf, **kw): + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + rc = process.process(cfg, pdf, **kw) + return rc, buf.getvalue() + + +@unittest.skipIf(make_test_pdf is None, "reportlab not installed (only needed to build the synthetic test PDF)") +class ProcessTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = http.server.HTTPServer(("127.0.0.1", 0), Stub) + 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 setUp(self): + self.tmp = tempfile.TemporaryDirectory() + d = self.tmp.name + self.pdf = os.path.join(d, "Pipeline Smoke Test Book - by Ada Testwell.pdf") + make_test_pdf.build(self.pdf) + self.cfg = settings.load() + self.cfg["books_dir"] = os.path.join(d, "books") + self.cfg["work_dir"] = os.path.join(d, "work") + self.cfg["voice"]["endpoint"] = self.url + "/tts" + self.cfg["voice"]["retries"] = 0 + self.cfg["ocr"]["required"] = False # OCR itself is proven in the CI job; here only the other stages + Stub.posts.clear() + Stub.loaded = True + + def tearDown(self): + self.tmp.cleanup() + + def book_dir(self): + return os.path.join(self.cfg["books_dir"], "testwell-ada", "pipeline-smoke-test-book") + + def test_full_run_idempotent_and_force(self): + rc, out = run_process(self.cfg, self.pdf) + print("\nTEST first run rc=%d" % rc) + self.assertEqual(rc, 0, out) + b = self.book_dir() + names = sorted(os.listdir(os.path.join(b, "audio"))) + self.assertEqual(names, ["chapter-01.mp3", "chapter-02.mp3", "closing-credits.mp3", "opening-credits.mp3", "retail-sample.mp3", "run.log"]) + self.assertIn("OVERALL: PASS", out) + for rel in ("text/chapter-01.txt", "text/chapter-02.txt", "script/chapter-02.json", "figures/skipped_figures.md", + "book.json", "warnings.log", "LICENSE_STATUS.md"): + self.assertTrue(os.path.exists(os.path.join(b, rel)), rel) + self.assertIn("public domain", open(os.path.join(b, "LICENSE_STATUS.md")).read()) + mtimes = {n: os.path.getmtime(os.path.join(b, "audio", n)) for n in names if n.endswith(".mp3")} + posts = len(Stub.posts) + self.assertGreater(posts, 0) + rc, out = run_process(self.cfg, self.pdf) # second run: nothing to render + print("TEST second run rc=%d, new voice requests: %d" % (rc, len(Stub.posts) - posts)) + self.assertEqual(rc, 0, out) + self.assertEqual(len(Stub.posts), posts) + self.assertIn("nothing to render", out) + self.assertEqual(mtimes, {n: os.path.getmtime(os.path.join(b, "audio", n)) for n in mtimes}) + rc, out = run_process(self.cfg, self.pdf, force=True) # forced: renders again + self.assertEqual(rc, 0, out) + self.assertGreater(len(Stub.posts), posts) + + def test_changed_script_rerenders_only_that_file(self): + self.assertEqual(run_process(self.cfg, self.pdf)[0], 0) + b = self.book_dir() + path = os.path.join(b, "text", "chapter-01.txt") + # an edit to the cleaned text is overwritten by the text stages, so change the voice setting for one run instead + self.cfg["voice"]["extra"]["seed"] = 8 + posts = len(Stub.posts) + rc, out = run_process(self.cfg, self.pdf) + self.assertEqual(rc, 0, out) + self.assertGreater(len(Stub.posts), posts) + self.assertNotIn("skip chapter-01", out) + + def test_dry_run_renders_nothing_and_checks_voice(self): + rc, out = run_process(self.cfg, self.pdf, dry_run=True) + self.assertEqual(rc, 0, out) + self.assertIn("voice server reachable", out) + self.assertEqual(Stub.posts, []) + self.assertFalse(os.path.exists(os.path.join(self.book_dir(), "audio"))) + + def test_unreachable_voice_fails_clearly_with_no_audio(self): + self.cfg["voice"]["endpoint"] = "http://127.0.0.1:9/tts" + rc, out = run_process(self.cfg, self.pdf) + print("\nTEST unreachable voice rc=%d: %s" % (rc, out.strip().splitlines()[-1])) + self.assertEqual(rc, 1) + self.assertIn("PROCESS FAILED: voice: voice server not reachable", out) + self.assertFalse(os.path.exists(os.path.join(self.book_dir(), "audio"))) + + def test_unloaded_model_is_loaded_on_demand(self): + Stub.loaded = False + rc, out = run_process(self.cfg, self.pdf) + self.assertEqual(rc, 0, out) + self.assertEqual(Stub.posts[0], "/restart_server") + + def test_speed_factor_is_refused(self): + self.cfg["voice"]["extra"]["speed_factor"] = 0.9 + with self.assertRaises(voice.VoiceError): + voice.speak(self.cfg, "hello") + + def test_ocr_required_fails_without_ocr(self): + self.cfg["ocr"]["required"] = True + self.cfg["ocr"]["enabled"] = False + rc, out = run_process(self.cfg, self.pdf) + self.assertEqual(rc, 1) + self.assertIn("have no text layer", out) + + +class SmallTests(unittest.TestCase): + def test_filename_guess(self): + g = process.guess_from_filename + self.assertEqual(g("/x/The Man Who Tapped - by Glenn Clark (Walter Russell).pdf"), ("The Man Who Tapped", "Glenn Clark")) + self.assertEqual(g("/x/Some Title by Jane Doe.pdf"), ("Some Title", "Jane Doe")) + self.assertEqual(g("/x/scan0001.pdf"), (None, None)) + + def test_env_overrides(self): + cfg = settings.load() + settings.apply_env(cfg, {"AUDIOBOOK_VOICE_ENDPOINT": "http://1.2.3.4:8090/tts", "AUDIOBOOK_REQUIRE_OCR": "0"}) + self.assertEqual(cfg["voice"]["endpoint"], "http://1.2.3.4:8090/tts") + self.assertFalse(cfg["ocr"]["required"]) + + def test_locked_voice_settings_unchanged(self): + v = settings.load()["voice"] + self.assertEqual((v["voice"], v["extra"]["exaggeration"], v["extra"]["cfg_weight"], v["extra"]["temperature"], + v["extra"]["seed"], v["extra"]["split_text"], v["extra"]["chunk_size"]), ("Michael.wav", 0.25, 0.2, 0.6, 7, True, 240)) + self.assertNotIn("speed_factor", v["extra"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/_pipeline/tools/make_test_pdf.py b/_pipeline/tools/make_test_pdf.py new file mode 100644 index 0000000..a6aeb8e --- /dev/null +++ b/_pipeline/tools/make_test_pdf.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""Build a small synthetic test book (NOT a real book) to prove the pipeline end to end. +Two chapters of about 300 words, title page, copyright page, contents, chapter headings, running headers, +page numbers, and one page that is only a picture of text (no text layer), so the OCR path is used. + + pip install reportlab pillow (this tool only; the pipeline itself does not need them) + python3 _pipeline/tools/make_test_pdf.py out.pdf +""" +import sys +import tempfile +import os +from reportlab.lib.pagesizes import letter +from reportlab.lib.utils import simpleSplit, ImageReader +from reportlab.pdfgen import canvas +from PIL import Image, ImageDraw, ImageFont + +TITLE = "Pipeline Smoke Test Book" +AUTHOR = "Ada Testwell" +W, H = letter +LEFT, RIGHT, TOP = 72, W - 72, H - 90 + +CH1 = [ + "The harbor was quiet on the morning the keeper first climbed the stairs of the old lantern tower. The fishing boats had gone out before dawn, and only the gulls remained, turning slowly above the water as if they were waiting for someone to speak.", + "He had been told that the work was simple. Each evening he was to light the lamp, and each morning he was to put it out. Between those two tasks there was nothing at all, only the sea, the wind, and the long patience of the stone walls.", + "On the third day he began to notice small things. The stairs had forty-two steps, and the forty-first was always cold. The window on the north side hummed when the wind came from the east. A single brass key hung beside the door, and nobody could say what it opened.", + "He wrote these facts in a notebook, not because anyone would read them, but because a fact written down seemed to belong to him in a way that a fact merely seen did not. By the end of the first week the notebook held thirty pages, and the tower no longer felt like a stranger's house.", + "In the evenings the fishermen rowed past and raised their hands. He raised his own in return. It was a small conversation, but it was enough, and it taught him that a person can be useful without being busy, and present without being loud.", + "When the lamp was lit for the first time, the whole room turned gold, and he understood at last what the work was for. It was not the light itself, but the promise of the light, spoken every night to people he would never meet.", +] +CH2 = [ + "The road out of the harbor town climbed through pastures and then through pine woods, and it was long enough that most travelers walked it in silence. The keeper took it one spring when the supply boat failed to come, carrying only a loaf of bread and his notebook.", + "At the first crossroads he met a woman driving a cart of apples. She asked where he was going, and he said he did not know yet, only that the road seemed to know. She laughed and gave him an apple, and told him that was the best answer she had heard all year.", + "By noon the pines gave way to open hills, and from the highest of them he could see both the sea behind him and a river ahead. He sat for a while and wrote in the notebook that distance is not a number but a feeling, which grows smaller each time you decide to keep going.", + "That evening he reached a village where the lamps were already lit in every window. The scanned page of the old village register recorded his name in a careful hand, and beneath it the innkeeper wrote the date, the weather, and the single word welcome.", + "He slept soundly for the first time in many weeks. In the morning the innkeeper handed him a small brass key, saying it had been left there years before by another keeper, and that it opened the door of a tower he had not yet seen.", + "He walked home by the same road, and it seemed shorter. When the lantern tower came into view, he saw that someone had lit the lamp in his absence, and he knew then that the work had never belonged to him alone.", +] + + +def wrap(c, text, font, size, width): + return simpleSplit(text, font, size, width) + + +def draw_paragraphs(c, paras, y, font="Times-Roman", size=11, leading=14, bottom=90): + """Draws paragraphs with a first-line indent. Returns (y, leftover paragraphs).""" + left = list(paras) + while left: + lines = wrap(c, left[0], font, size, RIGHT - LEFT - 18) + # first line is narrower because of the indent; re-wrap properly + first = simpleSplit(left[0], font, size, RIGHT - LEFT - 18)[0] + rest = left[0][len(first):].strip() + lines = [first] + (simpleSplit(rest, font, size, RIGHT - LEFT) if rest else []) + if y - leading * len(lines) < bottom: + break + c.setFont(font, size) + for i, ln in enumerate(lines): + c.drawString(LEFT + (18 if i == 0 else 0), y, ln) + y -= leading + left.pop(0) + return y, left + + +def running_header(c, number): + c.setFont("Times-Roman", 10) + c.drawString(LEFT, H - 50, TITLE.upper()) + c.drawRightString(RIGHT, H - 50, str(number)) + + +def scanned_page_image(paras, path): + """A picture of typeset text: 200 dpi, serif font, no text layer once placed in the PDF.""" + dpi = 200 + pw, ph = int((RIGHT - LEFT) / 72 * dpi), int((H - 160) / 72 * dpi) + img = Image.new("L", (pw, ph), 255) + d = ImageDraw.Draw(img) + font = None + for name in ("DejaVuSerif.ttf", "LiberationSerif-Regular.ttf", "NotoSerif-Regular.ttf"): + for root in ("/usr/share/fonts/truetype/dejavu", "/usr/share/fonts/truetype/liberation", "/usr/share/fonts/truetype/noto"): + if os.path.exists(os.path.join(root, name)): + font = ImageFont.truetype(os.path.join(root, name), int(11 * dpi / 72)) + break + if font: + break + if font is None: + font = ImageFont.load_default(size=int(11 * dpi / 72)) + y = 20 + for p in paras: + words, line, first = p.split(), "", True + for w in words: + trial = (line + " " + w).strip() + if d.textlength(trial, font=font) > pw - 20 - (60 if first else 0): + d.text((10 + (60 if first else 0), y), line, font=font, fill=0) + y += int(font.size * 1.3) + line, first = w, False + else: + line = trial + d.text((10 + (60 if first else 0), y), line, font=font, fill=0) + y += int(font.size * 1.3) + img.convert("1").save(path, dpi=(dpi, dpi)) + return pw / dpi * 72, ph / dpi * 72 + + +def build(out): + c = canvas.Canvas(out, pagesize=letter) + c.setTitle(TITLE) + c.setAuthor(AUTHOR) + # page 1 title page + c.setFont("Times-Bold", 26) + c.drawCentredString(W / 2, H - 260, TITLE.upper()) + c.setFont("Times-Roman", 14) + c.drawCentredString(W / 2, H - 310, "By " + AUTHOR) + c.showPage() + # page 2 copyright + c.setFont("Times-Roman", 11) + c.drawString(LEFT, H - 200, "First published in 1925.") + c.drawString(LEFT, H - 218, "This is a synthetic book made by code to test an audiobook pipeline.") + c.showPage() + # page 3 contents + c.setFont("Times-Bold", 14) + c.drawString(LEFT, H - 120, "CONTENTS") + c.setFont("Times-Roman", 12) + c.drawString(LEFT, H - 160, "I. The Quiet Harbor " + "." * 40 + " 1") + c.drawString(LEFT, H - 180, "II. The Long Road " + "." * 42 + " 3") + c.showPage() + # chapter 1, book pages 1-2 (pdf pages 4-5) + page_no = 1 + for chapter_no, name, paras in ((1, "THE QUIET HARBOR", CH1), (2, "THE LONG ROAD", CH2)): + first, left = True, list(paras) + while left: + if first: + c.setFont("Times-Bold", 14) + c.drawString(LEFT, TOP, "CHAPTER " + ("ONE" if chapter_no == 1 else "TWO")) + c.drawString(LEFT, TOP - 20, name) + y = TOP - 60 + else: + running_header(c, page_no) + y = TOP - 20 + if chapter_no == 2 and not first: + break # the rest of chapter 2 is the scanned page + y, left = draw_paragraphs(c, left[:3] if chapter_no == 1 and first else left[:3] if chapter_no == 2 else left, y) + if chapter_no == 1 and first: + # keep paragraphs for the next page + left = left + CH1[3:] + c.setFont("Times-Roman", 10) + c.drawCentredString(W / 2, 50, str(page_no)) + c.showPage() + page_no += 1 + first = False + if chapter_no == 1 and not left: + break + if chapter_no == 1 and len(left) <= len(CH1) - 3 and not first: + pass + if chapter_no == 2: + # scanned page: only a picture of the remaining paragraphs + with tempfile.TemporaryDirectory() as tmp: + png = os.path.join(tmp, "scan.png") + w, h = scanned_page_image(CH2[3:], png) + c.drawImage(ImageReader(png), LEFT, 90, width=w, height=h) + c.showPage() + c.save() + + +if __name__ == "__main__": + build(sys.argv[1] if len(sys.argv) > 1 else "pipeline-smoke-test-book.pdf")