#!/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) return match.group(1).lower() if match else None 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 except ModuleNotFoundError as exc: raise RuntimeError( "Unfamiliar-word timing needs english-words. 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. return get_english_words_set(["web2"], lower=True) 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 key in 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()