yau-plant-assistant/api/convert.py

"""Uploaded file -> Markdown, before anything is chunked or embedded.

WHY THERE IS A CONVERSION STEP AT ALL. The raw PDF is never what the assistant
reads - it never was. Retrieval only ever sees `doc_chunks.chunk_text`. What
this module adds is that the extracted text becomes a thing a person can LOOK
AT before approving it. If a table comes out as garbage, the reviewer sees the
garbage and rejects the document, instead of an operator discovering it months
later inside a citation.

WHAT THESE CONVERTERS DO AND DO NOT DO. They extract text, not layout. A
multi-column page interleaves. A merged-cell spreadsheet flattens. A scanned
page yields nothing at all, and is refused rather than stored empty.

That limitation is acceptable HERE, and only here, because a person reads the
converted text before it can be cited - the same safety net the design already
required for the document header. It would not be acceptable in a pipeline that
published without review.

REPLACING THEM. Everything below the `convert()` boundary is swappable: return
Markdown from bytes, raise ConversionError when you cannot. Nothing outside this
module knows which library did the work, so a better converter is a change to
this file alone. If the real documents turn out to be scans, that is the
conversation to have - OCR is the missing capability, and nothing else here
changes.
"""

from __future__ import annotations

import io
import re
from dataclasses import dataclass

# A scanned page yields a handful of stray characters, not nothing - so an
# emptiness check has to have a floor above zero. Below this, across the whole
# document, we refuse rather than store a blank document that a reviewer might
# approve without noticing there is nothing in it.
_MIN_CHARS = 200

SUPPORTED = {".pdf", ".docx", ".xlsx", ".xlsm", ".md", ".txt"}


class ConversionError(Exception):
    """Conversion failed, or produced something not worth reviewing."""


@dataclass(frozen=True)
class Converted:
    markdown: str
    converter: str          # recorded on the upload row - which code produced this
    page_count: int | None


