llm-ingestion-okf/src/llm_ingestion_okf/extract.py

"""Door B extraction registry: dropped file bytes -> text, per file type.

All file-type -> text extraction lives here (the guard is text-only). The core
registry is stdlib-only and deterministic: `md`/`txt` pass through, `csv` renders
the Phase 1 markdown table, `json` is fenced verbatim, and `html`/`htm` are
reduced to text with `html.parser`. Binary types are `[extract]`-gated and each
now has a reader: `pdf` through `pdfplumber`, and the five office rows
(`docx`/`xlsx`/`pptx`/`odt`/`rtf`) through a table-driven converter seam over
the vendored binary. Every one of those gates is an IMPORT PROBE rather than a
membership test, so an absent extra is rejected with the same typed error
whatever the type. Never a silent skip and never a bundled parser in core.

Two of the five office rows are `measured` and three are `unmeasured` -- the
corpus this arm was built on contains zero `pptx`, `odt` or `rtf` files, so
those rows work by construction and have never met a document anyone wrote.
`_EVIDENCE` carries that per row and the suite asserts it, because an
unmeasured row must not read as a supported one.

`extract_text` returns the extracted text *content*; final LF framing and the
concept frontmatter are the materializer's concern (Phase 2 step 2), not this
registry's. No guard call and no model call anywhere in this module.
"""

from __future__ import annotations

import collections
import csv
import functools
import io
import re
import statistics
import tempfile
import warnings
import zipfile
from xml.etree import ElementTree
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from html.parser import HTMLParser
from pathlib import Path

from .errors import ExtractionError, ExtractionWarning
from .render import render_fenced_block, render_table

# Binary types gated behind the optional `[extract]` extra that it ships no
# parser for. EMPTY, and kept rather than deleted: the dispatch branch it feeds
# still raises `extractor_extra_missing`, and a later type that arrives before
# its reader belongs here rather than in a new mechanism. Every type the extra
# names now has a reader -- `.pdf` through the import probe in `_extract_pdf`,
# the five office rows through the converter seam below -- so the gate for all
# of them is an import probe, which is why the two tests for that code reach it
# that way.
_UNPARSED_OPTIONAL_EXTENSIONS: frozenset[str] = frozenset()

# The office rows: suffix -> the converter's reader name. THESE ROWS AND NO
# OTHERS. `.html` is excluded although the converter can read it: it already
# has a stdlib extractor here, so routing it through the converter would buy
# nothing and would add CVE-2025-51591 (SSRF via an iframe in HTML input),
# unpatched in every converter version. `.epub` is excluded on the "no gain"
# half of the same reason.
_PANDOC_FORMATS: dict[str, str] = {
    ".docx": "docx",
    ".xlsx": "xlsx",
    ".pptx": "pptx",
    ".odt": "odt",
    ".rtf": "rtf",
}

# What each row's behaviour actually rests on, asserted in the suite rather
# than written in a comment that rots. `measured` means real corpus files and a
# hand-counted fasit; `unmeasured` means the corpus contains ZERO files of that
# type, so the row works by construction and has never been checked against a
# document anyone wrote. An unmeasured row must not read as a supported one.
_EVIDENCE: dict[str, str] = {
    ".docx": "measured",
    ".xlsx": "measured",
    ".pptx": "unmeasured",
    ".odt": "unmeasured",
    ".rtf": "unmeasured",
}

# Load-bearing, all three, and none of them hygiene:
#
#   --eol=lf     the defaults produce DIFFERENT BYTES (maximum line length 75
#   --wrap=none  against 447), which a byte-pinned golden registers as a change
#                nobody made.
#   -t markdown  never `-t plain`: plain destroys the headings the segment
#                proposer reads. Measured -- a document yielding 15 entries
#                including two real headings yields 13 with none under `plain`,
#                so the writer choice silently sets the ceiling for the arm
#                downstream of it.
_PANDOC_WRITER = "markdown"
_PANDOC_ARGS = ("--eol=lf", "--wrap=none")

# The spreadsheet row writes PIPE tables, and it is the only row that does.
#
# The default writer prefers simple tables, which pad every cell out to the
# width of the widest cell in its column. Measured on the K2 price sheet: one
# 594-character prose cell turned every other row in that column into a run of
# up to 887 spaces between a label and its amount, 100 795 characters in all,
# and the header row named ONE column because only the first cell of the source
# row 1 is filled. The bytes reached the reader and the structure did not. The
# same sheet through this writer is 11 221 characters with no whitespace run
# longer than two, one row per line, each source column its own cell.
#
# `--columns=1` is load-bearing rather than cosmetic: the pipe writer pads cells
# out to the column width it computes from that setting, so at the default 72 a
# NARROW table gains runs of up to 45 spaces -- the same defect at a smaller
# scale. Measured across every office fixture and every K2 office file, the
# longest whitespace run with it is 2.
#
# SPREADSHEET-ONLY, deliberately. The other four rows have the same defect
# available to the same one-line fix (measured: the odt fixture 1366 -> 1105
# characters), but a spreadsheet IS a grid with no prose fallback, while moving
# the prose rows would move a corpus denominator that nothing has measured.
# `tests/test_extract.py` pins that scoping with three digests.
_SPREADSHEET_WRITER = "markdown-simple_tables-multiline_tables-grid_tables"
_SPREADSHEET_ARGS = (*_PANDOC_ARGS, "--columns=1")

