path: root/converter/extractors.py

"""Text extraction from book files (TXT, PDF, EPUB) and text/HTML cleaning."""

import codecs
import logging
import re
import zipfile
from html import unescape
from pathlib import Path
from typing import List, NamedTuple

try:
    from bs4 import BeautifulSoup
    BS4_AVAILABLE = True
except ImportError:
    BS4_AVAILABLE = False

logger = logging.getLogger(__name__)


class Section(NamedTuple):
    """A titled chunk of a book (e.g. an EPUB chapter)."""

    title: str
    text: str


class Book(NamedTuple):
    """A book's metadata plus its titled sections."""

    title: str
    author: str
    sections: List[Section]


def extract_text(file_path: Path) -> str:
    """Extract text from a book file based on its extension."""
    extension = file_path.suffix.lower()
    if extension == ".txt":
        return _extract_txt(file_path)
    if extension == ".pdf":
        return _extract_pdf(file_path)
    if extension == ".epub":
        return extract_epub(file_path)
    raise ValueError(f"Unsupported file format: {extension}")


def extract_sections(file_path: Path) -> List[Section]:
    """Extract the book's text as titled sections (chapters).

    EPUB files are split on their spine documents so they can be converted
    one chapter at a time. TXT and PDF files have no chapter structure and
    always yield a single section.
    """
    if file_path.suffix.lower() == ".epub":
        chapters = _extract_epub_chapters(file_path)
        if not chapters:
            raise RuntimeError("All EPUB extraction methods failed")
        return chapters

    return [Section(file_path.stem, extract_text(file_path))]


def extract_book(file_path: Path) -> Book:
    """Extract sections plus book-level metadata (title, author).

    EPUB and PDF files carry embedded metadata; missing fields (and TXT
    files, which have none) fall back to the file stem for the title and
    an empty author.
    """
    title, author = "", ""
    extension = file_path.suffix.lower()
    if extension == ".epub":
        title, author = _epub_metadata(file_path)
    elif extension == ".pdf":
        title, author = _pdf_metadata(file_path)
    return Book(title or file_path.stem, author.strip(), extract_sections(file_path))


def _epub_metadata(file_path: Path) -> tuple:
    """Return (title, author) from an EPUB's Dublin Core metadata."""
    try:
        import ebooklib
        from ebooklib import epub

        book = epub.read_epub(str(file_path))
        title = _first_dc_value(book.get_metadata("DC", "title"))
        author = _first_dc_value(book.get_metadata("DC", "creator"))
        return title, author
    except Exception as exc:
        logger.warning("Could not read EPUB metadata: %s", exc)
        return "", ""


def _pdf_metadata(file_path: Path) -> tuple:
    """Return (title, author) from a PDF's document info dictionary."""
    try:
        from pypdf import PdfReader

        reader = PdfReader(str(file_path))
        info = reader.metadata or {}
        title = str(info.get("/Title") or "")
        author = str(info.get("/Author") or "")
        return title, author
    except Exception as exc:
        logger.warning("Could not read PDF metadata: %s", exc)
        return "", ""


def _first_dc_value(entries) -> str:
    """First value of an ebooklib DC metadata list: [(value, ...), ...]."""
    if not entries:
        return ""
    value = entries[0][0]
    return str(value).strip() if value else ""


def _extract_epub_chapters(file_path: Path) -> List[Section]:
    """Return one Section per EPUB spine document (chapter), in reading order."""
    import ebooklib

    book = None
    for method in (_read_epub_ebooklib, _read_epub_zipfile, _read_epub_manual):
        try:
            book = method(file_path)
        except Exception as exc:
            logger.warning("EPUB chapter method %s failed: %s", method.__name__, exc)
            continue
        if book:
            break

    if book is None:
        return []

    chapters = []
    for title, text in book:
        cleaned = clean_html(text)
        if cleaned.strip():
            chapters.append(Section(title or file_path.stem, cleaned))
    return chapters


def _toc_titles(book) -> dict:
    """Flatten an ebooklib TOC into a ``{href: title}`` mapping."""
    titles = {}

    def walk(nodes) -> None:
        for node in nodes:
            if isinstance(node, (tuple, list)):
                walk(node[1] if len(node) > 1 else [])
                continue
            href = getattr(node, "href", None)
            title = getattr(node, "title", None)
            if href and title:
                titles[href.split("#")[0]] = title

    walk(book.toc)
    return titles


def _read_epub_ebooklib(file_path: Path):
    """Read EPUB spine documents as (title, html) pairs via ebooklib."""
    import ebooklib
    from ebooklib import epub

    book = epub.read_epub(str(file_path))
    titles = _toc_titles(book)
    items = []
    for entry in book.spine:
        item_id = entry[0] if isinstance(entry, (tuple, list)) else entry
        try:
            item = book.get_item_with_id(item_id)
        except Exception as exc:
            logger.debug("Skipping EPUB spine item %r: %s", item_id, exc)
            continue
        if not item or item.get_type() != ebooklib.ITEM_DOCUMENT:
            continue
        if isinstance(item, epub.EpubNav):
            continue
        content = item.get_body_content()
        if content:
            if isinstance(content, bytes):
                content = content.decode("utf-8", errors="ignore")
            title = (titles.get(item.file_name)
                     or titles.get(item.get_name())
                     or getattr(item, "title", None)
                     or item.get_name())
            items.append((title, str(content)))
    return items


