Structure: fuzzy CHAPTER keyword for OCR, cap run-on titles, prefer contents-page title when it has fewer unknown words; metadata: multi-resolution OCR of title pages, OCR name correction from repeated words; ingest: qpdf optional

_pipeline/ingest.py

@@ -97,9 +97,10 @@ def ocr_if_needed(cfg, book_dir, book):
         # 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
+        if q is not None:
+            if 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])]
+            src = dec                    # without qpdf, ocrmypdf is tried on the copy as it is
     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:]))

_pipeline/meta.py

@@ -92,7 +92,8 @@ def looks_like_name(tokens):
 
 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]
+    text = re.sub(r"[^A-Za-z.'’\-,& ]", "", text)             # OCR specks after the name (RAMACHARAKA°, 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
 
@@ -215,15 +216,15 @@ def pdf_info(pdf):
     return {"title": get("Title"), "author": get("Author")}
 
 
-def ocr_first_pages(pdf, first, last, lang):
+def ocr_first_pages(pdf, first, last, lang, dpi=300):
     """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)
+        return OCR_FUNC(pdf, first, last, lang)         # tests: one pass only
     if shutil.which("tesseract") is None or shutil.which("pdftoppm") is None:
         return None
     out = []
     with tempfile.TemporaryDirectory() as tmp:
-        subprocess.run(["pdftoppm", "-r", "300", "-gray", "-f", str(first), "-l", str(last), "-png", pdf, os.path.join(tmp, "p")],
+        subprocess.run(["pdftoppm", "-r", str(dpi), "-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)
@@ -279,17 +280,23 @@ def detect_from_pdf(cfg, pdf, hint=None):
     allpages = [(p, t) for p, t, _ in pages]
     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:
+            for dpi in ((300,) if OCR_FUNC else (300, 200, 600)):      # display type reads differently at each size: first pass that finds a 'By' line wins
+                ocr = ocr_first_pages(pdf, min(blank), max(blank), cfg["ocr"]["language"], dpi)
+                if ocr is None:
+                    notes.append("the first pages have no text layer and tesseract is not installed, so they could not be read")
+                    break
                 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
+                if page.get("author_source") == "by-line":
+                    notes.append("first pages read by OCR at %d dpi" % dpi)
+                    break
     src = "first-page-ocr" if ocr_used and page.get("page") in blank else "first-page-text"
     if page.get("author"):
         if src == "first-page-ocr":
+            page["author"], fixes = correct_ocr_name(page["author"], [t for _, t in allpages])
+            notes.extend(fixes)
             notes.append("author %r was read by OCR from pdf page %s; check it. A file named like the PDF with .json (author field) overrides it" % (page["author"], page["page"]))
         res["author"] = page["author"]
         res["author_source"] = ("%s (%s, pdf page %s)" % (src, page["author_source"], page["page"]))
@@ -305,6 +312,29 @@ def detect_from_pdf(cfg, pdf, hint=None):
     return res
 
 
+def correct_ocr_name(name, texts):
+    """OCR can misread letters of display type ('Yoct' for 'Yogi'). A name word of 4+ letters that appears only once on the first
+    pages (in the author line) but has a same-length twin that appears 2+ times there, sharing the first letter and at least half
+    of the letters in the same positions, is replaced by that twin. Returns (name, [what was changed])."""
+    words = {}
+    for t in texts:
+        for w in re.findall(r"[A-Za-z][A-Za-z'’]{2,}", t):
+            words[w.lower()] = words.get(w.lower(), 0) + 1
+    out, notes = [], []
+    for tok in name.split(" "):
+        low = re.sub(r"[^a-z]", "", tok.lower())
+        twins = [(n, w) for w, n in words.items() if len(low) >= 4 and words.get(low, 0) <= 1 and w != low and len(w) == len(low) and w[0] == low[0]
+                 and n >= 2 and sum(a == b for a, b in zip(w, low)) * 2 >= len(low)]
+        if twins:
+            n, w = max(twins)
+            fixed = w.upper() if tok.isupper() else w[:1].upper() + w[1:]
+            notes.append("OCR read the name word %r once; the first pages show %r %d times, so %r was used" % (tok, w, n, fixed))
+            out.append(fixed)
+        else:
+            out.append(tok)
+    return " ".join(out), notes
+
+
 def cached_detect(cfg, pdf, hint=None):
     h = hashlib.sha256()
     with open(pdf, "rb") as f:

_pipeline/structure.py

@@ -1,4 +1,5 @@
 """Stage 3, structure: find chapter boundaries from the text layer. Audible style file list is made by the script stage."""
+import difflib
 import re
 
 import bookio
@@ -7,7 +8,7 @@ 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\b[\s.:]*(.*?)[\s.,:;]*$")
+HEADING_RE = re.compile(r"^\s*([A-Z]{6,9}|CHAPTER)\b[\s.:]*(.*?)[\s.,:;]*$")
 
 
 def chapter_number(token):
@@ -23,6 +24,12 @@ def chapter_number(token):
     return None
 
 
+def is_chapter_word(word):
+    """CHAPTER, or what OCR makes of it (CUAPTER, CIEAPTER, OUAPTER): upper-case letters, 6 to 9 of them, 70% like CHAPTER."""
+    return word == "CHAPTER" or (re.fullmatch(r"[A-Z]{6,9}", word) is not None and difflib.SequenceMatcher(None, word, "CHAPTER").ratio() >= 0.7
+                                 and word not in ("CHAPTERS", "CAPTAIN", "CHARTER", "CHAPLET"))
+
+
 def heading_number(line, expected):
     """(number, guessed) for a 'CHAPTER <number>' line, else None. The number may be a roman numeral, digits or a word.
     OCR often damages the numeral ('I I r', 'X V', 'X f I'): a short unreadable token after an upper-case CHAPTER is
