Repository Public Research preview
1"""Book parser (plain text, HTML, EPUB) for the "books" stratum: Project Gutenberg plain text, OpenStax/CNX2text extracted from PDFs, and HTML/XHTML book files. NCBI Bookshelf BITS XML is parsed by ``htr.parse.jats``.34Plain text pipeline (``raw_format == "txt"``):5 1. strip the Project Gutenberg header/footer (``*** START OF ... ***`` / ``*** END OF ...``) when present;6 2. split into blank-line separated blocks and classify each block as a heading (``CHAPTER X``, roman7 numerals, ``PREFACE``, ``INTRODUCTION``, short ALL-CAPS lines), a special section (contents, index,8 notes, appendix, bibliography, attributions...), an illustration, a footnote, an indented/table block,9 or a prose paragraph;10 3. a small state machine drops the front matter (title page, epigraphs) before the first strong heading,11 the table of contents, the index/back matter, footnote blocks and illustrations (each counted in12 ``warnings``) and re-flows hard-wrapped lines of the remaining paragraphs;13 4. paragraphs are emitted exactly as in ``htr.parse.jats`` (``protect()`` per paragraph, ``body_text`` =14 paragraphs joined by ``"\\n\\n"`` with consistent ``char_start``/``char_end``).1516Roles: ``preface``, ``introduction``, ``chapter``. Dates and licences are NOT decided here: they come from the17ingest stub (catalog / front matter / author death year), so the parser returns ``pub_date=None`` and18``license_code="UNKNOWN"`` and ``htr.ingest.corpus.document_from`` uses the stub values with their evidence.1920Front-matter date detection (``detect_frontmatter_year``) lives here because it is a pure text function; the21Gutenberg ingest calls it at acquisition time and records the matched string as evidence.22"""23from __future__ import annotations2425import posixpath26import re27import zipfile28from dataclasses import dataclass, field29from pathlib import Path3031from htr.parse.jats import ParsedDocument, ParsedParagraph, section_role32from htr.parse.protect import protect33from htr.utils.text import normalize_text, word_count3435# --------------------------------------------------------------------------------------------------------------36# Project Gutenberg boilerplate37# --------------------------------------------------------------------------------------------------------------38PG_START_RE = re.compile(r"^[ \t]*\*{3}[ \t]*START OF (?:THE |THIS )?PROJECT GUTENBERG E-?(?:BOOK|TEXT)\b[^\n]*$", re.IGNORECASE | re.MULTILINE)39PG_END_RE = re.compile(r"^[ \t]*\*{3}[ \t]*END OF (?:THE |THIS )?PROJECT GUTENBERG E-?(?:BOOK|TEXT)\b[^\n]*$", re.IGNORECASE | re.MULTILINE)40PG_OLD_START_RE = re.compile(r"^[^\n]*\*END\*THE SMALL PRINT[^\n]*$", re.IGNORECASE | re.MULTILINE)41PG_OLD_END_RE = re.compile(r"^[ \t]*End of (?:the )?Project Gutenberg(?:'s)?\b[^\n]*$", re.IGNORECASE | re.MULTILINE)42PG_HEADER_PD_RE = re.compile(r"for the use of anyone anywhere in the United States", re.IGNORECASE)43PG_HEADER_COPYRIGHTED_RE = re.compile(r"copyrighted (?:Project Gutenberg )?e-?book|this (?:e-?book|e-?text) is copyrighted|is a copyrighted", re.IGNORECASE)444546@dataclass47class GutenbergSplit:48 body: str49 header: str50 footer: str51 start_found: bool52 end_found: bool535455def strip_gutenberg_boilerplate(text: str) -> GutenbergSplit:56 """Return the text between the PG start and end markers (the whole text if a marker is missing)."""57 m = PG_START_RE.search(text) or PG_OLD_START_RE.search(text)58 start = m.end() if m else 059 ends = [e.start() for e in (PG_END_RE.search(text, start), PG_OLD_END_RE.search(text, start)) if e]60 end = min(ends) if ends else len(text)61 return GutenbergSplit(text[start:end], text[:start], text[end:], m is not None, bool(ends))626364def gutenberg_header_status(header: str) -> str:65 """'public_domain_us' when the header carries the standard US public-domain statement, 'copyrighted' when it66 declares a copyrighted eBook, otherwise 'unknown'."""67 if PG_HEADER_COPYRIGHTED_RE.search(header):68 return "copyrighted"69 if PG_HEADER_PD_RE.search(" ".join(header.split())):70 return "public_domain_us"71 return "unknown"727374# --------------------------------------------------------------------------------------------------------------75# Front-matter publication year76# --------------------------------------------------------------------------------------------------------------77_YEAR = r"(?:1[4-9]\d\d|20[01]\d)"78_YEAR_RE = re.compile(rf"(?<!\d){_YEAR}(?!\d)")79_COPYRIGHT_LINE_RE = re.compile(r"\b(?:copyright(?:ed)?|entered according to act of congress)\b|©|\(c\)\s*\d", re.IGNORECASE)80_PUBLISHED_LINE_RE = re.compile(r"\b(?:first published|published|first edition|first printed|first issued|printed|reprinted|edition|impression)\b", re.IGNORECASE)81_PG_ADDED_LINE_RE = re.compile(r"^\s*\d{3,6}\s{2,}\S|gutenberg|release date|posting date|last updated|most recently updated|e-?book #|e-?text #|\bproduced by\b|transcriber", re.IGNORECASE)82_ROMAN_YEAR_RE = re.compile(r"^\s*(?:[A-Z][A-Z.,&:'’\- ]{0,80}?[\s.,:])?(M{1,2}(?:CM|CD|D?C{0,3})(?:XC|XL|L?X{0,3})(?:IX|IV|V?I{0,3}))\.?\s*$")83_IMPRINT_YEAR_RE = re.compile(rf"^\s*(?:[A-Z][A-Za-z.,&:'’\- ]{{0,100}}?[\s.,:])?({_YEAR})\.?\s*$")84_BODY_START_RE = re.compile(r"^\s*(?:contents|table of contents|preface|introduction|chapter\s+(?:[ivxlc]+|\d+|one|first)\b)", re.IGNORECASE | re.MULTILINE)85_ROMAN_VALUES = {"M": 1000, "D": 500, "C": 100, "L": 50, "X": 10, "V": 5, "I": 1}868788def roman_to_int(s: str) -> int | None:89 s = s.upper().strip()90 if not s or any(c not in _ROMAN_VALUES for c in s):91 return None92 total = 093 for i, c in enumerate(s):94 v = _ROMAN_VALUES[c]95 nxt = _ROMAN_VALUES[s[i + 1]] if i + 1 < len(s) else 096 total += -v if v < nxt else v97 return total9899100@dataclass101class FrontMatterDate:102 year: int103 matched: str104 rule: str # copyright | published | imprint | roman105 all_years: list[int] = field(default_factory=list)106107108def detect_frontmatter_year(body: str, max_chars: int = 5000, min_year: int = 1450, max_year: int = 2017) -> FrontMatterDate | None:109 """Find the publication year printed in the front matter of a book (text AFTER the PG start marker).110111 Priority: copyright / "Entered according to Act of Congress" lines → "published/edition/printed" lines →112 publisher imprint line ending with a year (only before the first CONTENTS/PREFACE/CHAPTER heading) →113 roman-numeral imprint year (e.g. ``MDCCCXC``). Lines added by Project Gutenberg (release date, producer114 credits, transcriber notes) are ignored. Several years on the selected line (e.g. ``Copyright, 1898, 1905``)115 → the LATEST one is used: it is an upper bound on the edition reproduced, which is what pre-2018 evidence116 requires. Returns None when nothing plausible is found.117 """118 region = body[:max_chars]119 lines = region.split("\n")120121 def ok(y: int) -> bool:122 return min_year <= y <= max_year123124 def scan(pattern: re.Pattern, rule: str, two_lines: bool) -> FrontMatterDate | None:125 for i, line in enumerate(lines):126 if not pattern.search(line) or _PG_ADDED_LINE_RE.search(line):127 continue128 seg = line + (" " + lines[i + 1] if two_lines and i + 1 < len(lines) and not _PG_ADDED_LINE_RE.search(lines[i + 1]) else "")129 years = [int(y) for y in _YEAR_RE.findall(seg) if ok(int(y))]130 if years:131 return FrontMatterDate(max(years), " ".join(seg.split())[:160], rule, years)132 return None133134 hit = scan(_COPYRIGHT_LINE_RE, "copyright", True) or scan(_PUBLISHED_LINE_RE, "published", False)135 if hit:136 return hit137 m = _BODY_START_RE.search(region)138 pre = region[: m.start()] if m else region139 for line in pre.split("\n"):140 if _PG_ADDED_LINE_RE.search(line) or not line.strip():141 continue142 im = _IMPRINT_YEAR_RE.match(line)143 if im and ok(int(im.group(1))):144 # the line must look like an imprint: the year alone, or mostly upper-case words / a place name145 letters = [c for c in line if c.isalpha()]146 if not letters or sum(c.isupper() for c in letters) / len(letters) > 0.6 or re.search(r"London|New York|Boston|Philadelphia|Edinburgh|Oxford|Cambridge|Chicago|Paris|Leipzig|Berlin|Toronto|Washington|Glasgow|Dublin|Baltimore|Montreal|Melbourne", line):147 return FrontMatterDate(int(im.group(1)), " ".join(line.split())[:160], "imprint", [int(im.group(1))])148 rm = _ROMAN_YEAR_RE.match(line)149 if rm and len(rm.group(1)) >= 4:150 v = roman_to_int(rm.group(1))151 if v and ok(v):152 return FrontMatterDate(v, " ".join(line.split())[:160], "roman", [v])153 return None154155156# --------------------------------------------------------------------------------------------------------------157# Block classification (plain text)158# --------------------------------------------------------------------------------------------------------------159CHAPTER_RE = re.compile(r"^(?:chapter|chap\.|lecture|book|part|section|essay|letter)\s+(?:[ivxlcdm]+|\d+|[a-z]+(?:-[a-z]+)?)\b", re.IGNORECASE)160ROMAN_ONLY_RE = re.compile(r"^[IVXLC]{1,7}\.?$")161ROMAN_TITLE_RE = re.compile(r"^[IVXLC]{1,7}\.\s+\S")162NUMBERED_SECTION_RE = re.compile(r"^\d{1,2}\.\d{1,2}\s+[A-Z][^\n]{2,90}$") # textbook sections: "1.1 What Is Sociology?"163PREFACE_RE = re.compile(r"^(?:(?:the\s+)?(?:author'?s?|translator'?s?|editor'?s?)\s+)?(?:preface|foreword|prefatory note|advertisement to the|to the reader|preface to the [\w\s]+ edition)\b", re.IGNORECASE)164# Only a bare "Introduction" heading gets the introduction role ("Introduction to Sociology" is a chapter title)165INTRO_RE = re.compile(r"^(?:general\s+|an\s+)?(?:introduction|introductory(?:\s+(?:remarks|chapter|note|observations))?)\s*[.:]?\s*$", re.IGNORECASE)166SPECIAL_SKIP: list[tuple[re.Pattern, str]] = [167 (re.compile(r"^(?:table of )?contents\b|^(?:analytical|detailed|deteailed) (?:table of )?contents\b|^list of (?:illustrations|plates|figures|tables|maps)\b|^illustrations$|^plates$", re.IGNORECASE), "toc"),168 (re.compile(r"^(?:general |alphabetical |subject )?index\b|^index of\b", re.IGNORECASE), "index"),169 (re.compile(r"^(?:foot-?notes|notes|endnotes|notes and references|references)$", re.IGNORECASE), "notes"),170 (re.compile(r"^(?:appendix|appendices)\b", re.IGNORECASE), "appendix"),171 (re.compile(r"^(?:bibliography|works cited|list of authorities|literature cited)\b", re.IGNORECASE), "bibliography"),172 (re.compile(r"^glossary\b", re.IGNORECASE), "glossary"),173 (re.compile(r"^attributions\b", re.IGNORECASE), "attributions"),174 (re.compile(r"^(?:advertisements?|catalogue of|a catalogue|list of (?:new )?(?:books|works|publications))\b|^(?:new|recent|standard) (?:books|works|publications)\b", re.IGNORECASE), "adverts"),175 (re.compile(r"^errata\b|^corrigenda\b", re.IGNORECASE), "errata"),176 # textbook apparatus (OpenStax / CNX): skipped until the next strong heading177 (re.compile(r"^(?:chapter review|key terms|key concepts(?: and summary)?|key equations|review questions|section quiz|short answer|further research|chapter summary|conceptual questions|problems(?: (?:&|and) exercises)?|exercises|homework|practice|test prep(?: for ap.? courses)?|critical thinking(?: questions)?|solutions?|answer key|check your (?:learning|understanding))$", re.IGNORECASE), "review"),178 (re.compile(r"^(?:the end|finis)\.?$", re.IGNORECASE), "end"),179]180SKIP_TO_END = {"index", "attributions", "end", "adverts"}181ILLUSTRATION_START_RE = re.compile(r"^\s*\[\s*(?:illustration|ilustration|picture|plate|fig(?:ure)?\.?)\b", re.IGNORECASE)182FOOTNOTE_START_RE = re.compile(r"^\s*(?:\[\s*(?:\d{1,3}|[A-Za-z]|\*{1,3}|†|‡|footnote[^\]]*?)\s*\]|footnote\s+\d+\s*:|\{\d{1,3}\})", re.IGNORECASE)183TRANSCRIBER_RE = re.compile(r"^\s*\[?\s*(?:transcriber'?s?|producer'?s?|editor'?s? note for the e-?text)\s*note", re.IGNORECASE)184PRODUCER_RE = re.compile(r"^\s*(?:produced by|e-?text prepared by|this (?:e-?text|e-?book|file) was produced|transcribed (?:from|by)|credits?:|html version by|scanned by|proofread by|updated editions will replace)", re.IGNORECASE)185INLINE_ILLUSTRATION_RE = re.compile(r"\[\s*(?:illustration|ilustration|picture)\b[^\]]*\]", re.IGNORECASE | re.DOTALL)186INLINE_FOOTNOTE_ANCHOR_RE = re.compile(r"(?<=[^\s\[])\[(?:\d{1,3}|[A-Z]|[a-z]|\*{1,3}|†|‡)\]|(?<=[^\s{])\{\d{1,3}\}")187ITALIC_RE = re.compile(r"(?<![\w_])_([^_\n]+?)_(?![\w_])")188BOLD_EQ_RE = re.compile(r"(?<![\w=])=([A-Za-z][^=\n]{0,80}?)=(?![\w=])")189PAGE_MARK_RE = re.compile(r"\{(?:pg|page|p\.)?\s*\d{1,4}\}|\[pg\s*\d{1,4}\]|\[Page \d{1,4}\]", re.IGNORECASE)190191192def _caps_ratio(s: str) -> float:193 letters = [c for c in s if c.isalpha()]194 return sum(c.isupper() for c in letters) / len(letters) if letters else 0.0195196197def _strip_heading(s: str) -> str:198 s = ITALIC_RE.sub(r"\1", re.sub(r"\s+", " ", s))199 return s.strip().strip("_*=").strip()200201202def _special_kind(title: str) -> str | None:203 t = re.sub(r"^[\dIVXLC]+[.)]\s*", "", _strip_heading(title)).rstrip(".:;, ").strip()204 for pat, kind in SPECIAL_SKIP:205 if pat.search(t):206 return kind207 return None208209210def _heading_kind(lines: list[str]) -> str | None:211 """Classify a short block as a heading. Returns 'chapter', 'preface', 'introduction', 'caps', a special kind212 (toc, index, notes, ...), or None when the block is not a heading."""213 if not lines or len(lines) > 4:214 return None215 joined = " ".join(lines)216 n = word_count(joined)217 if n == 0 or n > 20:218 return None219 first = _strip_heading(lines[0])220 sk = _special_kind(first)221 if sk and (n <= 8 or sk in ("toc",)):222 return sk223 if PREFACE_RE.search(first) and n <= 10:224 return "preface"225 if INTRO_RE.search(first) and n <= 10 and (_caps_ratio(first) > 0.6 or first[:1].isupper()):226 return "introduction"227 if CHAPTER_RE.match(first) and first[:1].isupper():228 return "chapter"229 if NUMBERED_SECTION_RE.match(first) and n <= 14 and len(lines) <= 2:230 return "chapter"231 if ROMAN_ONLY_RE.match(first) or (ROMAN_TITLE_RE.match(first) and _caps_ratio(first) > 0.8):232 return "chapter"233 if _caps_ratio(joined) >= 0.9 and n <= 15 and not joined.rstrip().endswith((",", ";")) and len(re.findall(r"[A-Za-z]", joined)) >= 3:234 return "caps"235 return None236237238def _is_table_like(lines: list[str]) -> bool:239 if len(lines) < 2:240 return False241 gaps = sum(1 for ln in lines if re.search(r"\S {3,}\S", ln.strip()))242 dotted = sum(1 for ln in lines if re.search(r"\.{4,}|(?:\.\s){4,}", ln))243 numeric = sum(1 for ln in lines if len(re.findall(r"\d", ln)) > 0.3 * max(1, len(ln.strip())))244 return (gaps + dotted) / len(lines) >= 0.5 or numeric / len(lines) >= 0.5245246247def _is_prose(lines: list[str]) -> bool:248 text = " ".join(ln.strip() for ln in lines)249 if word_count(text) < 25 or not re.search(r"[.!?][\"'’”)]?(\s|$)", text):250 return False251 ends_digit = sum(1 for ln in lines if re.search(r"\d\s*$", ln))252 avg = sum(len(ln.strip()) for ln in lines) / len(lines)253 return ends_digit / len(lines) < 0.3 and (len(lines) == 1 or avg >= 40)254255256def reflow(lines: list[str]) -> str:257 """Join hard-wrapped lines into one paragraph; a line ending in a single hyphen followed by a lower-case258 continuation is joined without space (the hyphen is kept: we never guess whether it was a soft hyphen)."""259 out = ""260 for ln in lines:261 s = ln.strip()262 if not s:263 continue264 if out.endswith("-") and not out.endswith("--") and s[:1].islower():265 out += s266 else:267 out += (" " if out else "") + s268 return out269270271def clean_inline(text: str, counters: dict) -> str:272 n0 = len(text)273 text2 = INLINE_ILLUSTRATION_RE.sub(" ", text)274 if len(text2) != n0:275 counters["inline_illustrations_removed"] = counters.get("inline_illustrations_removed", 0) + 1276 text3, k = INLINE_FOOTNOTE_ANCHOR_RE.subn("", text2)277 if k:278 counters["footnote_anchors_removed"] = counters.get("footnote_anchors_removed", 0) + k279 text3, k = PAGE_MARK_RE.subn(" ", text3)280 if k:281 counters["page_marks_removed"] = counters.get("page_marks_removed", 0) + k282 text3 = ITALIC_RE.sub(r"\1", text3)283 text3 = BOLD_EQ_RE.sub(r"\1", text3)284 return normalize_text(text3)285286287# --------------------------------------------------------------------------------------------------------------288# Events + state machine shared by the text and HTML front-ends289# --------------------------------------------------------------------------------------------------------------290@dataclass291class _Event:292 type: str # heading | para | drop293 text: str = ""294 kind: str | None = None # heading kind295 prose: bool = False296 flags: list[str] = field(default_factory=list)297 reason: str | None = None298299300def _txt_events(body: str, counters: dict) -> list[_Event]:301 blocks = re.split(r"\n[ \t]*\n", body)302 events: list[_Event] = []303 open_bracket: str | None = None # multi-block [Illustration ...] / [Footnote ...] / [Transcriber's note ...]304 for blk in blocks:305 raw_lines = [ln.rstrip() for ln in blk.split("\n")]306 lines = [ln for ln in raw_lines if ln.strip()]307 if not lines:308 continue309 joined = "\n".join(lines)310 if open_bracket:311 events.append(_Event("drop", reason=open_bracket))312 if "]" in joined:313 open_bracket = None314 continue315 for pat, reason in ((ILLUSTRATION_START_RE, "illustration"), (FOOTNOTE_START_RE, "footnote"), (TRANSCRIBER_RE, "transcriber_note")):316 if pat.match(lines[0]):317 events.append(_Event("drop", reason=reason))318 if lines[0].lstrip().startswith("[") and joined.count("[") > joined.count("]"):319 open_bracket = reason320 break321 else:322 if PRODUCER_RE.match(lines[0]):323 events.append(_Event("drop", reason="producer_credit"))324 continue325 hk = _heading_kind([ln.strip() for ln in lines])326 if hk:327 events.append(_Event("heading", _strip_heading(" ".join(ln.strip() for ln in lines)), kind=hk))328 continue329 if all(re.match(r"^ {4,}|^\t", ln) for ln in lines):330 events.append(_Event("drop", reason="indented_block", prose=_is_prose(lines)))331 continue332 if _is_table_like(lines):333 events.append(_Event("drop", reason="table_like"))334 continue335 text = clean_inline(reflow(lines), counters)336 if not text:337 events.append(_Event("drop", reason="empty_after_cleaning"))338 continue339 flags = ["list_item"] if re.match(r"^\s*(?:[•◦▪●■–]|-\s|\(?[a-z0-9]\)\s)", lines[0]) else []340 events.append(_Event("para", text, prose=_is_prose(lines) and not flags, flags=flags))341 return events342343344_STRONG = {"chapter", "preface", "introduction"}345346347def _heading_key(title: str) -> str:348 t = re.sub(r"^(?:chapter|chap\.|lecture|book|part|section)\s+\S+\s*", "", title, flags=re.IGNORECASE)349 t = re.sub(r"^[\dIVXLC]+[.)]?\s+", "", t)350 return re.sub(r"[^a-z]", "", t.lower())351352353def _drop_toc_duplicates(events: list[_Event], max_paras: int = 2) -> list[_Event]:354 """A strong heading whose title reappears later and that is followed by at most ``max_paras`` paragraphs355 before the next heading is a table-of-contents entry (e.g. 'detailed contents' with chapter summaries):356 the heading and its few paragraphs are turned into drop events."""357 idx = [i for i, e in enumerate(events) if e.type == "heading"]358 keys = {i: _heading_key(events[i].text) for i in idx if events[i].kind in _STRONG}359 out = list(events)360 for n, i in enumerate(idx):361 k = keys.get(i)362 if not k or not any(keys.get(j) == k for j in idx[n + 1:]):363 continue364 nxt = idx[n + 1] if n + 1 < len(idx) else len(events)365 paras = [j for j in range(i + 1, nxt) if events[j].type == "para"]366 if len(paras) <= max_paras:367 out[i] = _Event("drop", reason="toc_duplicate_heading")368 for j in paras:369 out[j] = _Event("drop", reason="toc_duplicate_heading")370 return out371372373def _role_for(title: str | None, kind: str | None, parent_role: str) -> str:374 if kind == "preface" or (title and PREFACE_RE.search(re.sub(r"^(?:chapter|chap\.)\s+\S+\s*", "", title, flags=re.IGNORECASE))):375 return "preface"376 if kind == "introduction":377 return "introduction"378 if title:379 rest = re.sub(r"^(?:chapter|chap\.|lecture|part|book)\s+\S+\s*", "", title, flags=re.IGNORECASE).strip()380 if INTRO_RE.search(rest) and section_role(rest) == "introduction":381 return "introduction"382 # A generic ALL-CAPS heading cannot be told apart from a new main division, so it never inherits383 # preface/introduction from the previous section (that mislabelled whole books as "preface").384 return "chapter"385386387@dataclass388class _Emitter:389 paragraphs: list[ParsedParagraph] = field(default_factory=list)390 parts: list[str] = field(default_factory=list)391 cursor: int = 0392393 def emit(self, text: str, sid: str, title: str | None, role: str, depth: int, flags: list[str]) -> None:394 if not text:395 return396 pr = protect(text)397 start = self.cursor398 self.parts.append(text + "\n\n")399 self.cursor += len(text) + 2400 self.paragraphs.append(ParsedParagraph(sid, title, role, depth, len(self.paragraphs), text, start, start + len(text), pr.citations, pr.math, pr.urls, pr.numbers, flags))401402403def _run_events(events: list[_Event], counters: dict, drop_front: bool) -> _Emitter:404 em = _Emitter()405 state = "front" if drop_front else "body"406 sec_n = 0407 sid, title, role, depth = "body", None, "other", 1408 pending: _Event | None = None # last heading seen inside the table of contents409 chapter_open_without_text = False410411 def drop(reason: str) -> None:412 counters[f"dropped_{reason}"] = counters.get(f"dropped_{reason}", 0) + 1413414 def open_section(ev: _Event) -> None:415 nonlocal sec_n, sid, title, role, depth, chapter_open_without_text416 if ev.kind == "caps" and chapter_open_without_text and title:417 title = f"{title} {ev.text}" # chapter number on one line, chapter title on the next418 role = _role_for(title, None, role)419 return420 sec_n += 1421 sid = f"s{sec_n}"422 if ev.kind == "caps":423 title, depth = ev.text, 2424 role = _role_for(ev.text, "caps", role)425 else:426 title, depth = ev.text, 1427 role = _role_for(ev.text, ev.kind, role)428 chapter_open_without_text = ev.kind in _STRONG429430 for ev in events:431 if state == "end":432 drop("back_matter")433 continue434 if ev.type == "drop":435 drop(ev.reason or "other")436 continue437 if state in ("toc",):438 if ev.type == "para" and ev.prose:439 state = "body"440 if pending is not None:441 open_section(pending)442 pending = None443 else:444 if ev.type == "heading" and ev.kind in _STRONG | {"caps"}:445 pending = ev446 drop("toc")447 continue448 if state.startswith("skip:"):449 if ev.type == "heading" and ev.kind in _STRONG:450 state = "body"451 else:452 drop(state.split(":", 1)[1])453 continue454 if state == "front":455 if ev.type == "heading" and ev.kind in _STRONG:456 state = "body"457 elif ev.type == "heading" and ev.kind == "toc":458 state = "toc"459 drop("toc")460 continue461 else:462 drop("front_matter")463 continue464 # state == body465 if ev.type == "heading":466 if ev.kind == "toc":467 state, pending = "toc", None468 drop("toc")469 elif ev.kind in SKIP_TO_END:470 state = "end"471 drop(ev.kind)472 elif ev.kind not in _STRONG and ev.kind != "caps":473 state = f"skip:{ev.kind}"474 drop(ev.kind)475 else:476 open_section(ev)477 continue478 em.emit(ev.text, sid, title, role, depth, list(ev.flags))479 chapter_open_without_text = False480 return em481482483def _finish(em: _Emitter, stub: dict, counters: dict, warnings: list[str]) -> ParsedDocument:484 for k, v in sorted(counters.items()):485 if v:486 warnings.append(f"{k}: {v}")487 if not em.paragraphs:488 warnings.append("no paragraphs extracted")489 authors = list(stub.get("authors") or [])490 author_group = stub.get("author_group")491 if not author_group and authors:492 toks = authors[0].split()493 if toks:494 author_group = re.sub(r"[^a-z]", "", toks[-1].lower()) + ("_" + toks[0][0].lower() if len(toks) > 1 else "")495 return ParsedDocument(496 title=stub.get("title"),497 authors=authors,498 author_group=author_group,499 venue=stub.get("venue"),500 publisher=stub.get("publisher"),501 doc_type=stub.get("doc_type") or "book",502 pub_date=None,503 pub_year=None,504 pub_date_evidence=None,505 pre2018_verified=False,506 subjects=list(stub.get("subject_raw") or []),507 language=stub.get("language"),508 license_code="UNKNOWN",509 license_uri=None,510 license_evidence=None,511 body_text="".join(em.parts),512 paragraphs=em.paragraphs,513 warnings=warnings,514 article_type="book",515 ids={},516 )517518519def parse_book_text(text: str, stub: dict | None = None) -> ParsedDocument:520 stub = stub or {}521 warnings: list[str] = []522 counters: dict = {}523 text = text.replace("\r\n", "\n").replace("\r", "\n").lstrip("")524 body = text525 if stub.get("source") == "gutenberg" or PG_START_RE.search(text) or PG_OLD_START_RE.search(text):526 sp = strip_gutenberg_boilerplate(text)527 body = sp.body528 if not sp.start_found:529 warnings.append("gutenberg start marker not found: header kept")530 if not sp.end_found:531 warnings.append("gutenberg end marker not found: footer kept")532 events = _drop_toc_duplicates(_txt_events(body, counters))533 has_strong = any(e.type == "heading" and e.kind in _STRONG for e in events)534 if not has_strong:535 warnings.append("no chapter/preface/introduction heading found: front matter not dropped")536 em = _run_events(events, counters, drop_front=has_strong)537 return _finish(em, stub, counters, warnings)538539540# --------------------------------------------------------------------------------------------------------------541# HTML / XHTML / EPUB542# --------------------------------------------------------------------------------------------------------------543_HTML_DROP_CLASSES = re.compile(r"\b(?:pagenum|fnanchor|footnote|footnotes|fnote|toc|figcenter|figleft|figright|figure|caption|illustration|index|pg-boilerplate|sidenote|marginnote|tnote|transnote|transcriber)\b", re.IGNORECASE)544_HTML_DROP_IDS = re.compile(r"^(?:pg-header|pg-footer|pg-start-separator|pg-end-separator|pg-machine-header|toc|index|footnotes)$", re.IGNORECASE)545546547def _html_events(data: bytes, counters: dict) -> list[_Event]:548 from lxml import html as lhtml549550 root = lhtml.fromstring(data)551 for bad in root.xpath("//script|//style|//nav|//head|//table|//figure|//figcaption|//aside|//sup[a]|//a[contains(@class,'fnanchor')]"):552 tag = bad.tag if isinstance(bad.tag, str) else "node"553 counters[f"html_dropped_{tag}"] = counters.get(f"html_dropped_{tag}", 0) + 1554 bad.drop_tree()555 for el in list(root.iter()):556 if not isinstance(el.tag, str) or el.getparent() is None:557 continue558 if _HTML_DROP_CLASSES.search(el.get("class") or "") or _HTML_DROP_IDS.match(el.get("id") or ""):559 counters["html_dropped_by_class"] = counters.get("html_dropped_by_class", 0) + 1560 el.drop_tree()561 events: list[_Event] = []562 for el in root.iter("h1", "h2", "h3", "h4", "p"):563 txt = normalize_text(el.text_content())564 if not txt:565 continue566 if el.tag in ("h1", "h2", "h3"):567 kind = _heading_kind([txt]) if word_count(txt) <= 20 else None568 if kind is None:569 kind = "caps" # generic heading: a section, but not an anchor that ends the front matter570 events.append(_Event("heading", txt, kind=kind))571 continue572 if el.tag == "h4" or any(a.tag in ("blockquote",) for a in el.iterancestors()):573 events.append(_Event("drop", reason="blockquote" if el.tag == "p" else "h4"))574 continue575 if ILLUSTRATION_START_RE.match(txt):576 events.append(_Event("drop", reason="illustration"))577 continue578 if FOOTNOTE_START_RE.match(txt):579 events.append(_Event("drop", reason="footnote"))580 continue581 text = clean_inline(txt, counters)582 if text:583 events.append(_Event("para", text, prose=word_count(text) >= 25))584 return events585586587def parse_book_html(data: bytes, stub: dict | None = None) -> ParsedDocument:588 stub = stub or {}589 counters: dict = {}590 events = _drop_toc_duplicates(_html_events(data, counters))591 has_strong = any(e.type == "heading" and e.kind in _STRONG for e in events)592 em = _run_events(events, counters, drop_front=has_strong)593 return _finish(em, stub, counters, [] if has_strong else ["no strong heading found: front matter not dropped"])594595596def _epub_spine(zf: zipfile.ZipFile) -> list[str]:597 from lxml import etree598599 container = etree.fromstring(zf.read("META-INF/container.xml"))600 opf_path = container.xpath("//*[local-name()='rootfile']/@full-path")[0]601 opf = etree.fromstring(zf.read(opf_path))602 base = posixpath.dirname(opf_path)603 manifest = {it.get("id"): it.get("href") for it in opf.xpath("//*[local-name()='manifest']/*[local-name()='item']")}604 spine = [manifest.get(ref.get("idref")) for ref in opf.xpath("//*[local-name()='spine']/*[local-name()='itemref']")]605 return [posixpath.normpath(posixpath.join(base, h)) for h in spine if h]606607608def parse_book_epub(path: Path, stub: dict | None = None) -> ParsedDocument:609 stub = stub or {}610 counters: dict = {}611 events: list[_Event] = []612 with zipfile.ZipFile(path) as zf:613 names = set(zf.namelist())614 for href in _epub_spine(zf):615 if href in names:616 events.extend(_html_events(zf.read(href), counters))617 events = _drop_toc_duplicates(events)618 has_strong = any(e.type == "heading" and e.kind in _STRONG for e in events)619 em = _run_events(events, counters, drop_front=has_strong)620 return _finish(em, stub, counters, [] if has_strong else ["no strong heading found: front matter not dropped"])621622623def parse_book_file(path: Path, stub: dict) -> ParsedDocument:624 """Entry point used by ``htr.ingest.corpus.parse_raw`` for raw_format txt | html | epub."""625 fmt = stub.get("raw_format", "txt")626 path = Path(path)627 if fmt == "html":628 return parse_book_html(path.read_bytes(), stub)629 if fmt == "epub":630 return parse_book_epub(path, stub)631 raw = path.read_bytes()632 text = raw.decode("utf-8", errors="replace")633 doc = parse_book_text(text, stub)634 n_bad = text.count("�")635 if n_bad:636 doc.warnings.append(f"utf-8 decoding replaced {n_bad} invalid byte sequences")637 return doc638