diff --git a/_pipeline/ingest.py b/_pipeline/ingest.py index 577e736..6aecf05 100644 --- a/_pipeline/ingest.py +++ b/_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:])) diff --git a/_pipeline/meta.py b/_pipeline/meta.py index 2c4f059..43f4605 100644 --- a/_pipeline/meta.py +++ b/_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: diff --git a/_pipeline/structure.py b/_pipeline/structure.py index deee590..0c99f5a 100644 --- a/_pipeline/structure.py +++ b/_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 ' 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) diff --git a/_pipeline/tests/test_structure.py b/_pipeline/tests/test_structure.py index 17f5c76..6a40e10 100644 --- a/_pipeline/tests/test_structure.py +++ b/_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")]), [])