tds3p-rsvp-reader/epub_to_chapters.py

1713 lines
46 KiB
Python

#!/usr/bin/env python3
"""
EPUB -> a TOC-ordered folder of chapter TXT files.
Usage:
python epub_to_chapters.py book.epub
python epub_to_chapters.py book.epub output_directory
Output:
output_directory/
TOC.txt
Dedication.txt
Chapter 1.txt
Chapter 2.txt
...
Each line of TOC.txt is both a section title and the basename of its text file.
The output files contain only section text. No chapter heading is added.
Strategy:
1. Use the EPUB table of contents when available.
2. Correctly handle #fragment links inside XHTML files.
3. Handle multiple chapters inside one XHTML file.
4. Handle chapters spanning multiple XHTML files.
5. Fall back to the EPUB spine when the TOC is unavailable.
6. Clean HTML into readable plain text.
"""
import re
import struct
import sys
import unicodedata
from collections import Counter, defaultdict
from pathlib import Path
from urllib.parse import unquote, urlsplit
from bs4 import BeautifulSoup, NavigableString
try:
from ebooklib import epub, ITEM_DOCUMENT
except ModuleNotFoundError:
epub = None
ITEM_DOCUMENT = None
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
SKIP_TAGS = {
"script",
"style",
"noscript",
"svg",
"canvas",
"nav",
"form",
"iframe",
"object",
"embed",
}
BLOCK_TAGS = {
"p",
"div",
"section",
"article",
"header",
"footer",
"aside",
"blockquote",
"pre",
"figure",
"figcaption",
"table",
"thead",
"tbody",
"tfoot",
"tr",
"td",
"th",
"li",
"dt",
"dd",
"h1",
"h2",
"h3",
"h4",
"h5",
"h6",
}
FAT_FORBIDDEN_FILENAME_CHARACTERS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
STYLE_ITALIC = 0x01
STYLE_BOLD = 0x02
STYLE_MARKER = "\x01"
SIDECAR_MAGIC = b"RSVP"
# Version 2 activates timing modifiers. Version 1 sidecars remain readable by
# the firmware, but their reserved timing bytes are required to be zero.
SIDECAR_VERSION = 2
SIDECAR_HEADER = struct.Struct("<4sBBQI")
CSS_RULE = re.compile(r"([^{}]+)\{([^{}]*)\}")
CSS_CLASS = re.compile(r"\.([A-Za-z_-][A-Za-z0-9_-]*)")
WORD_LOOKUP_KEY = re.compile(r"^[^A-Za-z]*([A-Za-z]+(?:['-][A-Za-z]+)*)[^A-Za-z]*$")
MAX_SECTION_BYTES = 1024 * 1024
MAX_SECTION_WORDS = 150_000
MAX_TOKEN_BYTES = 65_535
MAX_SECTION_NAME_BYTES = 200
RESERVED_SECTION_NAMES = {"toc", "conversion_report"}
TIMING_NONE = 0
TIMING_DOUBLE = 2
TIMING_TRIPLE = 3
DISPLAY_PUNCTUATION = str.maketrans({
"\ufeff": "", # UTF-8 byte-order mark.
"\u00a0": " ", # Non-breaking space.
"\u00ad": "-", # Soft hyphen.
"\u2010": "-", # Hyphen.
"\u2011": "-", # Non-breaking hyphen.
"\u2012": "-", # Figure dash.
"\u2013": "-", # En dash.
"\u2014": "- ", # Em dash; retain an RSVP token boundary.
"\u2015": "- ", # Horizontal bar; retain an RSVP token boundary.
"\u2018": "'", # Left single quotation mark.
"\u2019": "'", # Right single quotation mark.
"\u201a": "'", # Single low-9 quotation mark.
"\u201b": "'", # Single high-reversed-9 quotation mark.
"\u201c": '"', # Left double quotation mark.
"\u201d": '"', # Right double quotation mark.
"\u201e": '"', # Double low-9 quotation mark.
"\u2026": "...", # Ellipsis.
"\u2022": "*", # Bullet.
"\u2023": "*", # Triangular bullet.
"\u2043": "-", # Hyphen bullet.
"\u2219": "*", # Bullet operator.
"\u25e6": "*", # White bullet.
"\u25aa": "*", # Small square.
"\u25cf": "*", # Black circle.
"\u2212": "-", # Minus sign.
"\u00d7": "x", # Multiplication sign.
"\u00f7": "/", # Division sign.
"\u2044": "/", # Fraction slash.
"\u2260": "!=", # Not equal.
"\u2264": "<=", # Less than or equal.
"\u2265": ">=", # Greater than or equal.
"\u2248": "~=", # Almost equal.
"\u2020": "+", # Dagger.
"\u2021": "++", # Double dagger.
"\u2032": "'", # Prime.
"\u2033": "\"", # Double prime.
"\u215b": "1/8", # Vulgar fractions.
"\u215c": "3/8",
"\u215d": "5/8",
"\u215e": "7/8",
"\u00bc": "1/4",
"\u00bd": "1/2",
"\u00be": "3/4",
"\u2153": "1/3",
"\u2154": "2/3",
"\u2155": "1/5",
"\u2156": "2/5",
"\u2157": "3/5",
"\u2158": "4/5",
"\u2159": "1/6",
"\u215a": "5/6",
"\u2042": "***", # Asterism.
"\u2055": "***", # Flower punctuation mark.
"\u2766": "***", # Floral heart.
"\u2767": "***", # Rotated floral heart.
"\u200b": " ", # Zero-width space; retain a word boundary.
"\u00c6": "AE", # Latin letters not expanded by Unicode NFKD.
"\u00e6": "ae",
"\u00d0": "D",
"\u00f0": "d",
"\u00d8": "O",
"\u00f8": "o",
"\u00de": "TH",
"\u00fe": "th",
"\u00df": "ss",
"\u0131": "i",
"\u0141": "L",
"\u0142": "l",
"\u0152": "OE",
"\u0153": "oe",
"\u0110": "D",
"\u0111": "d",
"\u0126": "H",
"\u0127": "h",
"\u014a": "N",
"\u014b": "n",
})
# ---------------------------------------------------------------------------
# General helpers
# ---------------------------------------------------------------------------
def normalize_display_text(text):
"""Convert common Unicode typography to device-safe ASCII where possible.
Unsupported scripts are deliberately retained so ``write_conversion_report``
can identify them for a book-specific transliteration or replacement choice.
"""
text = text.translate(DISPLAY_PUNCTUATION)
# NFKD expands presentation ligatures, Latin letters with diacritics, and
# compatible number forms (for example, superscript digits). Removing the
# resulting combining marks yields readable ASCII for Latin text without
# discarding a non-Latin base character.
text = unicodedata.normalize("NFKD", text)
normalized = []
for character in text:
if character in "\r\n":
normalized.append(character)
elif character.isspace():
normalized.append(" ")
elif unicodedata.combining(character):
continue
elif unicodedata.category(character) == "Cf":
# Formatting controls such as direction marks and joiners are not
# printable on the device. U+200B was handled above as a space.
continue
else:
normalized.append(character)
return "".join(normalized)
def normalize_text(text):
"""
Normalize extracted text while preserving paragraph breaks.
"""
text = normalize_display_text(text)
text = text.replace("\r\n", "\n")
text = text.replace("\r", "\n")
# Remove spaces around line breaks.
text = re.sub(r"[ \t]*\n[ \t]*", "\n", text)
# Collapse horizontal whitespace.
text = re.sub(r"[ \t]+", " ", text)
# Remove excessive blank lines.
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
def style_flags_for_text_node(node):
"""Return the inline EPUB emphasis flags applying to a text node."""
flags = 0
parent = node.parent
while parent is not None:
if parent.name in {"i", "em"}:
flags |= STYLE_ITALIC
elif parent.name in {"b", "strong"}:
flags |= STYLE_BOLD
flags |= int(parent.attrs.get("data-rsvp-style-flags", 0))
parent = parent.parent
return flags
def append_styled_text(pieces, node):
"""Append text bracketed by private markers that survive normalization."""
text = str(node)
if not text:
return
flags = style_flags_for_text_node(node)
pieces.extend((STYLE_MARKER, chr(ord("0") + flags), text,
STYLE_MARKER, "0"))
def decode_styled_text(text):
"""Remove style markers and return sparse per-word style records."""
characters = []
character_styles = []
current_style = 0
position = 0
while position < len(text):
character = text[position]
if character == STYLE_MARKER:
if position + 1 >= len(text):
raise ValueError("unterminated style marker")
current_style = ord(text[position + 1]) - ord("0")
if current_style & ~(STYLE_ITALIC | STYLE_BOLD):
raise ValueError("invalid style marker")
position += 2
continue
characters.append(character)
character_styles.append(current_style)
position += 1
plain_text = "".join(characters)
style_records = []
for word_index, match in enumerate(re.finditer(r"\S+", plain_text)):
flags = 0
for style in character_styles[match.start():match.end()]:
flags |= style
if flags:
style_records.append((word_index, flags, 0))
return plain_text, style_records
def lookup_word_key(token):
"""Return a lower-case dictionary key for one RSVP token, if it has one."""
match = WORD_LOOKUP_KEY.fullmatch(token.replace("\u2019", "'"))
return match.group(1).lower() if match else None
IRREGULAR_WORD_BASES = {
"ain't": "be",
"can't": "can",
"did": "do",
"does": "do",
"had": "have",
"has": "have",
"says": "say",
"shan't": "shall",
"won't": "will",
}
CONTRACTION_SUBJECTS = {
"he", "here", "how", "i", "it", "let", "she", "that", "there", "they",
"we", "what", "when", "where", "who", "why", "you",
}
MINIMUM_ENGLISH_ZIPF_FREQUENCY = 2.0
class EnglishVocabulary:
"""Membership view combining a dictionary with common-word frequency."""
def __init__(self, words, zipf_frequency):
self.words = words
self.zipf_frequency = zipf_frequency
def __contains__(self, word):
return (
word in self.words or
self.zipf_frequency(word, "en") >= MINIMUM_ENGLISH_ZIPF_FREQUENCY
)
def english_word_bases(word):
"""Yield plausible dictionary forms for a normalized English word.
``english-words``' web2 list contains many stems but omits common
inflections and contractions. These conservative transformations prevent
those ordinary forms from being mistaken for unfamiliar vocabulary without
turning an unknown stem into a known word.
"""
stems = [word]
if word.endswith("n't"):
stems.append(word[:-3])
elif word.endswith(("'d", "'ll", "'m", "'re", "'s", "'ve")):
stems.append(word[:-2])
elif word.endswith("s'"):
stems.append(word[:-1])
for stem in stems:
yield stem
irregular_base = IRREGULAR_WORD_BASES.get(stem)
if irregular_base:
yield irregular_base
if stem.endswith("ies") and len(stem) > 3:
yield stem[:-3] + "y"
if stem.endswith("ing") and len(stem) > 4:
root = stem[:-3]
yield root
yield root + "e"
if len(root) > 2 and root[-1] == root[-2]:
yield root[:-1]
if stem.endswith("ed") and len(stem) > 3:
root = stem[:-2]
yield root
yield stem[:-1]
if len(root) > 2 and root[-1] == root[-2]:
yield root[:-1]
if stem.endswith(("er", "est")) and len(stem) > 3:
suffix_length = 2 if stem.endswith("er") else 3
root = stem[:-suffix_length]
yield root
if root.endswith("i"):
yield root[:-1] + "y"
if len(root) > 2 and root[-1] == root[-2]:
yield root[:-1]
if stem.endswith("es") and len(stem) > 3:
yield stem[:-2]
yield stem[:-1]
elif stem.endswith("s") and len(stem) > 2:
yield stem[:-1]
def is_familiar_english_word(word, english_words):
"""Return whether a lookup key is familiar English or a normal variant."""
if (
word.endswith(("'d", "'ll", "'m", "'re", "'s", "'ve")) and
word[:-2] in CONTRACTION_SUBJECTS
):
return True
if "-" in word:
components = word.split("-")
return all(
component and is_familiar_english_word(component, english_words)
for component in components
)
return any(candidate in english_words for candidate in english_word_bases(word))
def validate_token_lengths(text):
"""Reject sections whose token index would exceed the device memory budget."""
for word_number, match in enumerate(re.finditer(r"\S+", text), start=1):
if word_number > MAX_SECTION_WORDS:
raise ValueError(
f"section exceeds the {MAX_SECTION_WORDS} word limit"
)
if len(match.group().encode("utf-8")) > MAX_TOKEN_BYTES:
raise ValueError(
f"token {word_number} exceeds the {MAX_TOKEN_BYTES} byte limit"
)
def load_english_dictionary():
"""Load the desktop-only English dictionary required for timing analysis."""
try:
from english_words import get_english_words_set
from wordfreq import zipf_frequency
except ModuleNotFoundError as exc:
raise RuntimeError(
"Unfamiliar-word timing needs english-words and wordfreq. Install converter "
"dependencies with: python3 -m pip install -r requirements.txt"
) from exc
# web2 is a general-purpose English dictionary. Proper nouns and invented
# words are normally absent, which deliberately makes them qualify. The
# frequency fallback fills web2's gaps for ordinary inflections and words.
return EnglishVocabulary(
get_english_words_set(["web2"], lower=True),
zipf_frequency,
)
def add_unfamiliar_word_timing(chapters, english_words):
"""Merge book-wide unfamiliar-word timing into chapter sidecar records.
A word outside the supplied English dictionary receives a triple delay for
its first ten appearances in the book and on its first appearance in every
chapter. All later appearances receive a double delay. Timing is stored
sparsely alongside any EPUB style metadata already present for that token.
"""
global_occurrences = Counter()
for chapter in chapters:
records = {
index: [style_flags, timing_modifier]
for index, style_flags, timing_modifier in chapter["style_records"]
}
chapter_occurrences = set()
for word_index, match in enumerate(re.finditer(r"\S+", chapter["text"])):
key = lookup_word_key(match.group())
if key is None or is_familiar_english_word(key, english_words):
continue
prior_occurrences = global_occurrences[key]
first_in_chapter = key not in chapter_occurrences
timing_modifier = (
TIMING_TRIPLE
if prior_occurrences < 10 or first_in_chapter
else TIMING_DOUBLE
)
record = records.setdefault(word_index, [0, TIMING_NONE])
record[1] = max(record[1], timing_modifier)
global_occurrences[key] += 1
chapter_occurrences.add(key)
chapter["style_records"] = [
(word_index, style_flags, timing_modifier)
for word_index, (style_flags, timing_modifier) in sorted(records.items())
]
def source_fingerprint(text):
"""Return a stable 64-bit FNV-1a fingerprint of UTF-8 chapter text."""
value = 0xCBF29CE484222325
for byte in text.encode("utf-8"):
value ^= byte
value = (value * 0x100000001B3) & 0xFFFFFFFFFFFFFFFF
return value
def encode_varint(value):
"""Encode a non-negative integer as unsigned LEB128."""
if value < 0:
raise ValueError("varint value must be non-negative")
encoded = bytearray()
while True:
byte = value & 0x7F
value >>= 7
encoded.append(byte | (0x80 if value else 0))
if not value:
return bytes(encoded)
def style_sidecar_bytes(text, style_records):
"""Return validated sparse style and timing metadata for a chapter."""
payload = bytearray(SIDECAR_HEADER.pack(
SIDECAR_MAGIC,
SIDECAR_VERSION,
0,
source_fingerprint(text),
len(style_records),
))
previous_index = 0
for record_number, (word_index, style_flags, timing_modifier) in enumerate(style_records):
if word_index < 0 or style_flags & ~(STYLE_ITALIC | STYLE_BOLD):
raise ValueError("invalid style sidecar record")
if timing_modifier not in {TIMING_NONE, TIMING_DOUBLE, TIMING_TRIPLE}:
raise ValueError("invalid style sidecar record")
if style_flags == 0 and timing_modifier == TIMING_NONE:
raise ValueError("invalid style sidecar record")
if record_number and word_index <= previous_index:
raise ValueError("style records must have ascending word indexes")
delta = word_index if record_number == 0 else word_index - previous_index
payload.extend(encode_varint(delta))
payload.extend((style_flags, timing_modifier))
previous_index = word_index
return bytes(payload)
def write_style_sidecar(output_file, text, style_records):
"""Write validated sparse style and timing metadata beside a chapter."""
output_file.write_bytes(style_sidecar_bytes(text, style_records))
def output_section_name(title, number, used_names):
"""
Return a readable, FAT-safe, unique TOC entry and file basename.
"""
title = normalize_display_text(title or "")
title = re.sub(r"\s+", " ", title).strip()
title = FAT_FORBIDDEN_FILENAME_CHARACTERS.sub("-", title)
title = title.rstrip(". ")
if not title:
title = f"Section {number}"
encoded_title = title.encode("utf-8")
if len(encoded_title) > MAX_SECTION_NAME_BYTES:
title = encoded_title[:MAX_SECTION_NAME_BYTES].decode(
"utf-8", errors="ignore"
).rstrip(". ")
candidate = title
duplicate = 2
while candidate.casefold() in used_names:
suffix = f" ({duplicate})"
available = MAX_SECTION_NAME_BYTES - len(suffix.encode("utf-8"))
stem = title.encode("utf-8")[:available].decode("utf-8", errors="ignore")
candidate = f"{stem.rstrip('. ')}{suffix}"
duplicate += 1
used_names.add(candidate.casefold())
return candidate
def remaining_non_ascii(text):
"""Return counts of characters the device's printable ASCII font lacks."""
return Counter(character for character in text if ord(character) > 127)
def apply_substitutions(text, substitutions):
"""Replace each unsupported character using its chosen ASCII text."""
return "".join(substitutions.get(character, character) for character in text)
def prompt_for_substitutions(characters, input_fn=input, output_fn=print):
"""Ask once for the printable-ASCII replacement of each character."""
substitutions = {}
for character in sorted(characters, key=ord):
codepoint = f"U+{ord(character):04X}"
name = unicodedata.name(character, "UNNAMED")
count = characters[character]
prompt = (
f"Replace {codepoint} {name} ({count} occurrence"
f"{'s' if count != 1 else ''}; {character!r}) with ASCII "
"text, or press Enter to remove it: "
)
while True:
try:
replacement = input_fn(prompt)
except EOFError as exc:
raise RuntimeError(
f"input ended before a replacement was supplied for {codepoint}"
) from exc
if all(32 <= ord(replacement_character) <= 126
for replacement_character in replacement):
substitutions[character] = replacement
break
output_fn("Replacement must contain printable ASCII characters only.")
return substitutions
def substitute_unsupported_characters(chapters, input_fn=input, output_fn=print):
"""Interactively make generated section titles and text ASCII-safe.
Text is changed before style records and timing metadata are generated, so
replacements that add or remove characters cannot invalidate word indexes.
"""
characters = Counter()
for chapter in chapters:
chapter["title"] = normalize_display_text(chapter["title"] or "")
chapter["text"] = normalize_text(chapter["text"])
characters.update(remaining_non_ascii(chapter["title"]))
characters.update(remaining_non_ascii(chapter["text"]))
if not characters:
return {}
output_fn()
output_fn("Unsupported characters need an ASCII replacement for this book.")
substitutions = prompt_for_substitutions(characters, input_fn, output_fn)
for chapter in chapters:
chapter["title"] = apply_substitutions(chapter["title"], substitutions)
chapter["text"] = normalize_text(
apply_substitutions(chapter["text"], substitutions)
)
return substitutions
def write_conversion_report(output_dir, toc_entries, written_sections,
substitutions=None):
"""Write a report for unsupported characters left in generated text."""
occurrences = defaultdict(Counter)
source_files = defaultdict(set)
generated_files = [("TOC.txt", "\n".join(toc_entries))]
generated_files.extend(written_sections)
for filename, text in generated_files:
for character, count in remaining_non_ascii(text).items():
occurrences[character][filename] += count
source_files[character].add(filename)
report_file = output_dir / "CONVERSION_REPORT.txt"
lines = [
"EPUB conversion report",
"",
"The reader's current font supports printable ASCII only.",
]
if substitutions:
lines.extend([
"",
"User substitutions applied to all occurrences:",
])
for character in sorted(substitutions, key=ord):
codepoint = f"U+{ord(character):04X}"
name = unicodedata.name(character, "UNNAMED")
lines.append(
f"{codepoint} {name} -> {substitutions[character]!r}"
)
if not occurrences:
lines.extend([
"",
"Result: all generated TOC and chapter text is ASCII-safe.",
])
else:
lines.extend([
"",
"WARNING: unsupported non-ASCII characters remain.",
"Choose a per-book treatment: transliterate, replace deliberately,",
"or defer the book until a Unicode-capable font is available.",
"",
"Character Count Generated files",
"-------------- ----- ---------------",
])
for character in sorted(occurrences, key=ord):
codepoint = f"U+{ord(character):04X}"
name = unicodedata.name(character, "UNNAMED")
total = sum(occurrences[character].values())
filenames = ", ".join(sorted(source_files[character]))
lines.append(
f"{codepoint} {name:<30} {total:5} {filenames}"
)
report_file.write_text("\n".join(lines) + "\n", encoding="utf-8")
return report_file, bool(occurrences)
def normalize_href(href):
"""
Return:
(normalized_path, fragment)
Example:
Text/Chapter01.xhtml#part2
becomes:
("text/chapter01.xhtml", "part2")
"""
if not href:
return "", ""
href = unquote(href)
href = href.replace("\\", "/")
parsed = urlsplit(href)
path = parsed.path.lstrip("/")
path = re.sub(r"/+", "/", path)
return path, parsed.fragment
def same_document(item_a, item_b):
"""
Compare two ebooklib document items.
"""
if item_a is item_b:
return True
try:
return item_a.get_name() == item_b.get_name()
except Exception:
return False
# ---------------------------------------------------------------------------
# EPUB document indexing
# ---------------------------------------------------------------------------
def build_document_index(book):
"""
Build indexes allowing TOC hrefs to be resolved to ebooklib items.
"""
by_path = {}
by_filename = {}
for item in book.get_items():
if item.get_type() != ITEM_DOCUMENT:
continue
names = set()
try:
name = item.get_name()
if name:
names.add(name)
except Exception:
pass
file_name = getattr(item, "file_name", None)
if file_name:
names.add(file_name)
for name in names:
path, _ = normalize_href(name)
if not path:
continue
by_path[path] = item
by_filename[Path(path).name] = item
return by_path, by_filename
def find_document(href, by_path, by_filename):
"""
Resolve a TOC href to an EPUB document.
"""
path, fragment = normalize_href(href)
if not path:
return None, fragment
# Exact path.
item = by_path.get(path)
if item:
return item, fragment
# Try without leading ../ segments.
simplified = path
while simplified.startswith("../"):
simplified = simplified[3:]
item = by_path.get(simplified)
if item:
return item, fragment
# Last resort: filename only.
filename = Path(path).name
item = by_filename.get(filename)
if item:
return item, fragment
return None, fragment
# ---------------------------------------------------------------------------
# HTML parsing
# ---------------------------------------------------------------------------
def style_flags_from_css(declarations):
"""Return supported italic/bold flags from a CSS declaration block."""
flags = 0
for property_name, value in re.findall(
r"([A-Za-z-]+)\s*:\s*([^;{}]+)",
declarations.lower(),
):
if property_name == "font-style" and (
"italic" in value or "oblique" in value
):
flags |= STYLE_ITALIC
elif property_name == "font-weight" and (
"bold" in value or re.search(r"\b[6-9]00\b", value)
):
flags |= STYLE_BOLD
elif property_name == "font":
if "italic" in value or "oblique" in value:
flags |= STYLE_ITALIC
if "bold" in value or re.search(r"\b[6-9]00\b", value):
flags |= STYLE_BOLD
return flags
def stylesheet_class_styles(book):
"""Collect supported text styles declared by CSS classes in an EPUB."""
class_styles = {}
for item in book.get_items():
name = getattr(item, "file_name", "") or ""
try:
name = item.get_name() or name
except Exception:
pass
if not name.lower().endswith(".css"):
continue
content = item.get_content().decode("utf-8", errors="replace")
content = re.sub(r"/\*.*?\*/", "", content, flags=re.DOTALL)
for selectors, declarations in CSS_RULE.findall(content):
flags = style_flags_from_css(declarations)
if not flags:
continue
for selector in selectors.split(","):
for class_name in CSS_CLASS.findall(selector):
class_styles[class_name] = (
class_styles.get(class_name, 0) | flags
)
return class_styles
def apply_css_styles(soup, class_styles):
"""Mark tags whose class or inline CSS supplies a supported text style."""
for tag in soup.find_all(True):
flags = style_flags_from_css(tag.get("style", ""))
for class_name in tag.get("class", []):
flags |= class_styles.get(class_name, 0)
if flags:
tag["data-rsvp-style-flags"] = str(flags)
def parse_document(item, class_styles=None):
"""
Parse an EPUB XHTML document.
"""
soup = BeautifulSoup(
item.get_content(),
"html.parser",
)
for tag in soup.find_all(SKIP_TAGS):
tag.decompose()
if class_styles is not None:
apply_css_styles(soup, class_styles)
return soup
def document_root(soup):
"""
Return the best content root.
"""
if soup.body:
return soup.body
if soup.html:
return soup.html
return soup
def find_fragment(soup, fragment):
"""
Locate an EPUB #fragment anchor.
Supports both:
id="foo"
and older:
name="foo"
"""
if not fragment:
return None
element = soup.find(id=fragment)
if element:
return element
return soup.find(attrs={"name": fragment})
# ---------------------------------------------------------------------------
# HTML -> text
# ---------------------------------------------------------------------------
def html_to_text(root):
"""
Convert an HTML element into readable text.
Paragraphs and other block-level elements become separated by
blank lines.
"""
pieces = []
def walk(node):
if isinstance(node, NavigableString):
append_styled_text(pieces, node)
return
if not getattr(node, "name", None):
return
if node.name in SKIP_TAGS:
return
if node.name == "br":
pieces.append("\n")
return
is_block = node.name in BLOCK_TAGS
if is_block:
pieces.append("\n")
for child in node.children:
walk(child)
if is_block:
pieces.append("\n")
walk(root)
return normalize_text("".join(pieces))
# ---------------------------------------------------------------------------
# Precise range extraction
# ---------------------------------------------------------------------------
def extract_range(root, start_element=None, end_element=None):
"""
Extract text from start_element up to, but not including,
end_element.
If start_element is None, extraction starts at the beginning.
"""
pieces = []
started = start_element is None
for node in root.descendants:
if node is start_element:
started = True
if node is end_element:
break
if not started:
continue
if isinstance(node, NavigableString):
append_styled_text(pieces, node)
elif getattr(node, "name", None) == "br":
pieces.append("\n")
elif getattr(node, "name", None) in BLOCK_TAGS:
pieces.append("\n")
return normalize_text("".join(pieces))
# ---------------------------------------------------------------------------
# TOC handling
# ---------------------------------------------------------------------------
def flatten_toc(toc):
"""
Flatten ebooklib's nested TOC structure.
Returns a list of:
{
"title": "...",
"href": "..."
}
"""
result = []
def walk(items):
for item in items:
# Nested TOC section.
if isinstance(item, tuple):
section, children = item
if section is not None:
href = getattr(section, "href", None)
title = getattr(section, "title", "") or ""
if href:
result.append({
"title": title.strip(),
"href": href,
})
walk(children)
else:
href = getattr(item, "href", None)
title = getattr(item, "title", "") or ""
if href:
result.append({
"title": title.strip(),
"href": href,
})
walk(toc)
return result
def get_toc(book):
"""
Return the EPUB TOC.
"""
toc = getattr(book, "toc", None)
if not toc:
return []
return flatten_toc(toc)
# ---------------------------------------------------------------------------
# Resolve TOC entries
# ---------------------------------------------------------------------------
def resolve_toc(book):
"""
Resolve every usable TOC entry to an EPUB document.
"""
by_path, by_filename = build_document_index(book)
entries = []
for entry in get_toc(book):
item, fragment = find_document(
entry["href"],
by_path,
by_filename,
)
if item is None:
print(
f"Warning: cannot resolve TOC entry: "
f"{entry['title']!r} -> {entry['href']}"
)
continue
entries.append({
"title": entry["title"],
"href": entry["href"],
"item": item,
"fragment": fragment,
})
return entries
# ---------------------------------------------------------------------------
# Determine document order
# ---------------------------------------------------------------------------
def spine_documents(book):
"""
Return EPUB document items in spine/reading order.
"""
result = []
by_id = {}
for item in book.get_items():
if item.get_type() == ITEM_DOCUMENT:
by_id[item.get_id()] = item
for entry in book.spine:
# ebooklib versions represent spine entries differently.
if isinstance(entry, tuple):
item_id = entry[0]
else:
item_id = entry
item = by_id.get(item_id)
if item is not None:
result.append(item)
return result
# ---------------------------------------------------------------------------
# Improved multi-document chapter extraction
# ---------------------------------------------------------------------------
def document_order(book):
"""
Return document items in EPUB spine order.
If the spine is unusable, fall back to document iteration order.
"""
result = spine_documents(book)
if result:
return result
return [
item
for item in book.get_items()
if item.get_type() == ITEM_DOCUMENT
]
def item_index_map(book):
"""
Map EPUB document identity to its spine position.
"""
docs = document_order(book)
return {
id(item): index
for index, item in enumerate(docs)
}
def extract_chapter(book, entries, index, doc_positions, class_styles=None):
"""
Extract a complete chapter, including any XHTML documents between
its starting document and the next chapter's starting document.
"""
current = entries[index]
next_entry = (
entries[index + 1]
if index + 1 < len(entries)
else None
)
current_item = current["item"]
current_pos = doc_positions.get(id(current_item))
if current_pos is None:
current_soup = parse_document(current_item, class_styles)
root = document_root(current_soup)
start = find_fragment(
current_soup,
current["fragment"],
)
return extract_range(root, start, None)
# Position of next chapter's document.
next_pos = None
if next_entry is not None:
next_pos = doc_positions.get(id(next_entry["item"]))
# If we cannot determine the next position, only use current document.
if next_entry is not None and next_pos is None:
next_pos = current_pos + 1
docs = document_order(book)
pieces = []
# Process documents from current through the document before the next
# chapter. A fragment in the current document is a special case: include
# that document once and stop at the next fragment.
end_pos = (
next_pos + 1 if next_pos == current_pos and next_entry is not None else next_pos
if next_pos is not None
else len(docs)
)
for pos in range(current_pos, end_pos):
if pos >= len(docs):
break
item = docs[pos]
soup = parse_document(item, class_styles)
root = document_root(soup)
# -----------------------------------------------------------
# Current chapter's starting point.
# -----------------------------------------------------------
start = None
if pos == current_pos:
start = find_fragment(
soup,
current["fragment"],
)
# -----------------------------------------------------------
# Next chapter starts in this same document.
# -----------------------------------------------------------
end = None
if (
next_entry is not None
and pos == next_pos
and same_document(item, next_entry["item"])
):
end = find_fragment(
soup,
next_entry["fragment"],
)
# If the next chapter starts at the beginning of this document,
# nothing from this document belongs to the current chapter.
if pos == next_pos and next_entry is not None:
if next_entry["fragment"]:
if end is None:
continue
else:
# No fragment means the next chapter starts at the
# beginning of the document.
continue
text = extract_range(
root,
start,
end,
)
if text:
pieces.append(text)
return normalize_text(
"\n\n".join(pieces)
)
# ---------------------------------------------------------------------------
# Fallback: spine-based extraction
# ---------------------------------------------------------------------------
def fallback_spine_chapters(book, class_styles=None):
"""
Fallback when no usable TOC exists.
Each XHTML document becomes one output chapter.
"""
chapters = []
for item in document_order(book):
soup = parse_document(item, class_styles)
root = document_root(soup)
text = html_to_text(root)
if not text:
continue
chapters.append({
"title": "",
"text": text,
})
return chapters
# ---------------------------------------------------------------------------
# Duplicate removal
# ---------------------------------------------------------------------------
def comparison_text(text):
"""
Normalize text for duplicate detection.
"""
return re.sub(
r"\s+",
" ",
text,
).strip().lower()
def remove_duplicate_chapters(chapters):
"""
Remove immediately duplicated chapter content.
EPUB TOCs occasionally contain duplicate navigation entries.
"""
result = []
previous = None
for chapter in chapters:
normalized = comparison_text(
chapter["text"]
)
if not normalized:
continue
if normalized == previous:
continue
result.append(chapter)
previous = normalized
return result
# ---------------------------------------------------------------------------
# Main conversion
# ---------------------------------------------------------------------------
def convert_epub(epub_path, output_dir, overwrite=False):
epub_path = Path(epub_path)
output_dir = Path(output_dir)
if epub is None:
print(
"Error: the EPUB converter needs ebooklib. Install dependencies "
"with: python3 -m pip install -r requirements.txt"
)
sys.exit(1)
if output_dir.exists() and any(output_dir.iterdir()):
if not overwrite:
print(
f"Error: output directory is not empty: {output_dir}. "
"Choose an empty directory or pass --overwrite."
)
sys.exit(1)
for output_file in output_dir.iterdir():
if output_file.is_file() and (
output_file.suffix in {".txt", ".rsvp"} or
output_file.name == "CONVERSION_REPORT.txt"
):
output_file.unlink()
output_dir.mkdir(parents=True, exist_ok=True)
print(f"Reading: {epub_path}")
try:
book = epub.read_epub(
str(epub_path)
)
except Exception as exc:
print(
f"Error: could not read EPUB:\n{exc}"
)
sys.exit(1)
class_styles = stylesheet_class_styles(book)
# ---------------------------------------------------------------
# First choice: EPUB TOC.
# ---------------------------------------------------------------
entries = resolve_toc(book)
chapters = []
if entries:
print(
f"TOC entries found: {len(entries)}"
)
positions = item_index_map(book)
for index, entry in enumerate(entries):
text = extract_chapter(
book,
entries,
index,
positions,
class_styles,
)
text = normalize_text(text)
if not text:
print(
f"Skipping empty TOC entry: "
f"{entry['title']!r}"
)
continue
chapters.append({
"title": entry["title"],
"text": text,
})
# ---------------------------------------------------------------
# Fallback: EPUB spine.
# ---------------------------------------------------------------
if not chapters:
print(
"No usable TOC chapters found."
)
print(
"Falling back to EPUB reading order..."
)
chapters = fallback_spine_chapters(
book,
class_styles,
)
try:
substitutions = substitute_unsupported_characters(chapters)
except RuntimeError as exc:
print(f"Error: {exc}")
sys.exit(1)
for chapter in chapters:
text, style_records = decode_styled_text(
normalize_text(chapter["text"])
)
chapter["text"] = text
chapter["style_records"] = style_records
# ---------------------------------------------------------------
# Remove duplicates.
# ---------------------------------------------------------------
chapters = remove_duplicate_chapters(
chapters
)
if not chapters:
print(
"ERROR: no readable text was found."
)
sys.exit(1)
try:
add_unfamiliar_word_timing(chapters, load_english_dictionary())
except RuntimeError as exc:
print(f"Error: {exc}")
sys.exit(1)
print(
f"Chapters to write: {len(chapters)}"
)
print()
# ---------------------------------------------------------------
# Write files.
# ---------------------------------------------------------------
toc_entries = []
prepared_sections = []
used_names = set(RESERVED_SECTION_NAMES)
for number, chapter in enumerate(
chapters,
start=1,
):
title = output_section_name(
chapter["title"],
number,
used_names,
)
toc_entries.append(title)
prepared_sections.append((title, chapter))
for title, chapter in prepared_sections:
text_bytes = chapter["text"].encode("utf-8")
try:
validate_token_lengths(chapter["text"])
except ValueError as exc:
print(f"Error: {title!r} exceeds reader limits: {exc}")
sys.exit(1)
sidecar_bytes = style_sidecar_bytes(
chapter["text"], chapter["style_records"]
)
if len(text_bytes) > MAX_SECTION_BYTES or len(sidecar_bytes) > MAX_SECTION_BYTES:
print(
f"Error: {title!r} exceeds the device's {MAX_SECTION_BYTES} byte "
"per-file limit. Split the section before converting."
)
sys.exit(1)
written_sections = []
for number, (title, chapter) in enumerate(prepared_sections, start=1):
output_file = output_dir / f"{title}.txt"
sidecar_file = output_dir / f"{title}.rsvp"
output_file.write_text(chapter["text"], encoding="utf-8")
write_style_sidecar(sidecar_file, chapter["text"], chapter["style_records"])
written_sections.append((output_file.name, chapter["text"]))
print(
f"{number:4}. {title} -> {output_file.name}, {sidecar_file.name}"
)
toc_file = output_dir / "TOC.txt"
toc_file.write_text(
"\n".join(toc_entries) + "\n",
encoding="utf-8",
)
report_file, has_unsupported_characters = write_conversion_report(
output_dir,
toc_entries,
written_sections,
substitutions,
)
print()
print(
f"Finished: {len(chapters)} files"
)
print(
f"Output: {output_dir.resolve()}"
)
if has_unsupported_characters:
print(
"WARNING: unsupported characters remain; review "
f"{report_file.name} before copying this book to the device."
)
else:
print(
f"Conversion report: {report_file.name} (ASCII-safe)"
)
# ---------------------------------------------------------------------------
# Command line interface
# ---------------------------------------------------------------------------
def main():
arguments = sys.argv[1:]
overwrite = "--overwrite" in arguments
if overwrite:
arguments.remove("--overwrite")
if not arguments or len(arguments) > 2:
print(
"Usage:\n"
" python epub_to_chapters.py book.epub\n"
" python epub_to_chapters.py book.epub output_directory\n"
" python epub_to_chapters.py book.epub output_directory --overwrite"
)
sys.exit(1)
epub_path = Path(
arguments[0]
)
if not epub_path.exists():
print(
f"Error: file not found: "
f"{epub_path}"
)
sys.exit(1)
if not epub_path.is_file():
print(
f"Error: not a file: "
f"{epub_path}"
)
sys.exit(1)
if len(arguments) >= 2:
output_dir = Path(
arguments[1]
)
else:
output_dir = Path(
"chapters"
)
convert_epub(epub_path, output_dir, overwrite)
if __name__ == "__main__":
main()