# SpreadsheetML's namespace, needed to read the workbook's shared string table.
_SSML = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"

# A table cell whose whole content is an integer with the converter's trailing
# `.0`. Bounded by unescaped pipes on both sides so a cell containing an escaped
# `\|` can never be split in the middle.
_INTEGRAL_CELL = re.compile(r"(?<!\\)\|(\s*)(-?\d+)\.0(\s*)(?=(?<!\\)\|)")

# Conversion recovers text, on the same terms as PDF extraction: a drawing has
# no text to recover. Said out loud on every conversion rather than detected
# per document, for the same reason.
_OFFICE_LOSSY_WARNING = (
    "office-file conversion recovers text only: figures, diagrams, images and "
    "drawn shapes are not represented in the output (their captions are). A "
    "bundle built from drawn documents is incomplete by construction."
)

# Text extraction recovers text. A figure is a vector drawing with no text to
# recover — only its caption survives — so any bundle built from drawn
# documents is incomplete by construction. Said out loud on every PDF rather
# than detected per document: deciding "is there a figure here" is a layout
# heuristic this library does not own.
_PDF_LOSSY_WARNING = (
    "PDF extraction recovers text only: figures, diagrams and images are not "
    "represented in the output (their captions are). A bundle built from "
    "drawn documents is incomplete by construction."
)

# Tags whose text content is never document prose.
_SKIP_TAGS = frozenset({"script", "style"})


def decode_text(data: bytes) -> str:
    """Decode file bytes as UTF-8 (BOM-stripping), typed on failure.

    utf-8-sig so a byte-order mark never leaks into the first character
    (baseline parity with Door A's read_csv). A non-UTF-8 file is a corrupt
    input: fail fast with a typed error rather than leaking UnicodeDecodeError.
    """
    try:
        return data.decode("utf-8-sig")
    except UnicodeDecodeError as exc:
        raise ExtractionError(
            f"file bytes are not valid UTF-8: {exc}", code="extractor_decode_error"
        ) from exc


def _extract_passthrough(data: bytes) -> str:
    """`md`/`txt`: the decoded text verbatim."""
    return decode_text(data)


def _extract_csv(data: bytes) -> str:
    """`csv`: parse with the stdlib reader, render the Phase 1 markdown table."""
    reader = csv.reader(io.StringIO(decode_text(data)))
    header = next(reader, None)
    if header is None:
        raise ExtractionError("CSV has no header row", code="extractor_empty_csv")
    rows = list(reader)
    return render_table(header, rows)


def _extract_json(data: bytes) -> str:
    """`json`: the decoded text verbatim inside a fenced block (Phase 1 renderer)."""
    return render_fenced_block(decode_text(data))


class _HTMLTextExtractor(HTMLParser):
    """Collect document text, skipping `script`/`style`, tags as word boundaries.

    Tags contribute no text of their own but do separate words: a boundary
    space is emitted at every tag so adjacent block text (``</h1><p>``) does not
    fuse. Runs of whitespace collapse to single spaces in :meth:`text`.
    """

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self._parts: list[str] = []
        self._skip_depth = 0

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        self._parts.append(" ")
        if tag in _SKIP_TAGS:
            self._skip_depth += 1

    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        self._parts.append(" ")

    def handle_endtag(self, tag: str) -> None:
        if tag in _SKIP_TAGS and self._skip_depth > 0:
            self._skip_depth -= 1
        self._parts.append(" ")

    def handle_data(self, data: str) -> None:
        if self._skip_depth == 0:
            self._parts.append(data)

    def text(self) -> str:
        return " ".join("".join(self._parts).split())


def _extract_html(data: bytes) -> str:
    """`html`/`htm`: text via `html.parser`, script/style stripped (spec B3)."""
    parser = _HTMLTextExtractor()
    parser.feed(decode_text(data))
    parser.close()
    return parser.text()