def _read_epub_zipfile(file_path: Path):
    """Read EPUB HTML members as (title, html) pairs, ordered by filename."""
    items = []
    with zipfile.ZipFile(file_path, "r") as epub_zip:
        for file_name in sorted(epub_zip.namelist(), key=_natural_key):
            if file_name.lower().endswith((".html", ".xhtml", ".htm")):
                try:
                    content = epub_zip.read(file_name).decode("utf-8", errors="ignore")
                    items.append((Path(file_name).stem, content))
                except Exception as exc:
                    logger.debug("Skipping EPUB member %r: %s", file_name, exc)
    return items


def _read_epub_manual(file_path: Path):
    """Last-resort read of any markup-looking EPUB member."""
    skipped_extensions = (".jpg", ".jpeg", ".png", ".gif", ".css", ".js")
    items = []
    with zipfile.ZipFile(file_path, "r") as epub_zip:
        for file_name in sorted(epub_zip.namelist(), key=_natural_key):
            if file_name.lower().endswith(skipped_extensions):
                continue
            try:
                content = epub_zip.read(file_name).decode("utf-8", errors="ignore")
                if "<" in content and len(content.strip()) > 100:
                    items.append((Path(file_name).stem, content))
            except Exception as exc:
                logger.debug("Skipping EPUB member %r: %s", file_name, exc)
    return items


def clean_text(text: str) -> str:
    """Normalize whitespace and strip standalone page numbers.

    Page numbers are removed only when they appear as a short number alone on
    its own line (before whitespace collapsing), so inline numbers like
    "42 years", "1,000" or "3.5" are preserved.
    """
    if not text:
        return ""
    # Standalone page numbers (digits alone on a line) must go BEFORE the
    # newline-collapsing step below.
    text = re.sub(r"(?m)^\s*\d{1,4}\s*$", " ", text)
    text = re.sub(r"\s+", " ", text)
    return text.strip()


def clean_html(html_content: str) -> str:
    """Strip markup, scripts and styles from HTML content."""
    if not html_content:
        return ""

    if BS4_AVAILABLE:
        try:
            soup = BeautifulSoup(html_content, "html.parser")
            for tag in soup(["script", "style"]):
                tag.decompose()
            text = soup.get_text(separator=" ")
            return re.sub(r"\s+", " ", text).strip()
        except Exception as exc:
            logger.debug("BeautifulSoup cleaning failed, falling back to regex: %s", exc)

    # Fallback regex cleaning
    html_content = re.sub(r"<style[^>]*>.*?</style>", "", html_content, flags=re.DOTALL | re.IGNORECASE)
    html_content = re.sub(r"<script[^>]*>.*?</script>", "", html_content, flags=re.DOTALL | re.IGNORECASE)
    html_content = re.sub(r"<[^>]+>", " ", html_content)
    html_content = unescape(html_content)
    html_content = re.sub(r"\s+", " ", html_content)
    return html_content.strip()


def extract_epub(file_path: Path) -> str:
    """Extract the book's text from EPUB, trying several methods in order."""
    chapters = _extract_epub_chapters(file_path)
    if not chapters:
        raise RuntimeError("All EPUB extraction methods failed")
    return "\n\n".join(section.text for section in chapters)


def _natural_key(name: str):
    """Sort key that orders numeric runs numerically (chapter2 before chapter10)."""
    return [int(part) if part.isdigit() else part.lower()
            for part in re.split(r"(\d+)", name)]


def _extract_txt(file_path: Path) -> str:
    """Extract from TXT, handling BOMs and common encodings (latin-1 is the catch-all).

    UTF-16 files without a BOM are detected via NUL bytes; otherwise they would
    silently decode as NUL-interleaved UTF-8 or cp1252/latin-1 garbage.
    """
    data = file_path.read_bytes()

    if data.startswith((codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE)):
        return clean_text(data.decode("utf-32"))
    if data.startswith((codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)):
        return clean_text(data.decode("utf-16"))
    if data.startswith(codecs.BOM_UTF8):
        return clean_text(data.decode("utf-8-sig"))

    # No BOM: UTF-16 without BOM is common on Windows; detect via NUL bytes.
    sample = data[:4096]
    even_nuls = sum(1 for i, b in enumerate(sample) if b == 0 and i % 2 == 0)
    odd_nuls = sum(1 for i, b in enumerate(sample) if b == 0 and i % 2 == 1)
    if even_nuls or odd_nuls:
        encoding = "utf-16-be" if even_nuls > odd_nuls else "utf-16-le"
        return clean_text(data.decode(encoding))

    for encoding in ("utf-8", "cp1252", "latin-1"):
        try:
            return clean_text(data.decode(encoding))
        except UnicodeError:
            continue
    raise ValueError(f"Could not decode text file: {file_path}")


def _extract_pdf(file_path: Path) -> str:
    """Extract from PDF."""
    from pypdf import PdfReader

    text = ""
    with open(file_path, "rb") as file:
        pdf_reader = PdfReader(file)
        total_pages = len(pdf_reader.pages)
        logger.info("PDF has %d pages", total_pages)

        for page_num, page in enumerate(pdf_reader.pages, 1):
            try:
                page_text = page.extract_text() or ""
                if page_text.strip():
                    text += f"\n\n{page_text}"
                if page_num % 10 == 0:
                    logger.debug("Extracted %d/%d pages", page_num, total_pages)
            except Exception as exc:
                logger.warning("Failed to extract page %d: %s", page_num, exc)

        logger.info("Extracted text from %d pages, %d characters total", total_pages, len(text))
    return clean_text(text)