From b72c0464ef7e21f118a12d13ade58243892265f4 Mon Sep 17 00:00:00 2001 From: Audiobook Producer Date: Sun, 18 Oct 2026 13:25:01 -0500 Subject: [PATCH] Metadata never fails a drop: deterministic title/author detection (metadata, first-page text, OCR of scanned first pages, file name), sources recorded in book.json; decrypt scratch copy before OCR --- _pipeline/ingest.py | 47 ++--- _pipeline/meta.py | 348 +++++++++++++++++++++++++++++++++++ _pipeline/process.py | 46 ++--- _pipeline/settings.yaml | 8 + _pipeline/tests/helpers.py | 26 +++ _pipeline/tests/test_meta.py | 169 +++++++++++++++++ 6 files changed, 583 insertions(+), 61 deletions(-) create mode 100644 _pipeline/meta.py create mode 100644 _pipeline/tests/helpers.py create mode 100644 _pipeline/tests/test_meta.py diff --git a/_pipeline/ingest.py b/_pipeline/ingest.py index 852ac3a..577e736 100644 --- a/_pipeline/ingest.py +++ b/_pipeline/ingest.py @@ -19,21 +19,6 @@ def sha256(path): 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() @@ -44,18 +29,13 @@ def book_folder(cfg, author, title): return os.path.join(settings.resolve(cfg, "books_dir"), author_slug(author), bookio.slug(title)) -def detect(pdf, title=None, author=None): - """Title and author from the PDF itself (title: PDF metadata; author: a 'By ' line on the first pages).""" - author = author or find_author([u["text"] for u in pdftext.units(pdf, first_pages=6)]) - title = title or find_title(pdf) - return title, author - - -def run(cfg, pdf, title=None, author=None): - """Returns the book folder path.""" - title, author = detect(pdf, title, author) - if not author or not title: - raise SystemExit("could not find title/author in the PDF; pass --title and --author") +def run(cfg, pdf, title=None, author=None, meta=None, source_url=None): + """Returns the book folder path. Title and author come from `meta` (see meta.resolve) or, if none is given, from + meta.resolve here. Missing metadata never stops ingest: it falls back and logs a warning.""" + import meta as meta_mod + if meta is None: + meta = meta_mod.resolve(cfg, pdf, title, author) + title, author = meta["title"], meta["author"] book_dir = book_folder(cfg, author, title) os.makedirs(book_dir, exist_ok=True) copy = os.path.join(book_dir, os.path.basename(pdf)) @@ -70,17 +50,21 @@ def run(cfg, pdf, title=None, author=None): book.update({ "title": title, "author": author, + "metadata_sources": {"title": meta["title_source"], "author": meta["author_source"]}, "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"]], }) + if source_url: + book["source_url"] = source_url book.setdefault("coauthors", []) book.setdefault("publisher", "") book.setdefault("edition", "") book.setdefault("series", "") book.setdefault("summaries", {}) bookio.write_book(book_dir, book) + bookio.write_warnings(book_dir, "ingest-metadata", meta["warnings"]) warnings = ocr_if_needed(cfg, book_dir, book) bookio.write_warnings(book_dir, "ingest", warnings) return book_dir @@ -109,7 +93,14 @@ def ocr_if_needed(cfg, book_dir, book): # behind a symlink outside its permitted paths, e.g. in a CI workspace. Measured in a job container (2026-10-18). src = os.path.join(out_dir, "in.pdf") shutil.copyfile(os.path.join(book_dir, book["source_pdf"]), src) - r = subprocess.run(["ocrmypdf", "--skip-text", "-l", cfg["ocr"]["language"], src, out], capture_output=True, text=True) + if "yes" in (re.search(r"^Encrypted:\s+(\w+)", subprocess.run(["pdfinfo", src], capture_output=True, text=True).stdout, re.M) or [0, ""])[1]: + # owner-password-only PDFs (copy/print restrictions) cannot be OCRed as they are: ocrmypdf refuses them. Decrypt the scratch copy only. + dec = os.path.join(out_dir, "in-decrypted.pdf") + q = subprocess.run(["qpdf", "--decrypt", src, dec], capture_output=True, text=True) if shutil.which("qpdf") else None + if q is None or q.returncode not in (0, 3) or not os.path.exists(dec): + return [("ocr-failed", where, "the PDF is encrypted and could not be decrypted for OCR: %s" % (q.stderr.strip()[:300] if q else "qpdf is not installed"))] + src = dec + r = subprocess.run(["ocrmypdf", "--skip-text", "--output-type", "pdf", "-l", cfg["ocr"]["language"], src, out], capture_output=True, text=True) if r.returncode != 0: print("ocrmypdf failed, exit %d. stdout: %r stderr: %r" % (r.returncode, r.stdout[-1500:], r.stderr[-3000:])) return [("ocr-failed", where, "ocrmypdf exit %d: %s" % (r.returncode, (r.stderr.strip()[:700] + " ... " + r.stderr.strip()[-500:]) if len(r.stderr) > 1250 else r.stderr.strip()))] diff --git a/_pipeline/meta.py b/_pipeline/meta.py new file mode 100644 index 0000000..83074ff --- /dev/null +++ b/_pipeline/meta.py @@ -0,0 +1,348 @@ +"""Title and author for a book. Never fails: whatever cannot be found is replaced and logged. + +Deterministic rules, in this order (the first that gives a value wins). Every value records where it came from. + + 1. Explicit: command-line arguments, then the optional sidecar .json {"title": "...", "author": "..."}. + 2. TITLE : PDF metadata Title ("pdf-metadata"; junk such as "Microsoft Word - x.doc" or "untitled" is ignored) + -> first-page text ("first-page-text", or "first-page-ocr" when the first pages have no text layer) + -> file name ("file-name") -> "Untitled". + AUTHOR : a "By " line on the first pages (text layer, else OCR) [explicit, so it beats metadata] + -> PDF metadata Author (names such as a typesetter or "admin" are ignored) + -> a name line under the title on the first page ("first-page-name-line") + -> "Title - by Author" in the file name -> "Unknown". + 3. The file name is read the same way for a downloaded PDF (the name from the URL), then the name of the .url file. + +Missing author -> "Unknown" (folder "unknown") and a warning. Nothing found at all -> title from the cleaned file name +and author "Unknown". The detection from the PDF itself is cached in the scratch folder, so asking twice costs nothing.""" +import difflib +import hashlib +import json +import os +import re +import shutil +import subprocess +import tempfile + +import bookio +import pdftext +import settings + +UNKNOWN_AUTHOR = "Unknown" +JUNK_AUTHORS = {"admin", "administrator", "user", "owner", "unknown", "scanner", "scan", "bluepen", "pdf", "adobe", "acrobat", + "microsoft", "windows user", "author", "anonymous", "none", "n/a", "ocr", "internet archive"} +NOT_NAME_WORDS = {"the", "of", "and", "a", "an", "in", "on", "for", "to", "with", "from", "press", "society", "publishing", + "publishers", "publication", "publications", "company", "co", "edition", "volume", "vol", "university", + "library", "books", "book", "chapter", "contents", "copyright", "printed", "published", "illustrated", + "manual", "philosophy", "science", "complete", "new", "york", "london", "chicago", "boston", "all", "rights"} +PUBLISHER_WORDS = {"press", "society", "publishing", "publishers", "publication", "company", "edition", "copyright", "printed", + "published", "rights", "reserved", "library", "university", "isbn", "volume"} + +OCR_FUNC = None # tests replace this: f(pdf, first_page, last_page, lang) -> [(page_number, text)] + + +def clean_text(s): + return re.sub(r"\s+", " ", s or "").strip() + + +def junk_title(t): + t = clean_text(t) + if len(re.sub(r"[^A-Za-z]", "", t)) < 3: + return True + low = t.lower() + return bool(re.match(r"^(microsoft word|untitled|document\d*|scan|new document|title|noname)\b", low) or re.search(r"\.(docx?|pdf|tiff?|indd|qxd|rtf|txt)$", low)) + + +def junk_author(a): + a = clean_text(a) + return len(re.sub(r"[^A-Za-z]", "", a)) < 3 or a.lower() in JUNK_AUTHORS or bool(re.search(r"\d{3,}", a)) + + +def display_name(name): + """'YOGI RAMACHARAKA' / 'yogi ramacharaka' -> 'Yogi Ramacharaka'. Mixed case is kept (McDonald).""" + out = [] + for w in clean_text(name).split(" "): + if w.isupper() or w.islower(): + w = "-".join(p[:1].upper() + p[1:].lower() for p in w.split("-")) + out.append(w) + return " ".join(out) + + +def looks_like_name(tokens): + if not 1 <= len(tokens) <= 5: + return False + letters = 0 + for t in tokens: + if t.lower().strip(".,") in NOT_NAME_WORDS: + return False + if not re.match(r"^[A-Z][A-Za-z'’\-]*\.?,?$", t): # Capitalized, ALL CAPS, or an initial + return False + letters += len(re.sub(r"[^A-Za-z]", "", t)) + return letters >= 4 + + +def first_author(text): + """'Glenn Clark and Walter Russell' / 'A. B, C. D' -> first person; None if it does not look like a name.""" + text = re.split(r"\s+(?:and|&|with)\s+|,\s+(?=[A-Z])", clean_text(text).strip(" .,;:"))[0] + tokens = text.split(" ") + return display_name(text) if looks_like_name(tokens) else None + + +def page_lines(text): + """Non-empty lines with a paragraph index: [(paragraph, line)]. OCR noise lines (under 2 letters) are dropped.""" + out, para, blank = [], 0, False + for raw in text.splitlines(): + s = clean_text(raw) + if not s: + blank = True + continue + if len(re.sub(r"[^A-Za-z]", "", s)) < 2: + continue + if blank and out: + para += 1 + blank = False + out.append((para, s)) + return out + + +BY_RE = re.compile(r"^(?:written\s+|translated\s+|compiled\s+|edited\s+)?by[\s:]+(.+)$", re.I) + + +def norm_words(s): + return re.sub(r"[^a-z0-9 ]", "", s.lower().replace("-", " ")).split() + + +def match_hint(line, hint): + """How well a page line matches the title suggested by the file name, 0..1.""" + a, b = " ".join(norm_words(line)), " ".join(norm_words(hint)) + return difflib.SequenceMatcher(None, a, b).ratio() if a and b else 0.0 + + +def read_title_page(pages, hint=None): + """pages: [(page_number, text)]. Returns {'author','author_source','title','page'} from the first page that has a + 'By ' line; else, from the first page with a plausible title block, {'title','page'} plus an 'author' when a + name stands alone right under the title; else {}.""" + for number, text in pages: + lines = page_lines(text) + for i, (para, s) in enumerate(lines): + m = BY_RE.match(s) + if not m or len(s.split()) > 7: + continue + author = first_author(m.group(1)) + if author: + return {"author": author, "author_source": "by-line", "title": title_block(lines[:i], hint)[0], "page": number} + for number, text in pages: + lines = page_lines(text) + title, last = title_block(lines, hint) + if not title: + continue + para = lines[last][0] + mine = [i for i in range(last + 1) if lines[i][0] == para] + tail = first_author(lines[last][1]) if len(mine) >= 2 and title_block(lines, hint)[0] == title else None + if tail and len(mine) >= 2 and " " in lines[last][1]: # "Title\nJane Doe" in one block: the last line is the name + return {"author": tail, "author_source": "name-line", "title": " ".join(lines[i][1] for i in mine[:-1]), "page": number} + for _, nxt in lines[last + 1:last + 3]: # a name on its own line under the title, e.g. "Jane Doe" + a = first_author(nxt) + if a and not any(w in PUBLISHER_WORDS for w in norm_words(nxt)): + return {"author": a, "author_source": "name-line", "title": title, "page": number} + return {"title": title, "page": number} + return {} + + +def title_block(lines, hint=None): + """(title, index of its last line) from the given lines: the line that matches the file-name hint (score 0.7 or more), + else the first paragraph. Publisher lines and lines with a year are skipped.""" + idx = [i for i, (p, s) in enumerate(lines) + if not any(w in PUBLISHER_WORDS for w in norm_words(s)) and not re.search(r"\b1[0-9]{3}\b|\b20[0-9]{2}\b", s)] + if not idx: + return "", -1 + if hint: + score, i = max((match_hint(lines[i][1], hint), i) for i in idx) + if score >= 0.7: + return lines[i][1], i + para = lines[idx[0]][0] + mine = [i for i in idx if lines[i][0] == para] + return " ".join(lines[i][1] for i in mine)[:200], mine[-1] + + +def clean_filename_title(stem): + """'2015.47927.Science-Of-Breath' -> 'Science Of Breath'; 'my_book_final' -> 'my book final' words, ids removed.""" + s = re.sub(r"\([^)]*\)|\[[^\]]*\]", " ", stem) + s = re.sub(r"[_\-.+]+", " ", s) + words = [w for w in s.split() if not re.fullmatch(r"[0-9v]+", w, re.I) and not re.fullmatch(r"[0-9a-f]{8,}", w, re.I)] + words = [w for w in words if not w.lower() in ("pdf", "final", "scan", "scanned", "ocr", "copy")] or words + return " ".join(words) + + +def guess_from_filename(name): + """'The Title - by Glenn Clark (Extra).pdf' / 'Title by Jane Doe.pdf' -> (title, author), else (None, None).""" + stem = os.path.splitext(os.path.basename(name))[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 filename_candidates(names): + """From file names (the PDF's own, then others such as a .url file), the first that carries a usable title: + (title, author_or_None, name). The title is the 'Title - by Author' part, else the cleaned name.""" + for name in names: + if not name: + continue + t, a = guess_from_filename(name) + if t: + return t, a, os.path.basename(name) + cleaned = clean_filename_title(os.path.splitext(os.path.basename(name))[0]) + if len(re.sub(r"[^A-Za-z]", "", cleaned)) >= 4: + return cleaned, None, os.path.basename(name) + return None, None, None + + +# ---------------------------------------------------------------- reading the PDF + +def pdf_info(pdf): + out = subprocess.run(["pdfinfo", pdf], capture_output=True, text=True).stdout + get = lambda k: (re.search(r"^%s:\s+(.+)$" % k, out, re.M) or [None, ""])[1].strip() + return {"title": get("Title"), "author": get("Author")} + + +def ocr_first_pages(pdf, first, last, lang): + """OCR of single pages with pdftoppm + tesseract, for pages that have no text layer. Returns [(page, text)].""" + if OCR_FUNC: + return OCR_FUNC(pdf, first, last, lang) + if shutil.which("tesseract") is None or shutil.which("pdftoppm") is None: + return None + out = [] + with tempfile.TemporaryDirectory() as tmp: + subprocess.run(["pdftoppm", "-r", "200", "-gray", "-f", str(first), "-l", str(last), "-png", pdf, os.path.join(tmp, "p")], + capture_output=True, timeout=600) + for name in sorted(os.listdir(tmp)): + m = re.match(r"p-0*(\d+)\.png$", name) + if not m: + continue + r = subprocess.run(["tesseract", os.path.join(tmp, name), "stdout", "-l", lang, "--psm", "3"], + capture_output=True, text=True, timeout=300) + out.append((int(m.group(1)), r.stdout)) + return out + + +def first_pages_text(cfg, pdf, n, notes): + """[(page, text, source)] for the first n pages: the text layer where there is one, OCR where there is none.""" + pages = [] + try: + units = pdftext.units(pdf, first_pages=n) + except Exception as e: + notes.append("could not read the text layer of the first pages: %s" % e) + units = [] + have = {} + for u in units: + have.setdefault(u["pdf_page"], []).append(u["text"]) + min_chars = cfg["ocr"]["min_chars_per_page"] + blank = [] + for p in range(1, n + 1): + text = "\n".join(have.get(p, [])) + if len(re.sub(r"\s", "", text)) >= min_chars: + pages.append((p, text, "first-page-text")) + elif p in have or not units: + blank.append(p) + return pages, blank + + +def detect_from_pdf(cfg, pdf, hint=None): + """What the PDF itself says. {'title','title_source','author','author_source','notes':[...]} (values may be empty).""" + n = int(cfg.get("meta", {}).get("first_pages", 10)) + notes = [] + try: + n = min(n, len(pdftext.page_geometry(pdf))) + except Exception: + pass + info = pdf_info(pdf) + res = {"title": "", "title_source": "", "author": "", "author_source": "", "notes": notes} + if info["title"] and not junk_title(info["title"]): + res["title"], res["title_source"] = bookio.title_case(clean_text(info["title"])), "pdf-metadata" + info_author = info["author"] if info["author"] and not junk_author(info["author"]) else "" + + pages, blank = first_pages_text(cfg, pdf, n, notes) + page = read_title_page([(p, t) for p, t, _ in pages], hint) + ocr_used = False + if not page.get("author") and blank and cfg["ocr"]["enabled"]: + if not (res["title"] and info_author): # nothing more to learn if metadata already gives both + ocr = ocr_first_pages(pdf, min(blank), max(blank), cfg["ocr"]["language"]) + if ocr is None: + notes.append("the first pages have no text layer and tesseract is not installed, so they could not be read") + else: + got = dict(ocr) + allpages = sorted([(p, t) for p, t, _ in pages] + [(p, got[p]) for p in blank if p in got]) + page = read_title_page(allpages, hint) + ocr_used = True + src = "first-page-ocr" if ocr_used and page.get("page") in blank else "first-page-text" + if page.get("author"): + res["author"] = page["author"] + res["author_source"] = ("%s (%s, pdf page %s)" % (src, page["author_source"], page["page"])) + elif info_author: + res["author"], res["author_source"] = display_name(info_author), "pdf-metadata" + if not res["title"] and page.get("title") and page.get("author"): + res["title"], res["title_source"] = bookio.title_case(page["title"]), "%s (pdf page %s)" % (src, page["page"]) + res["_page_title"] = bookio.title_case(page["title"]) if page.get("title") else "" + res["_page_src"] = "%s (pdf page %s)" % (src, page.get("page")) + return res + + +def cached_detect(cfg, pdf, hint=None): + h = hashlib.sha256() + with open(pdf, "rb") as f: + for block in iter(lambda: f.read(1 << 20), b""): + h.update(block) + h.update((hint or "").encode()) + path = os.path.join(settings.resolve(cfg, "work_dir"), "meta-%s.json" % h.hexdigest()[:20]) + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + return json.load(f) + res = detect_from_pdf(cfg, pdf, hint) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + json.dump(res, f) + return res + + +def resolve(cfg, pdf, title=None, author=None, sidecar=None, name_hints=None): + """The one entry point. Returns {'title','author','title_source','author_source','warnings':[(kind, where, message)], 'log':[lines]}. + Never raises for missing metadata.""" + sidecar = sidecar or {} + names = [os.path.basename(pdf)] + [n for n in (name_hints or []) if n] + ft, fa, fname = filename_candidates(names) + t, a, ts, as_ = title or sidecar.get("title") or "", author or sidecar.get("author") or "", "", "" + if t: + ts = "argument" if title else "sidecar-json" + if a: + as_ = "argument" if author else "sidecar-json" + det = {"notes": []} + if not (t and a): + try: + det = cached_detect(cfg, pdf, ft) + except Exception as e: # an unreadable PDF must not stop metadata; later stages report it properly + det = {"notes": ["could not read the PDF to find title and author: %s" % e]} + warnings = [("metadata-note", os.path.basename(pdf), n) for n in det.get("notes", [])] + if not t: + if det.get("title"): + t, ts = det["title"], det["title_source"] + elif ft: + t, ts = bookio.title_case(ft) if ft.islower() else ft, "file-name (%s)" % fname + if det.get("_page_title") and not det.get("author"): + warnings.append(("metadata-title", os.path.basename(pdf), "a first-page text block %r was not used because no author line confirmed it" % det["_page_title"])) + elif det.get("_page_title"): + t, ts = det["_page_title"], det["_page_src"] + ", unconfirmed" + else: + stem = clean_filename_title(os.path.splitext(os.path.basename(pdf))[0]) + t, ts = (stem or "Untitled"), "file-name (cleaned)" + if ts.startswith("file-name") or "unconfirmed" in ts: + warnings.append(("metadata-title", os.path.basename(pdf), "title %r taken from %s" % (t, ts))) + if not a: + if det.get("author"): + a, as_ = det["author"], det["author_source"] + elif fa: + a, as_ = fa, "file-name (%s)" % fname + else: + a, as_ = UNKNOWN_AUTHOR, "default" + warnings.append(("metadata-author", os.path.basename(pdf), "no author found in the PDF or its file name; using 'Unknown' (folder 'unknown'). Add %s.json with {\"author\": \"...\"} next to the PDF to set it" % os.path.splitext(os.path.basename(pdf))[0])) + log = ["metadata: title %r from %s" % (t, ts), "metadata: author %r from %s" % (a, as_)] + ["metadata: " + n for n in det.get("notes", [])] + return {"title": t, "author": a, "title_source": ts, "author_source": as_, "warnings": warnings, "log": log} diff --git a/_pipeline/process.py b/_pipeline/process.py index 1e0f6d3..baae2da 100644 --- a/_pipeline/process.py +++ b/_pipeline/process.py @@ -27,6 +27,7 @@ import acx_check import assemble import bookio import ingest +import meta import settings import voice @@ -59,12 +60,7 @@ def sidecar(pdf): 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<title>.+?)[\s\-–—]+by\s+(?P<author>[^()]+?)\s*(\(.*\))?\s*$", stem, re.I) - if m: - return m.group("title").strip(" -–—"), m.group("author").strip() - return None, None + return meta.guess_from_filename(pdf) def voice_fingerprint(cfg, script_path): @@ -112,24 +108,9 @@ def step(name): print("\n=== %s ===" % name, flush=True) -def resolve_meta(pdf, title=None, author=None): - """(title, author, note). Order: explicit arguments, the .json next to the PDF, the PDF itself, the file name. - Raises Failed, telling the operator what to add, if none of them works.""" - meta = sidecar(pdf) - title, author = title or meta.get("title"), author or meta.get("author") - try: - title, author = ingest.detect(pdf, title, author) - except Exception as e: - raise Failed("could not read the PDF: %s" % e) - note = None - if not (title and author): - t, a = guess_from_filename(pdf) - title, author = title or t, author or a - note = "title/author not found in the PDF; taking them from the file name: '%s' by '%s'" % (title, author) - if not (title and author): - raise Failed("ingest: could not find title and author in the PDF or its file name. Put a file with the same name and " - '.json next to the PDF: {"title": "...", "author": "..."}') - return title, author, note +def resolve_meta(cfg, pdf, title=None, author=None, name_hints=None): + """Title and author with their sources (see meta.py). Never fails: gaps are filled and logged.""" + return meta.resolve(cfg, pdf, title, author, sidecar(pdf), name_hints) def check_ocr(cfg, book_dir): @@ -158,19 +139,17 @@ def check_ocr(cfg, book_dir): "OCR found no text (blank page or picture only)")]) -def run_text_stages(cfg, pdf, title, author): +def run_text_stages(cfg, pdf, title, author, source_url=None, name_hints=None): """Ingest through license for every chapter. Returns (book_dir, book).""" import cli - title, author, note = resolve_meta(pdf, title, author) + step("metadata") + m = resolve_meta(cfg, pdf, title, author, name_hints) + print("\n".join(m["log"])) step("ingest") - if note: - print(note) try: - book_dir = ingest.run(cfg, pdf, title, author) + book_dir = ingest.run(cfg, pdf, meta=m, source_url=source_url) except SystemExit as e: raise Failed("ingest: %s" % e) - if note: - bookio.write_warnings(book_dir, "ingest-metadata", [("metadata-from-filename", os.path.basename(pdf), "title '%s', author '%s'" % (title, author))]) print("ingest done ->", book_dir) check_ocr(cfg, book_dir) for name in ("structure",): @@ -238,7 +217,8 @@ def run_audio_stages(cfg, book_dir, book, force): return rendered -def process(cfg, pdf, dry_run=False, force=False, title=None, author=None, compare=None, result_file=None, compare_strict=False): +def process(cfg, pdf, dry_run=False, force=False, title=None, author=None, compare=None, result_file=None, compare_strict=False, + source_url=None, name_hints=None): """Returns 0 on success, 1 on failure (message printed).""" pdf = os.path.abspath(pdf) if not os.path.isfile(pdf): @@ -251,7 +231,7 @@ def process(cfg, pdf, dry_run=False, force=False, title=None, author=None, compa rendered = False 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) + book_dir, book = run_text_stages(cfg, pdf, title, author, source_url, name_hints) 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) diff --git a/_pipeline/settings.yaml b/_pipeline/settings.yaml index eafa92d..fa36e8f 100644 --- a/_pipeline/settings.yaml +++ b/_pipeline/settings.yaml @@ -51,6 +51,14 @@ ocr: language: eng min_chars_per_page: 10 # fewer characters than this = no text layer +meta: + first_pages: 10 # title/author are looked for on the first pages (OCR only for pages with no text layer) + +fetch: # a .url/.txt file in _incoming/ holding a link to a PDF + max_mb: 300 # larger downloads are refused + connect_timeout_s: 30 + max_time_s: 900 # a whole download may take this long + script: retail_sample_words: 400 # about 2.5 minutes at 155 words per minute (ACX asks for 1 to 5 minutes) chime_label: Chime diff --git a/_pipeline/tests/helpers.py b/_pipeline/tests/helpers.py new file mode 100644 index 0000000..1a52f6d --- /dev/null +++ b/_pipeline/tests/helpers.py @@ -0,0 +1,26 @@ +"""Small PDF builder for the metadata and fetch tests (needs reportlab, which the CI unit-test step installs).""" +from reportlab.lib.pagesizes import letter +from reportlab.pdfgen import canvas + + +def make_pdf(path, pages, title=None, author=None): + """pages: list of lists of text lines; an empty list = a page with only a picture-like box (no text layer).""" + c = canvas.Canvas(path, pagesize=letter) + if title: + c.setTitle(title) + if author: + c.setAuthor(author) + for lines in pages: + y = 700 + if not lines: + c.rect(100, 100, 300, 400, fill=1) + for ln in lines: + if ln == "": + y -= 30 + continue + c.setFont("Helvetica", 18) + c.drawString(100, y, ln) + y -= 24 + c.showPage() + c.save() + return path diff --git a/_pipeline/tests/test_meta.py b/_pipeline/tests/test_meta.py new file mode 100644 index 0000000..bbfdd58 --- /dev/null +++ b/_pipeline/tests/test_meta.py @@ -0,0 +1,169 @@ +"""Title and author detection. Run: python3 -m unittest discover -s _pipeline/tests -v (from the repo root)""" +import json +import os +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(HERE, "..")) +sys.path.insert(0, HERE) + +import bookio +import ingest +import meta +import settings + +try: + from helpers import make_pdf +except ImportError: + make_pdf = None + +TITLE_PAGE = ["THE HINDU-YOGI", "", "Science of Breath", "", "A Complete Manual of the Oriental Breathing Philosophy", "", "By YOGI RAMACHARAKA", "", "THE YOGI PUBLICATION SOCIETY"] + + +@unittest.skipIf(make_pdf is None, "reportlab not installed") +class MetaTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.d = self.tmp.name + self.cfg = settings.load() + self.cfg["work_dir"] = os.path.join(self.d, "work") + self.cfg["books_dir"] = os.path.join(self.d, "books") + meta.OCR_FUNC = None + + def tearDown(self): + meta.OCR_FUNC = None + self.tmp.cleanup() + + def pdf(self, name, pages, **kw): + return make_pdf(os.path.join(self.d, name), pages, **kw) + + def test_pdf_metadata_title_and_byline_author(self): + p = self.pdf("x.pdf", [["Some Title", "", "By Jane Q. Doe"], ["text"]], title="a quiet book") + m = meta.resolve(self.cfg, p) + self.assertEqual((m["title"], m["author"]), ("A Quiet Book", "Jane Q. Doe")) + self.assertEqual(m["title_source"], "pdf-metadata") + self.assertTrue(m["author_source"].startswith("first-page-text (by-line")) + self.assertEqual(m["warnings"], []) + + def test_junk_metadata_author_is_ignored_when_page_names_the_author(self): + p = self.pdf("x.pdf", [["THE BOOK", "By Glenn Clark"]], title="THE BOOK", author="Bluepen") + self.assertEqual(meta.resolve(self.cfg, p)["author"], "Glenn Clark") + + def test_metadata_author_used_when_page_has_none(self): + p = self.pdf("x.pdf", [["Just a page of prose without any credits at all."]], title="Real Title", author="Mary Shelley") + m = meta.resolve(self.cfg, p) + self.assertEqual((m["author"], m["author_source"]), ("Mary Shelley", "pdf-metadata")) + + def test_page_title_and_uppercase_byline_without_metadata(self): + p = self.pdf("scan0001.pdf", [TITLE_PAGE, ["body"]]) + m = meta.resolve(self.cfg, p) + self.assertEqual(m["author"], "Yogi Ramacharaka") + self.assertTrue(m["title_source"].startswith("first-page-text")) + self.assertEqual(m["title"], "The Hindu-yogi") # first block without a file-name hint + + def test_file_name_hint_picks_the_title_line_on_the_page(self): + p = self.pdf("2015.47927.Science-Of-Breath.pdf", [TITLE_PAGE]) + m = meta.resolve(self.cfg, p) + self.assertEqual((m["title"], m["author"]), ("Science of Breath", "Yogi Ramacharaka")) + + def test_name_line_under_title_without_by(self): + p = self.pdf("x.pdf", [["A Study of Rivers", "Jane Doe", "", "Oxford Press 1921"]]) + m = meta.resolve(self.cfg, p) + self.assertEqual((m["title"], m["author"]), ("A Study of Rivers", "Jane Doe")) + self.assertTrue(m["author_source"].startswith("first-page-text (name-line")) + + def test_scanned_first_pages_are_read_by_ocr(self): + p = self.pdf("2015.47927.Science-Of-Breath.pdf", [[], [], [], [], []]) + calls = [] + + def fake(pdf, first, last, lang): + calls.append((first, last)) + return [(5, "\n".join(TITLE_PAGE))] + [(i, "") for i in range(first, 5)] + meta.OCR_FUNC = fake + m = meta.resolve(self.cfg, p) + self.assertEqual(calls, [(1, 5)]) + self.assertEqual((m["title"], m["author"]), ("Science of Breath", "Yogi Ramacharaka")) + self.assertIn("first-page-ocr", m["author_source"]) + self.assertIn("first-page-ocr", m["title_source"]) + meta.resolve(self.cfg, p) + self.assertEqual(len(calls), 1, "the detection from the PDF is cached") + + def test_nothing_found_uses_file_name_and_unknown_author_with_warning(self): + p = self.pdf("2015.47927.Science-Of-Breath.pdf", [[], []]) + m = meta.resolve(self.cfg, p) # no tesseract here and no OCR_FUNC: pages stay unreadable + self.assertEqual((m["title"], m["author"]), ("Science Of Breath", "Unknown")) + self.assertEqual(m["author_source"], "default") + self.assertTrue(m["title_source"].startswith("file-name")) + kinds = [w[0] for w in m["warnings"]] + self.assertIn("metadata-author", kinds) + self.assertIn("metadata-title", kinds) + self.assertEqual(ingest.author_slug(m["author"]), "unknown") + + def test_title_from_filename_pattern_with_author(self): + p = self.pdf("The Man Who Tapped - by Glenn Clark (Walter Russell).pdf", [[]]) + m = meta.resolve(self.cfg, p) + self.assertEqual((m["title"], m["author"]), ("The Man Who Tapped", "Glenn Clark")) + self.assertEqual(m["author_source"], "file-name (The Man Who Tapped - by Glenn Clark (Walter Russell).pdf)") + + def test_title_found_author_missing_gives_unknown_and_warning(self): + p = self.pdf("x.pdf", [["body text only"]], title="Only A Title") + m = meta.resolve(self.cfg, p) + self.assertEqual((m["title"], m["author"]), ("Only a Title", "Unknown")) + self.assertIn("metadata-author", [w[0] for w in m["warnings"]]) + + def test_sidecar_and_arguments_win(self): + p = self.pdf("x.pdf", [["T", "By Someone Else"]], title="Other") + m = meta.resolve(self.cfg, p, sidecar={"title": "Side Title", "author": "Side Author"}) + self.assertEqual((m["title"], m["author"], m["title_source"], m["author_source"]), ("Side Title", "Side Author", "sidecar-json", "sidecar-json")) + m = meta.resolve(self.cfg, p, title="Arg Title", sidecar={"title": "Side Title", "author": "Side Author"}) + self.assertEqual((m["title"], m["title_source"], m["author"]), ("Arg Title", "argument", "Side Author")) + + def test_url_file_name_hint_is_used_when_the_pdf_name_is_only_an_id(self): + p = self.pdf("2401.01234.pdf", [[]]) + m = meta.resolve(self.cfg, p, name_hints=["Walden Pond Notes.url"]) + self.assertEqual(m["title"], "Walden Pond Notes") + self.assertIn("Walden Pond Notes.url", m["title_source"]) + + def test_never_raises_on_a_file_that_is_not_a_pdf(self): + p = os.path.join(self.d, "My Odd_Scan 7.pdf") + open(p, "w").write("this is not a pdf") + m = meta.resolve(self.cfg, p) + self.assertEqual(m["author"], "Unknown") + self.assertTrue(m["title"]) + + def test_ingest_records_sources_and_warnings_in_the_book_folder(self): + p = self.pdf("2015.47927.Science-Of-Breath.pdf", [[], []]) + self.cfg["ocr"]["enabled"] = False + d = ingest.run(self.cfg, p, source_url="https://example.org/a.pdf") + book = bookio.read_book(d) + self.assertEqual(d, os.path.join(self.cfg["books_dir"], "unknown", "science-of-breath")) + self.assertEqual(book["author"], "Unknown") + self.assertEqual(book["metadata_sources"]["author"], "default") + self.assertTrue(book["metadata_sources"]["title"].startswith("file-name")) + self.assertEqual(book["source_url"], "https://example.org/a.pdf") + log = open(os.path.join(d, "warnings.log")).read() + self.assertIn("ingest-metadata | metadata-author |", log) + + def test_clark_style_page_gives_the_same_answer_as_before(self): + p = self.pdf("c.pdf", [["THE MAN WHO TAPPED THE SECRETS OF THE UNIVERSE"], ["By Glenn Clark"]], title="THE MAN WHO TAPPED THE SECRETS OF THE UNIVERSE", author="Bluepen") + m = meta.resolve(self.cfg, p) + self.assertEqual((m["title"], m["author"]), ("The Man Who Tapped the Secrets of the Universe", "Glenn Clark")) + + +class PureTests(unittest.TestCase): + def test_cleaning_file_names(self): + self.assertEqual(meta.clean_filename_title("2015.47927.Science-Of-Breath"), "Science Of Breath") + self.assertEqual(meta.clean_filename_title("my_book_final"), "my book") + self.assertEqual(meta.filename_candidates(["2401.01234.pdf", "Deep Work.url"])[0], "Deep Work") + + def test_author_rules(self): + self.assertEqual(meta.first_author("YOGI RAMACHARAKA"), "Yogi Ramacharaka") + self.assertEqual(meta.first_author("Glenn Clark and Walter Russell"), "Glenn Clark") + self.assertIsNone(meta.first_author("the sea and the sky")) + self.assertIsNone(meta.first_author("The Yogi Publication Society")) + + +if __name__ == "__main__": + unittest.main()