def _extra_missing(suffix: str) -> ExtractionError:
    """The one rejection for a `[extract]` type without the extra installed.

    One constructor, one wording: the import probe and the still-unparsed
    types must be indistinguishable to a consumer, because they are the same
    fact — the extra is not installed.
    """
    return ExtractionError(
        f"extracting {suffix!r} requires the optional 'extract' extra "
        f"(pip install 'llm-ingestion-okf[extract]'); it is not installed",
        code="extractor_extra_missing",
    )


def _ocr_group_missing() -> ExtractionError:
    """The one rejection for `--ocr` without the optional `ocr` group.

    A DIFFERENT code from `extractor_extra_missing`, because it is a different
    fact and a different remedy: the `[extract]` extra can be fully installed
    -- the document parsed, the pages counted -- and the OCR engine still be
    absent. One error naming both would send an operator to reinstall
    something they already have.
    """
    return ExtractionError(
        "reading a PDF page with OCR requires the optional 'ocr' group "
        "(pip install 'llm-ingestion-okf[extract,ocr]'), which ships rapidocr "
        "on onnxruntime; it is not installed",
        code="extractor_ocr_group_missing",
    )


#: The literal placeholder `pdfminer.six` (behind `pdfplumber`) emits for a
#: glyph whose font carries no usable ToUnicode mapping. The text is present on
#: the page and unreadable in the extraction -- a failure that looks like
#: success, which is why it needs a measurement rather than an exception.
_CID_CODE = re.compile(r"\(cid:\d+\)")

#: The share of a page's extracted characters that must be `(cid:N)` codes
#: before `--ocr` reads the page as an image instead.
#:
#: MEASURED, not chosen: `docs/2026-09-08-k3-runde4-pdf-skrift-og-ocr.md`
#: reports the per-page distribution over the K2 corpus, and it is bimodal
#: with nothing in between -- one document's pages sit near 1.0 and every other
#: page in the corpus sits at 0.0. Any value in that gap selects the same
#: pages, which is what makes 0.10 defensible and also what makes it
#: uninformative about a corpus that has intermediate pages. Stated rather than
#: implied: this threshold is bounded by the corpus, not by a property of the
#: format.
OCR_CID_SHARE = 0.10

#: The resolution a page is rendered at before it is read as an image.
#: 200 dpi is what the round-4 measurement was taken at; the engine's own
#: preprocessing rescales from there, so this is a floor on how much of the
#: page's detail reaches it rather than a tuning knob. It is part of the output
#: contract in the same way the parser version is: OCR text is deterministic
#: within one resolution and one model version, and across neither.
OCR_DPI = 200


def cid_share(text: str) -> float:
    """The share of `text` made of `(cid:N)` placeholder codes, 0.0 for empty.

    Module level and importable: `tools/okf_cid_measure.py` answers the same
    question at DOCUMENT level, and two definitions of one metric drift.
    """
    if not text:
        return 0.0
    return sum(len(match.group(0)) for match in _CID_CODE.finditer(text)) / len(text)


def needs_ocr(text: str) -> bool:
    """Whether a page's extracted text is unusable enough to read the image.

    TWO conditions, because there are two ways a page's text never arrives and
    they look nothing alike: a page with no text layer extracts as the empty
    string, and a page whose fonts carry no ToUnicode mapping extracts as a
    full page of `(cid:N)`. A trigger written for one of them would leave the
    other exactly where it was.
    """
    return not text.strip() or cid_share(text) >= OCR_CID_SHARE


def _ocr_reader() -> Callable[[object], list[str]]:
    """The OCR engine, or the typed refusal. The import IS the gate.

    Same shape as `_extract_pdf`'s probe and for the same reason: membership in
    a suffix set cannot tell whether a package is importable, and this group is
    the one a consumer is most likely not to have.
    """
    try:
        import rapidocr
    except ImportError as exc:
        raise _ocr_group_missing() from exc
    if rapidocr is None:  # pragma: no cover - the sys.modules probe in tests
        raise _ocr_group_missing()
    engine = rapidocr.RapidOCR()

    def read(image: object) -> list[str]:
        result = engine(image)
        # `txts` is None when the detector found nothing at all, which is a
        # legitimate answer for a blank page and not an error.
        return [str(line) for line in (getattr(result, "txts", None) or ())]

    return read


#: Bold as a PDF says it: in the font's NAME (`Helvetica-Bold`,
#: `ABCDEF+Arial-BoldMT`). There is no weight attribute on a character, so the
#: name is the only place a text extractor can read it.
_PDF_BOLD_MARKER = "bold"

#: The deepest ATX level the emitted markdown may use. `_ATX` in `propose.py`
#: reads one to six hashes, and a document with seven distinct heading sizes
#: would otherwise emit a line the proposer reads as body.
_PDF_MAX_HEADING_LEVEL = 6