def _clean(text: str) -> str:
    """Collapse the whitespace damage that text extraction always leaves.

    Not cosmetic: runs of blank lines and trailing spaces change where a chunker
    splits, so the same document converted twice should look the same.
    """
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    text = re.sub(r"[ \t]+\n", "\n", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


# A numbered section heading: "1. Purpose", "3.2 Pump control", "4.1.2 Alarms".
#
# THE HARD PART. A numbered STEP in a procedure looks identical to a numbered
# HEADING - both are "<n>. <text>". Getting this wrong in the dangerous
# direction would break a step sequence apart, which is the one thing chunking
# must never do. So the test is deliberately narrow: a heading is SHORT, has no
# terminal punctuation, and is a handful of words. "1. Purpose" passes.
# "1. Open the isolation valve and confirm zero pressure." does not - it is
# long, and it ends in a full stop.
_NUMBERED_HEADING = re.compile(r"^(\d+(?:\.\d+)*)[.)]?\s+(\S.*)$")
# A heading in capitals: "SECTION 4 - ALARM PHILOSOPHY". Requires a SPACE, so
# it needs at least two words. Found on the first real document: without that,
# "SCADA", "PLC-001" and every "WRPS-PRO-001" in a reference list became
# headings - which is worse than missing a heading, because each one cut a real
# section short and re-titled the remainder with something meaningless. A
# citation reading "SCADA" helps nobody.
_CAPS_HEADING = re.compile(r"^[A-Z][A-Z0-9&/(),.'-]*(?: +[A-Z0-9&/(),.'-]+)+$")

# "11 August 2026" is not section 11. Same document: the revision history dates
# parsed as numbered headings and split the header block into fragments.
_DATE_LIKE = re.compile(
    r"^\d{1,2}[ .-]+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)",
    re.IGNORECASE,
)

_HEADING_MAX_CHARS = 80
_HEADING_MAX_WORDS = 10


def _looks_like_heading(line: str) -> str | None:
    """Return the heading text, or None. Conservative by design - see above."""
    line = line.strip()
    if not line or len(line) > _HEADING_MAX_CHARS:
        return None
    if line.endswith((".", ";", ":", ",")):
        return None

    if _DATE_LIKE.match(line):
        return None

    numbered = _NUMBERED_HEADING.match(line)
    if numbered:
        title = numbered.group(2).strip()
        if title and title[0].isupper() and len(title.split()) <= _HEADING_MAX_WORDS:
            return line
        return None

    if (_CAPS_HEADING.match(line)
            and 2 <= len(line.split()) <= _HEADING_MAX_WORDS):
        return line
    return None


def _structure(lines: list[str]) -> str:
    """Flat PDF text lines -> Markdown with headings and real paragraphs.

    WHY THIS EXISTS. pypdf emits one line per line of the PDF and NO blank
    lines at all - a real document came through as 416 lines, none of them
    blank. The chunker splits sections on Markdown headings and paragraphs on
    blank lines, so without this an entire document is one untitled section and
    one paragraph: a single 18,000-character chunk, one embedding vector for
    eight pages, and every citation reading "(untitled), page 1".

    Paragraph reflow uses line width. PDF body text wraps at a consistent
    measure, so a line noticeably shorter than the running width is the LAST
    line of its paragraph. It is a heuristic and it will occasionally join two
    paragraphs or split one - which is tolerable, because a person reads this
    text before the document can be cited, and because chunk_section now has a
    hard ceiling that does not depend on getting paragraphs right.
    """
    body = [ln for ln in lines if ln.strip() and not _looks_like_heading(ln)]
    widths = sorted(len(ln.rstrip()) for ln in body)
    # Median width of body lines, with a floor so a very short document does
    # not produce a nonsense threshold.
    typical = widths[len(widths) // 2] if widths else 0
    short_line = max(int(typical * 0.75), 30)

    out: list[str] = []
    para: list[str] = []

    def flush() -> None:
        if para:
            out.append(" ".join(para))
            para.clear()

    for raw in lines:
        line = raw.strip()
        if not line:
            flush()
            continue
        heading = _looks_like_heading(line)
        if heading:
            flush()
            out.append(f"## {heading}")
            continue
        para.append(line)
        if len(line) < short_line:
            # Short line = end of a wrapped paragraph.
            flush()
    flush()

    return "\n\n".join(out)


def _from_pdf(data: bytes) -> Converted:
    from pypdf import PdfReader

    try:
        reader = PdfReader(io.BytesIO(data))
    except Exception as exc:
        raise ConversionError(f"not a readable PDF: {exc}") from exc

    if reader.is_encrypted:
        # Refuse rather than guess at an empty password. A locked document that
        # silently converts to nothing is the worst outcome here.
        raise ConversionError(
            "this PDF is encrypted - remove the protection and upload it again"
        )

    parts: list[str] = []
    for number, page in enumerate(reader.pages, start=1):
        try:
            text = page.extract_text() or ""
        except Exception:
            # One bad page must not lose the other ninety. Mark it so the
            # reviewer can see exactly what is missing.
            parts.append(f"<!-- page {number}: could not be extracted -->")
            continue
        if text.strip():
            # Structure each page separately: the width heuristic in
            # _structure() is per-page because a landscape table page and a
            # portrait text page have different measures, and mixing them
            # makes the threshold meaningless for both.
            parts.append(f"<!-- page {number} -->")
            parts.append(_structure(text.splitlines()))

    return Converted(
        markdown=_clean("\n\n".join(parts)),
        converter="pypdf",
        page_count=len(reader.pages),
    )


def _from_docx(data: bytes) -> Converted:
    import docx

    try:
        document = docx.Document(io.BytesIO(data))
    except Exception as exc:
        raise ConversionError(f"not a readable .docx: {exc}") from exc

    lines: list[str] = []
    for para in document.paragraphs:
        text = para.text.strip()
        if not text:
            continue
        # Word's built-in heading styles are the one piece of structure that
        # survives reliably, and headings matter: section_title is what a
        # citation falls back to.
        style = (para.style.name or "").lower() if para.style else ""
        if style.startswith("heading"):
            level = "".join(c for c in style if c.isdigit()) or "1"
            lines.append(f"{'#' * min(int(level), 6)} {text}")
        elif style.startswith("title"):
            lines.append(f"# {text}")
        else:
            lines.append(text)

    for index, table in enumerate(document.tables, start=1):
        lines.append(f"\n<!-- table {index} -->")
        for row in table.rows:
            cells = [c.text.strip().replace("|", "\\|") for c in row.cells]
            lines.append("| " + " | ".join(cells) + " |")

    return Converted(
        markdown=_clean("\n\n".join(lines)),
        converter="python-docx",
        page_count=None,
    )


def _from_xlsx(data: bytes) -> Converted:
    from openpyxl import load_workbook

    try:
        # read_only keeps a large workbook from being held in memory twice, and
        # data_only takes the cached VALUE of a formula rather than the formula
        # text. A cell reading "=SUM(B2:B9)" is not something to embed - and if
        # the workbook was never opened in Excel there is no cached value, so
        # that cell comes through empty. Say so in the output rather than
        # letting it look like a blank cell.
        book = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
    except Exception as exc:
        raise ConversionError(f"not a readable spreadsheet: {exc}") from exc

    lines: list[str] = []
    for sheet in book.worksheets:
        lines.append(f"## {sheet.title}")
        for row in sheet.iter_rows(values_only=True):
            cells = ["" if v is None else str(v).strip().replace("|", "\\|")
                     for v in row]
            if not any(cells):
                continue
            lines.append("| " + " | ".join(cells) + " |")
        lines.append("")
    book.close()

    return Converted(
        markdown=_clean("\n".join(lines)),
        converter="openpyxl",
        page_count=None,
    )


def _from_text(data: bytes) -> Converted:
    try:
        text = data.decode("utf-8")
    except UnicodeDecodeError:
        text = data.decode("latin-1")
    return Converted(markdown=_clean(text), converter="passthrough", page_count=None)


_CONVERTERS = {
    ".pdf": _from_pdf,
    ".docx": _from_docx,
    ".xlsx": _from_xlsx,
    ".xlsm": _from_xlsx,
    ".md": _from_text,
    ".txt": _from_text,
}


def convert(filename: str, data: bytes) -> Converted:
    """Convert an uploaded file to Markdown, or raise ConversionError.

    Dispatch is on the extension, which is a claim the uploader made about the
    file. The libraries below all fail loudly on a mismatch, so a .docx renamed
    to .pdf raises rather than producing plausible rubbish.
    """
    suffix = "." + filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
    handler = _CONVERTERS.get(suffix)
    if handler is None:
        raise ConversionError(
            f"{suffix or 'no extension'} is not supported - "
            f"expected one of {', '.join(sorted(SUPPORTED))}"
        )

    result = handler(data)

    if len(result.markdown) < _MIN_CHARS:
        # The common cause by far is a scanned document: a picture of text,
        # which text extraction cannot read. Name that explicitly, because
        # "conversion produced nothing" sends somebody looking for a bug in the
        # upload instead of at the file.
        raise ConversionError(
            f"only {len(result.markdown)} characters of text came out of this "
            f"file. If it is a scan or a photograph, no text can be extracted "
            f"from it - it needs OCR, which this converter does not do. "
            f"Nothing has been added to the library."
        )

    return result