@@ -30,7 +37,9 @@ def heading_number(line, expected):
     m = HEADING_RE.match(line)
     if not m:
         return None
-    rest = m.group(1).strip()
+    if not is_chapter_word(m.group(1)):
+        return None
+    rest = m.group(2).strip()
     if re.fullmatch(r"[A-Z]+|\d+", rest):
         n = chapter_number(rest)
         if n is not None:
@@ -68,6 +77,12 @@ def find_headings(units):
             while k < len(lines) and lines[k].strip():
                 title.append(lines[k].strip())
                 k += 1
+            if len(" ".join(title)) > 100:        # no blank line after the title: keep the leading upper-case lines only
+                keep = 1
+                while keep < len(title) and keep < 3 and pageparts._mostly_upper(title[keep]):
+                    keep += 1
+                k = k - (len(title) - keep)
+                title = title[:keep]
             h = {"unit": ui, "line": li, "title_end": k, "number": n, "title": bookio.title_case(" ".join(title))}
             if guessed:
                 h["guessed"] = True
@@ -100,6 +115,29 @@ def toc_chapters(units, before_unit):
     return out
 
 
+def unknown_words(text):
+    """How many words of a title the spelling dictionary does not know (used to choose between two readings of one title)."""
+    import cleanup
+    return sum(1 for w in re.findall(r"[A-Za-z]+", text) if len(w) > 2 and not cleanup.known(w))
+
+
+def toc_chapter_lines(units, before_unit):
+    """Contents in the form 'CHAPTER V.' on one line and 'Title ........ 20' on the next: {number: (title, page)}."""
+    lines = []
+    for u in units[:before_unit]:
+        lines += [ln for ln in u["text"].split("\n") if ln.strip()]
+    out, last = {}, 0
+    for i, ln in enumerate(lines[:-1]):
+        got = heading_number(ln, last + 1)
+        if got is None or not re.search(r"\.{3,}", lines[i + 1]):
+            continue
+        title = re.split(r"\s*\.{3,}", lines[i + 1].strip())[0].strip(" .,")
+        if len(re.sub(r"[^A-Za-z]", "", title)) >= 4 and got[0] not in out:
+            out[got[0]] = (bookio.title_case(re.sub(r"\s+", " ", title)), None)
+            last = got[0]
+    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."""
@@ -140,6 +178,8 @@ def run(book_dir, book, pdf):
             warnings.append(("heading-replaced", "chapter %d" % h["number"], "an earlier CHAPTER line with the same or a higher number (a contents entry or misread) was dropped in favour of the heading on pdf page %d" % units[h["unit"]]["pdf_page"]))
     pages, printed = book_pages(units, book["title"])
     toc = {n: (t, p) for n, t, p in toc_chapters(units, heads[0]["unit"])}
+    for n, tp in toc_chapter_lines(units, heads[0]["unit"]).items():
+        toc.setdefault(n, tp)
     old = {c["number"]: c for c in book.get("chapters", [])}
     chapters = []
     for k, h in enumerate(heads):
@@ -169,7 +209,11 @@ def run(book_dir, book, pdf):
             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:
+            sim = pageparts.similar(t_title, h["title"])
+            if 0.5 <= sim < 1 and t_page is None and not h.get("synthetic") and unknown_words(t_title) < unknown_words(h["title"]):
+                warnings.append(("title-from-contents", "chapter %d" % h["number"], "heading read '%s' (OCR damage?); the contents page says '%s' with fewer unknown words, which is used" % (h["title"], t_title)))
+                ch["title"] = t_title
+            elif sim < 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)

_pipeline/tests/test_structure.py

@@ -48,6 +48,30 @@ class HeadingTests(unittest.TestCase):
         self.assertTrue(got[0]["synthetic"])
         self.assertEqual((got[0]["unit"], got[0]["number"]), (2, 1))
 
+    def test_ocr_damaged_chapter_word_is_recognised(self):
+        units = [unit("CHAPTER I.\nOne\n\n" + BODY), unit("OUAPTER                 II.\nTwo\n\n" + BODY),
+                 unit("CUAPTER TX.\nNine?\n\n" + BODY), unit("CIEAPTER NEL,\nTwelve?\n\n" + BODY)]
+        got = structure.find_headings(units)
+        self.assertEqual([h["number"] for h in got], [1, 2, 3, 4])           # unreadable numerals count up in sequence
+        self.assertFalse(structure.is_chapter_word("CHAPTERS"))
+        self.assertFalse(structure.is_chapter_word("CAPTAIN"))
+        self.assertFalse(structure.is_chapter_word("SCIENCE"))
+
+    def test_long_title_without_blank_line_keeps_only_the_upper_case_lines(self):
+        body = "the Yogis not only bring about desired mental qualities and properties by will-power coupled with rhythmic breathing, but they also develop"
+        units = [unit("CHAPTER XVI.\nYOGI SPIRITUAL BREATHING.\n" + body + "\n" + body + "\n\n" + BODY)]
+        h = structure.find_headings(units)[0]
+        self.assertEqual(h["title"], "Yogi Spiritual Breathing.")
+        self.assertEqual(h["title_end"], 2)
+
+    def test_contents_in_chapter_line_form(self):
+        units = [unit("CONTENTS\nCHAPTER I.\nBreath Is Life........ 8\nCHAPTER II.\nThe Four Methods of Respiration ..... 27\n"), unit("CHAPTER I.\nX\n\n" + BODY)]
+        toc = structure.toc_chapter_lines(units, 1)
+        self.assertEqual(toc[2][0], "The Four Methods of Respiration")
+        self.assertEqual(toc[1][0], "Breath Is Life")
+        self.assertEqual(structure.unknown_words("The Four Metiiods of Respiration"), 1)
+        self.assertEqual(structure.unknown_words("The Four Methods of Respiration"), 0)
+
     def test_nothing_readable_gives_no_headings(self):
         self.assertEqual(structure.find_headings([unit(""), unit("x")]), [])