def _dominant(values: list[str]) -> str:
    """The most frequent value, ties broken by first occurrence.

    `Counter.most_common(1)` reduces to `max` over the items in insertion
    order, so the tie-break is document order and the result is deterministic
    for identical bytes -- which is the property everything downstream is
    pinned to.
    """
    return collections.Counter(values).most_common(1)[0][0]


def _typography(line: dict[str, object]) -> tuple[float, str] | None:
    """One line's dominant font size and font name, or `None` if it is blank.

    Blank characters are excluded from both: a space carries a size and a font
    like any other character, and a heading padded with body-sized spaces would
    read as body.
    """
    chars = [char for char in line["chars"] if str(char["text"]).strip()]  # type: ignore[attr-defined]
    if not chars:
        return None
    sizes = [f"{float(char['size']):.1f}" for char in chars]
    fonts = [str(char["fontname"]) for char in chars]
    return float(_dominant(sizes)), _dominant(fonts)


def _heading_levels(lines: list[tuple[str, float, str]]) -> dict[float, int]:
    """Which font sizes are headings in this document, and at what ATX level.

    The rule is the CONJUNCTION this repository already measured: larger than
    the body AND bold. `docs/2026-09-07-k3-arm-d.md`'s predecessor measured
    size-and-bold from poppler at recall 1.000 / precision 0.846, and measured
    that adding weight as a DISJUNCT made precision worse (0.786 -> 0.524). A
    disjunction here would mark every emphasised phrase in the body.

    The body size is the CHARACTER-weighted median over the whole document, not
    the page: a title page is 100 % heading by line count, and a per-page
    median would compare it with itself and mark nothing. Weighted by
    characters rather than lines for the same reason in miniature -- a document
    front-loaded with short lines has a line median that no paragraph shares.

    The ATX LEVEL is the size's rank among the heading sizes, largest first, so
    a document's own typographic hierarchy survives into the markdown instead
    of flattening to one level. Deeper than six is clamped, because `_ATX`
    reads six.
    """
    weighted: list[float] = []
    for text, size, _ in lines:
        weighted.extend([size] * len(text.replace(" ", "")))
    if not weighted:
        return {}
    body = statistics.median(weighted)
    sizes = {size for _, size, font in lines if size > body and _PDF_BOLD_MARKER in font.lower()}
    return {
        size: min(rank, _PDF_MAX_HEADING_LEVEL)
        for rank, size in enumerate(sorted(sizes, reverse=True), start=1)
    }


def _mark_headings(lines: list[tuple[str, float, str]], levels: dict[float, int]) -> str:
    """One page's lines as markdown, the heading sizes carrying their hashes.

    BOLD is checked again here rather than folded into the size map: a document
    can set a caption in the same size as a heading without setting it bold,
    and a map keyed on size alone would promote it.
    """
    out: list[str] = []
    for text, size, font in lines:
        level = levels.get(size) if _PDF_BOLD_MARKER in font.lower() else None
        out.append(f"{'#' * level} {text}" if level is not None and text else text)
    return "\n".join(out)


# How `_extract_pdf` joins its pages, named because the locator below has to
# reproduce the exact same arithmetic to turn a character offset back into a
# page number. Two constants that must agree, written once.
_PDF_PAGE_SEPARATOR = "\n\n"


