bitscuit
a1796e82aa
Pipeline: extract (pdftotext) -> scan (OCR-defect report + fix template) -> clean (noise/header removal, hyphenation, exact-string/regex OCR repairs, chapter-heading detection) -> voices (sample + comparison) -> chunk (sentence-aware) -> synth (resumable edge-tts) -> stitch (ffmpeg concat + RMS verify) -> chapters (ID3v2.4 CHAP). Config-driven per book; validated end-to-end on a clean digital PDF and on a 546-page scanned book (byte-identical clean output to the reference run, 95 auto-detected chapters).
208 lines
7.9 KiB
Python
208 lines
7.9 KiB
Python
"""Clean raw pdftotext output into TTS-ready text.
|
|
|
|
The pass order mirrors the pipeline proven on a 546-page scanned book:
|
|
|
|
1. exact-string raw fixes (typically U+FFFD glyph repairs)
|
|
2. keep window [include.start, include.end)
|
|
3. drop blocks (e.g. table of contents)
|
|
4. form feeds -> newlines
|
|
5. join hyphenated line breaks
|
|
6. line filter (noise lines, page headers, custom drop patterns)
|
|
7. join lines into one string
|
|
8. strip inline patterns (plate refs, running headers wedged mid-sentence)
|
|
9. regex fixes (order preserved from config)
|
|
10. exact-string fixes (longest first)
|
|
|
|
All patterns come from an optional JSON config; without one you still get
|
|
the generic noise/header removal and hyphen joining, which is enough for
|
|
digitally-born PDFs.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
FFFD = "\ufffd"
|
|
|
|
# Lines that are pure layout noise: lone characters, dots, page numbers,
|
|
# roman numerals, bullet/dash lines.
|
|
NOISE_RE = re.compile(
|
|
r"^(\s*[\u2022\u00b7.]+|\s*[0-9]{1,3}\s*|\s*[IVXLCDM]{1,6}\s*"
|
|
r"|\s*I\s*|\s*\u00a7\s*|\s*[\-\u2013]\s*)$"
|
|
)
|
|
|
|
# Headings are expected to be short standalone lines; longer "headings" are
|
|
# usually corrupted body text.
|
|
MAX_HEADING_LEN = 90
|
|
|
|
|
|
def unhyphenate(t: str) -> str:
|
|
t = t.replace("\u00ad\n", "") # soft hyphen
|
|
t = t.replace("\u2010\n", "") # hyphen char
|
|
t = re.sub(r"(\w)-\n(\w)", r"\1\2", t)
|
|
t = re.sub(r"(\w)\\\-\n(\w)", r"\1\2", t) # OCR backslash-hyphen
|
|
t = re.sub(r"(\w)\\ \\\\\n", r"\1", t) # OCR backslash-space-backslash
|
|
return t
|
|
|
|
|
|
def _find_line(lines, pattern, min_line=0):
|
|
rx = re.compile(pattern)
|
|
for i in range(min_line, len(lines)):
|
|
# match on the stripped line: pdftotext prefixes the first line of
|
|
# each page with a form feed, which would defeat ^...$ anchors
|
|
if rx.search(lines[i].strip()):
|
|
return i
|
|
return None
|
|
|
|
|
|
def clean(raw_path: str, cfg_path: str = None, out_path: str = "clean.txt") -> dict:
|
|
cfg = {}
|
|
if cfg_path:
|
|
with open(cfg_path, encoding="utf-8") as f:
|
|
cfg = json.load(f)
|
|
|
|
text = open(raw_path, encoding="utf-8").read()
|
|
|
|
# 1. raw-level exact fixes (applied before any structure is touched)
|
|
rawfix = cfg.get("raw_fixes", {})
|
|
for k in sorted(rawfix, key=len, reverse=True):
|
|
if k in text:
|
|
text = text.replace(k, rawfix[k])
|
|
|
|
lines = text.split("\n")
|
|
|
|
# 3. drop blocks FIRST (a TOC entry can look like the include.start
|
|
# anchor; the old LOY run removed the TOC before cutting front matter)
|
|
for block in cfg.get("drop_blocks", []):
|
|
s = _find_line(lines, block["start"], block.get("min_line_start", 0))
|
|
if s is None:
|
|
continue
|
|
e = len(lines)
|
|
if block.get("end"):
|
|
followed = block.get("end_followed_by")
|
|
for i in range(s + 1, len(lines)):
|
|
if re.search(block["end"], lines[i]):
|
|
if followed and not any(
|
|
lines[k].strip() in followed
|
|
for k in range(i + 1, min(i + 4, len(lines)))
|
|
):
|
|
continue # keep looking
|
|
e = i
|
|
break
|
|
lines = lines[:s] + lines[e:]
|
|
|
|
# 2. include window
|
|
inc = cfg.get("include")
|
|
if inc and inc.get("start"):
|
|
s = _find_line(lines, inc["start"], inc.get("min_line_start", 0))
|
|
if s is None:
|
|
raise SystemExit(f"include.start pattern not found: {inc['start']!r}")
|
|
e = _find_line(lines, inc["end"], max(inc.get("min_line_end", 0), s + 1))
|
|
if e is None and inc.get("end_fallback"):
|
|
e = _find_line(lines, inc["end_fallback"], max(inc.get("min_line_end", 0), s + 1))
|
|
if e is None:
|
|
e = len(lines)
|
|
lines = lines[s:e]
|
|
|
|
# 4. form feeds
|
|
text = "\n".join(lines).replace("\f", "\n")
|
|
|
|
# 5. hyphenated line breaks
|
|
text = unhyphenate(text)
|
|
|
|
# 6. line filter (and record chapter heading positions)
|
|
hdr_pats = [re.compile(p) for p in cfg.get("page_header_patterns", [])]
|
|
drop_pats = [re.compile(p) for p in cfg.get("drop_line_patterns", [])]
|
|
heading_re = re.compile(cfg["heading_pattern"]) if cfg.get("heading_pattern") else None
|
|
# optional guard: a heading line only counts if it is followed (within a
|
|
# few lines) by this pattern. Filters running headers that share the
|
|
# heading's shape ("Appendix I" on every page of the appendix).
|
|
heading_next_re = re.compile(cfg["heading_next_pattern"]) \
|
|
if cfg.get("heading_next_pattern") else None
|
|
heading_next_skip = cfg.get("heading_next_skip", 3)
|
|
# lines matching these are never headings (but stay in the text):
|
|
# numbered instruction sentences that look like "N. Name" headings
|
|
heading_excl = [re.compile(p) for p in cfg.get("heading_exclude_patterns", [])]
|
|
heading_count_max = cfg.get("heading_count_max", 150)
|
|
|
|
out_lines = []
|
|
pos = 0 # char offset of the current line in the joined text
|
|
chapters = [] # (char_offset_in_clean_text, title)
|
|
lines2 = text.split("\n")
|
|
for i2, line in enumerate(lines2):
|
|
s = line.strip()
|
|
if not s:
|
|
continue
|
|
if NOISE_RE.match(s):
|
|
continue
|
|
if any(p.match(s) for p in hdr_pats):
|
|
continue
|
|
if any(p.match(s) for p in drop_pats):
|
|
continue
|
|
# (heading patterns are written to accept only proper-case or
|
|
# numbered lines, so no extra case check is needed here)
|
|
if (heading_re and len(chapters) < heading_count_max
|
|
and len(s) <= MAX_HEADING_LEN
|
|
and heading_re.match(s)
|
|
and not any(p.search(s) for p in heading_excl)):
|
|
if heading_next_re:
|
|
ok = any(
|
|
heading_next_re.search(lines2[k].strip())
|
|
for k in range(i2 + 1, min(i2 + 1 + heading_next_skip, len(lines2)))
|
|
)
|
|
if not ok:
|
|
continue
|
|
title = s.rstrip(" .:;-")
|
|
chapters.append((pos, title))
|
|
out_lines.append(s)
|
|
pos += len(s) + 1
|
|
|
|
joined = re.sub(r" +", " ", " ".join(out_lines))
|
|
|
|
# 8. inline strips (applied on the joined text)
|
|
for p in cfg.get("strip_patterns", []):
|
|
joined = re.sub(p, " ", joined)
|
|
|
|
# 9. regex fixes, in config order (runs BEFORE exact-string fixes,
|
|
# matching the reference pipeline: regexes may leave tokens that the
|
|
# exact map then clobbers correctly)
|
|
for pat, rep in cfg.get("regex_fixes", []):
|
|
joined = re.sub(pat, rep, joined)
|
|
|
|
# 10. exact-string fixes, longest first
|
|
for k in sorted(cfg.get("fixes", {}), key=len, reverse=True):
|
|
if k in joined:
|
|
joined = joined.replace(k, cfg["fixes"][k])
|
|
|
|
# 11. word-boundary regex fixes (anchored, last, like the reference run)
|
|
for pat, rep in cfg.get("word_fixes", []):
|
|
joined = re.sub(pat, rep, joined)
|
|
|
|
joined = re.sub(r" +", " ", joined).strip()
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
f.write(joined)
|
|
|
|
# Re-anchor chapter positions in the final text: the string fixes above
|
|
# can shift offsets, so re-find each title near its recorded position.
|
|
# Also drop duplicate titles (running headers that share a heading's
|
|
# shape would otherwise create a chapter on every page).
|
|
seen_titles = set()
|
|
final_chapters = []
|
|
for pos, title in chapters:
|
|
if title in seen_titles:
|
|
continue
|
|
seen_titles.add(title)
|
|
i = joined.find(title, max(0, pos - 600))
|
|
final_chapters.append([i if i >= 0 else pos, title])
|
|
chapters_path = os.path.splitext(out_path)[0] + ".chapters.json"
|
|
with open(chapters_path, "w", encoding="utf-8") as f:
|
|
json.dump(final_chapters, f, indent=1, ensure_ascii=False)
|
|
|
|
return {
|
|
"out": out_path,
|
|
"chars": len(joined),
|
|
"words": len(joined.split()),
|
|
"fffd_left": joined.count(FFFD),
|
|
"chapters": len(final_chapters),
|
|
"chapters_path": chapters_path,
|
|
}
|