@functools.lru_cache(maxsize=1)
def _pdf_pages(
    data: bytes, headings: bool = False, ocr: bool = False
) -> tuple[tuple[int, str], ...]:
    """Every page that produced text, as `(page number, text)`, in page order.

    The page NUMBER is 1-based and comes from the document, so a page that
    yielded nothing removes itself from the sequence without renumbering the
    ones after it -- which is the difference between "the third page that
    produced text" and "page 3", and the whole reason a locator is worth
    writing down.

    Memoised on the bytes AND on the two options, with room for exactly one
    entry: extraction and location are two calls about the same file with the
    same options, back to back, and parsing it twice would double the PDF cost
    of every corpus run for nothing. The options are part of the key because
    two renderings of one document are two different strings, and a locator
    built against the wrong one points at the wrong place with full confidence.

    `headings` and `ocr` are INDEPENDENT and compose. With both off this is the
    path every byte-pinned golden was measured on, unchanged: the default
    branch still calls `page.extract_text()` rather than reassembling the page
    from its lines. Measured, the two agree on 11 of 11 pages of a real tender
    PDF -- but "agree on the document I tried" is not a contract, so the
    default does not depend on it.
    """
    try:
        import pdfplumber
    except ImportError as exc:
        raise _extra_missing(".pdf") from exc

    read = _ocr_reader() if ocr else None
    try:
        with pdfplumber.open(io.BytesIO(data)) as pdf:
            # PASS ONE. Nothing is emitted here, because the heading rule needs
            # a fact about the WHOLE document -- the body's size -- and a page
            # cannot supply it. A title page is 100 % heading, and a per-page
            # median would compare it with itself and mark nothing.
            recovered: list[str | list[tuple[str, float, str]]] = []
            for page in pdf.pages:
                flat = (page.extract_text() or "").rstrip()
                if read is not None and needs_ocr(flat):
                    # The page's own text is unusable, so it is replaced
                    # WHOLESALE rather than merged with: a page of `(cid:N)`
                    # has nothing worth keeping, and interleaving two readings
                    # of one page would put a guess and a fact in one paragraph
                    # with no way to tell them apart. An OCR'd page carries no
                    # typography either -- the engine reports text, not fonts --
                    # so it is a finished string and never a heading candidate.
                    recovered.append("\n".join(read(page.to_image(resolution=OCR_DPI).original)))
                elif not headings:
                    recovered.append(flat)
                else:
                    recovered.append(
                        [
                            (str(line["text"]), *found)
                            for line in page.extract_text_lines()
                            if (found := _typography(line)) is not None
                        ]
                    )
            levels = _heading_levels(
                [
                    line
                    for page_lines in recovered
                    if not isinstance(page_lines, str)
                    for line in page_lines
                ]
            )
            # PASS TWO.
            pages = [
                page_lines
                if isinstance(page_lines, str)
                else _mark_headings(page_lines, levels).rstrip()
                for page_lines in recovered
            ]
    except ExtractionError:
        raise
    except Exception as exc:  # noqa: BLE001 - third-party parser, wrapped never leaked
        raise ExtractionError(
            f"the PDF parser failed on this file: {exc}", code="extractor_pdf_error"
        ) from exc
    return tuple((number, page) for number, page in enumerate(pages, start=1) if page)


def _extract_pdf(data: bytes, *, headings: bool = False, ocr: bool = False) -> str:
    """`pdf`: page text via `pdfplumber`, in page order, pages separated by a
    blank line.

    The gate is this import, not a membership test: without the `[extract]`
    extra the very same typed rejection is raised as for the types that ship
    no parser at all. Text is returned VERBATIM — no Unicode normalization,
    matching `md`/`txt` passthrough; normalizing would edit source content,
    and NFC folding belongs to filenames and titles, not to document bodies.

    `pdfplumber` was chosen on ONE measured property (2026-08-21,
    docs/2026-08-21-g2-pdf-extraction-measurement.md): on a real requirement
    table it keeps label and value on the same line, where pypdf, pdfminer.six
    and pymupdf each emit all labels then all values. Re-pairing those is
    guesswork, and in a requirements document a wrong pairing looks right.
    """
    pages = _pdf_pages(data, headings, ocr)
    text = _PDF_PAGE_SEPARATOR.join(page for _, page in pages)
    if not text:
        raise ExtractionError(
            "the PDF yielded no text on any page; a scanned or image-only "
            "document needs OCR, which this registry does only behind the "
            "optional 'ocr' group and only when asked",
            code="extractor_empty_pdf",
        )
    # After the parse, not before: a run that produced no text has nothing to
    # be lossy about, and warning there would just add noise to a failure.
    warnings.warn(_PDF_LOSSY_WARNING, ExtractionWarning, stacklevel=3)
    return text


def _convert_bytes(source: bytes, to: str, format: str, extra_args: Sequence[str]) -> str:
    """The one converter call, isolated so the seam above it is testable.

    Separated for a reason beyond tidiness: every test of the seam's behaviour
    would otherwise need the binary present and a real office document, which
    would make the seam's own logic untestable on a machine without the extra.
    The conversion itself is covered by the frozen-text fixtures instead.

    THE INPUT GOES THROUGH A FILE, NOT THROUGH THE TEXT ENTRY POINT. Every
    format here is a binary container, and the converter's text entry point
    takes an `encoding` because it treats its source as text -- which corrupts
    a zip. Measured: a hand-laid `.xlsx` that pandoc reads correctly from disk
    fails through the text path with `Failed to unpack XLSX archive: not enough
    bytes`. A `.docx` of the same shape happened to survive, which is what
    makes this worth writing down: the defect is SILENT for some inputs and
    fatal for others, so "it worked on the file I tried" is not evidence here.

    The temporary directory is removed on every path, including the failure
    one, and nothing outside it is written.
    """
    import pypandoc

    from ._pandoc import converter_path

    with tempfile.TemporaryDirectory() as staging:
        staged = Path(staging) / f"input.{format}"
        staged.write_bytes(source)
        with converter_path():
            return str(
                pypandoc.convert_file(str(staged), to, format=format, extra_args=list(extra_args))
            )


def _extract_office(suffix: str, data: bytes) -> str:
    """The five office rows, converted through the vendored binary.

    Shaped after `_extract_pdf`: the gate is an import probe rather than a
    membership test, third-party failures are wrapped rather than leaked, empty
    output is refused rather than persisted, and the lossiness is stated after
    the parse rather than before it.
    """
    try:
        import pypandoc  # noqa: F401
    except ImportError as exc:
        raise _extra_missing(suffix) from exc

    spreadsheet = suffix == ".xlsx"
    writer = _SPREADSHEET_WRITER if spreadsheet else _PANDOC_WRITER
    args = _SPREADSHEET_ARGS if spreadsheet else _PANDOC_ARGS
    try:
        text = _convert_bytes(data, writer, _PANDOC_FORMATS[suffix], args)
    except ExtractionError:
        raise
    except Exception as exc:  # noqa: BLE001 - third-party converter, wrapped never leaked
        raise ExtractionError(
            f"the converter failed on this {suffix} file: {exc}",
            code="extractor_convert_error",
        ) from exc

    text = text.strip()
    if not text:
        raise ExtractionError(
            f"the converter returned no text for this {suffix} file; refused "
            "rather than persisted as an empty concept",
            code="extractor_empty_conversion",
        )

    if spreadsheet:
        text = _drop_converter_decimals(text, data)

    # After the parse, not before: a run that produced no text has nothing to
    # be lossy about, and warning there would just add noise to a failure.
    warnings.warn(_OFFICE_LOSSY_WARNING, ExtractionWarning, stacklevel=3)
    return text


def _shared_strings(data: bytes) -> frozenset[str]:
    """Every literal in a workbook's shared string table, or nothing.

    Read for one purpose: to tell a NUMBER from TEXT THAT LOOKS LIKE ONE. The
    converter renders a numeric cell as a double, so an integral value arrives
    as `5647500.0` -- and a text cell reading `92.0` arrives as `92.0` too. The
    output alone cannot separate them, and rewriting on the output alone would
    silently edit somebody's authored text.

    Shared strings are the only text the converter recovers from a sheet at
    all: an inline string (`t="inlineStr"`) is read as an EMPTY cell, measured
    while the first xlsx fixture was built (`tests/fixtures/README.md`). So a
    `<digits>.0` that is not in this set did not come from text.

    Every failure returns the empty set, which makes the rewrite a no-op rather
    than a guess: a workbook this cannot read keeps its converter decimals.
    """
    try:
        with zipfile.ZipFile(io.BytesIO(data)) as archive:
            raw = archive.read("xl/sharedStrings.xml")
        root = ElementTree.fromstring(raw)
    except (KeyError, OSError, zipfile.BadZipFile, ElementTree.ParseError):
        return frozenset()
    return frozenset(
        "".join(node.text or "" for node in item.iter(f"{{{_SSML}}}t")) for item in root
    )


def _drop_converter_decimals(text: str, data: bytes) -> str:
    """Undo the converter's `N.0` on cells the workbook stores as integers.

    Cell-scoped and never applied to prose: the pattern is anchored between two
    unescaped pipes, so only a cell whose ENTIRE content is an integer with a
    trailing `.0` is rewritten, and only when that same literal is absent from
    the shared string table.
    """
    literals = _shared_strings(data)

    def rewrite(match: re.Match[str]) -> str:
        digits = match.group(2)
        if f"{digits}.0" in literals:
            return match.group(0)
        return f"|{match.group(1)}{digits}{match.group(3)}"

    return _INTEGRAL_CELL.sub(rewrite, text)


_CORE_EXTRACTORS: dict[str, Callable[[bytes], str]] = {
    ".md": _extract_passthrough,
    ".txt": _extract_passthrough,
    ".csv": _extract_csv,
    ".json": _extract_json,
    ".html": _extract_html,
    ".htm": _extract_html,
}

# Types the `[extract]` extra ships a parser for. Kept separate from the core
# registry so "adds no runtime dependency" stays readable at a glance.
_OPTIONAL_EXTRACTORS: dict[str, Callable[[bytes], str]] = {
    ".pdf": _extract_pdf,
    **{suffix: functools.partial(_extract_office, suffix) for suffix in _PANDOC_FORMATS},
}


# --- provenance: a character range of the extracted text -> a place in the
# original document ---------------------------------------------------------
#
# `source_offset` alone is a position in OUR extraction, so following it back
# needs the corpus directory, the extractor and its exact version -- none of
# which a bundle carries. A unit table is that mapping, saved AT EXTRACTION
# where the two are known to agree, rather than guessed afterwards from text
# whose page breaks are gone.
#
# THE UNIT IS PER FORMAT AND IS NAMED, never assumed:
#
#   pages  a PDF page number, from the document itself.
#   rows   a spreadsheet row, within the sheet named by `scope_of`.
#   lines  a line of the EXTRACTED text. For `md`/`txt` that text is the
#          dropped file, so the number is the original's own line; for the
#          converted formats it is not, and the key says `lines` rather than
#          `paragraphs` for exactly that reason. Measured on the five K2
#          `.docx` documents: `<w:p>` counts 108/27/65/176/57 against
#          converted-markdown line counts 75/33/67/144/63 -- not one pair
#          agrees, so a `paragraphs` key would name a number the original does
#          not have.
#
# The heading a spreadsheet's sheet becomes, as the converter writes it:
# `## <sheet name> {#sheet-<n>}`. Anchored to the line start so a pipe cell
# containing a `#` cannot be read as a sheet.
_SHEET_HEADING = re.compile(r"^#{1,6} (?P<name>.*?) \{#sheet-\d+\}$")

# A line the converter wrote as part of a pipe table. Whether one of them is
# the table's SEPARATOR is decided by POSITION, never by content: an empty
# spreadsheet row renders as `|  |  |` and a separator as `|----|----|`, and
# every content rule that tells those apart also swallows a data row that
# happens to hold only dashes. Measured on the K2 price sheet: a content rule
# ate 8 empty rows and reported the sheet's last row as 92 against a workbook
# that says 100.
_TABLE_LINE = "|"


@dataclass(frozen=True)
class SourceUnits:
    """Where in the ORIGINAL each stretch of the extracted text came from.

    `starts[i]` is the character offset in the extracted text at which unit
    `numbers[i]` begins, and `scopes[i]` is the sheet that unit belongs to (or
    `None` for a format that has no sheets). The three tuples are parallel and
    `starts` ascends, which is what lets `covering` be a bisection rather than
    a scan.

    `numbers` is separate from the index on purpose. A PDF page that yielded no
    text is not in this table, and a pipe table's separator line is a row of
    nothing -- in both cases the position in the table and the number in the
    original have already parted company, and an index standing in for a number
    is the off-by-one this whole object exists to prevent.
    """

    unit: str
    starts: tuple[int, ...]
    numbers: tuple[int, ...]
    scopes: tuple[str | None, ...] = ()

    def __post_init__(self) -> None:
        if len(self.starts) != len(self.numbers):
            raise ValueError("a unit table needs one number per start offset")
        if self.scopes and len(self.scopes) != len(self.starts):
            raise ValueError("a unit table needs one scope per start offset, or none at all")

    def _index(self, offset: int) -> int:
        """The table row covering `offset`, clamped to the table's own ends."""
        low, high = 0, len(self.starts) - 1
        while low < high:
            middle = (low + high + 1) // 2
            if self.starts[middle] <= offset:
                low = middle
            else:
                high = middle - 1
        return low

    def covering(self, start: int, end: int) -> tuple[int, int]:
        """The first and last original unit the half-open `[start, end)` touches.

        `end` is exclusive, so a range ending exactly where the next unit
        begins does not claim that unit -- a segment that stops at a page
        boundary is on the page it was written on.
        """
        if not self.starts:
            raise ValueError("an empty unit table locates nothing")
        first = self._index(start)
        last = self._index(max(start, end - 1))
        return self.numbers[first], self.numbers[last]

    def scope_of(self, offset: int) -> str | None:
        """The sheet `offset` falls in, or `None` for a format without sheets."""
        if not self.scopes:
            return None
        return self.scopes[self._index(offset)]

    def scopes_covering(self, start: int, end: int) -> tuple[str | None, ...]:
        """Every distinct scope the range touches, in order, without repeats."""
        if not self.scopes:
            return ()
        first = self._index(start)
        last = self._index(max(start, end - 1))
        seen: list[str | None] = []
        for scope in self.scopes[first : last + 1]:
            if not seen or seen[-1] != scope:
                seen.append(scope)
        return tuple(seen)


def _line_units(text: str) -> SourceUnits:
    starts: list[int] = []
    offset = 0
    for line in text.split("\n"):
        starts.append(offset)
        offset += len(line) + 1
    return SourceUnits("lines", tuple(starts), tuple(range(1, len(starts) + 1)))


def _pdf_units(data: bytes, headings: bool, ocr: bool) -> SourceUnits:
    starts: list[int] = []
    numbers: list[int] = []
    offset = 0
    for number, page in _pdf_pages(data, headings, ocr):
        starts.append(offset)
        numbers.append(number)
        offset += len(page) + len(_PDF_PAGE_SEPARATOR)
    return SourceUnits("pages", tuple(starts), tuple(numbers))


def _spreadsheet_units(text: str) -> SourceUnits | None:
    """Sheet and row for a converted spreadsheet, or `None` if it is not one.

    The converter writes one heading per sheet and then one pipe-table line per
    source row, with a separator line after the first. Row numbering therefore
    restarts at every heading and skips that one line by POSITION.

    The row number is the ORIGINAL sheet's, and that holds exactly as far as
    one converted line per `<row>` element holds. Measured on the two K2
    spreadsheets and both fixtures: 39 rows for 39, 100 for 100, 4 for 4, 6 for
    6 and 3 for 3 -- every one contiguous from row 1. A sheet whose XML omits a
    row entirely would number from the converted table instead, and nothing
    here can see that.
    """
    starts: list[int] = []
    numbers: list[int] = []
    scopes: list[str | None] = []
    sheet: str | None = None
    seen = 0
    offset = 0
    for line in text.split("\n"):
        heading = _SHEET_HEADING.match(line)
        if heading is not None:
            sheet = heading.group("name")
            seen = 0
        elif sheet is not None and line.startswith(_TABLE_LINE):
            seen += 1
            # The SECOND table line of a sheet is the separator the converter
            # writes under the header, and it is a row of no spreadsheet. Every
            # line after it is one row further on than its position suggests.
            if seen != 2:
                starts.append(offset)
                numbers.append(seen if seen == 1 else seen - 1)
                scopes.append(sheet)
        offset += len(line) + 1
    if not starts:
        return None
    return SourceUnits("rows", tuple(starts), tuple(numbers), tuple(scopes))


def source_units(
    filename: str, data: bytes, text: str, *, pdf_headings: bool = False, ocr: bool = False
) -> SourceUnits | None:
    """The unit table for one dropped file, or `None` when it has none.

    `text` must be what `extract_text` returned for these exact bytes: the
    table indexes that string, and a table built against a different rendering
    would point a consumer at the wrong place with full confidence.

    `None` is a measurement, not a failure -- a spreadsheet the converter wrote
    no table for has no rows to name, and the caller writes the address without
    a locator rather than inventing one.
    """
    suffix = Path(filename).suffix.lower()
    if suffix == ".pdf":
        return _pdf_units(data, pdf_headings, ocr)
    if suffix == ".xlsx":
        return _spreadsheet_units(text)
    if suffix in _CORE_EXTRACTORS or suffix in _PANDOC_FORMATS:
        return _line_units(text)
    return None


def extract_text(
    filename: str,
    data: bytes,
    *,
    renderer: Callable[[str], str] | None = None,
    pdf_headings: bool = False,
    ocr: bool = False,
) -> str:
    """Convert one dropped file's bytes to OKF concept text, dispatched by type.

    `filename` supplies the extension (case-insensitive); `data` is the raw
    bytes. A core stdlib type is extracted; a `[extract]`-gated binary type
    without the extra, and any unregistered extension, fail fast with a typed
    :class:`ExtractionError`. Extracting a `pdf` also emits an
    :class:`ExtractionWarning`: drawn content has no text to recover.

    `renderer`, when given, is applied to the EXTRACTED TEXT before it is
    returned -- after extraction, never instead of it, so a renderer never has
    to re-implement a reader and the two cannot drift. It is a plain callable
    rather than anything profile-shaped ON PURPOSE: this module is the
    extraction registry and must not import the contract layer, or the
    dependency would run backwards and the registry would stop standing on its
    own. Resolving a profile's NAMED renderer to a function is the caller's
    job, in the layer that already holds the profile.

    The default is identity, which is what keeps every existing byte-pinned
    golden byte-pinned.

    `pdf_headings` and `ocr` are PDF-only and both default to off. They are
    branched on here rather than expressed as two more registry rows because
    the registry's contract is `bytes -> str`: a row per option combination
    would be four rows for one reader, and a reader chosen by a suffix lookup
    that also has to consult two flags is not a lookup. A non-PDF caller
    passing either argument gets today's behaviour, silently, which is correct
    -- the options describe a reader, not a policy for the run.
    """
    suffix = Path(filename).suffix.lower()
    extractor = _CORE_EXTRACTORS.get(suffix) or _OPTIONAL_EXTRACTORS.get(suffix)
    if extractor is not None:
        if suffix == ".pdf" and (pdf_headings or ocr):
            text = _extract_pdf(data, headings=pdf_headings, ocr=ocr)
        else:
            text = extractor(data)
        return renderer(text) if renderer is not None else text
    if suffix in _UNPARSED_OPTIONAL_EXTENSIONS:
        raise _extra_missing(suffix)
    raise ExtractionError(
        f"no extractor is registered for file extension {suffix!r} ({filename!r})",
        code="extractor_unknown",
    )