chatbot_v3/src/application/ingestion/tabular.py

"""Shared row-to-chunk rendering for every tabular source (ADR-0004, ADR-0018).

A table row is an atomic structural unit regardless of the container it
arrived in -- a docx table, an xlsx sheet, or a csv file -- so one renderer
serves all three and a Q&A sheet needs no special case against a branch
directory:

    q: ...                 ردیف: 1
    a: ...                 استان: اردبیل
                           شعبه: پارس آباد

The header rules exist because real tables are not uniform. Of the tables in
the sample corpus, some carry a header row, one is a bare list of values with
no header at all, and one is page decoration. The guiding constraint is
therefore: **never invent structure that is not provably there, and never
discard a row.** A wrongly-detected header turns every chunk into nonsense
(`80: 70`), which is worse than an unlabeled row.
"""

from collections.abc import Iterable, Sequence

from src.application.ingestion.normalization import normalize_persian_text

Row = Sequence[str]

# A header cell is a label, not a sentence.
MAX_HEADER_CELL_LENGTH = 80

# How much shorter a header cell must be than the column beneath it before
# length alone is taken as evidence of a header (the `q`/`a` case, where
# one-character labels sit above paragraph-long answers).
_HEADER_LENGTH_RATIO = 3.0


def clean_cell(value: object) -> str:
    """Normalize a cell to text; None and blank cells become the empty string.

    `object` rather than a union: a spreadsheet cell holds whatever the
    workbook stored -- str, int, float, bool, datetime, a formula error -- and
    every one of them is handled the same way, by rendering it.
    """
    if value is None:
        return ""
    return normalize_persian_text(str(value))


def _is_numeric(text: str) -> bool:
    return bool(text) and text.replace(",", "").replace(".", "").replace("-", "").isdigit()


def _looks_like_title_row(row: Row) -> bool:
    """A merged title spanning the sheet resolves to one value, or repeats it."""
    populated = [cell for cell in row if cell]
    return len(populated) < 2 or len(set(populated)) == 1


def has_header(rows: Sequence[Row]) -> bool:
    """Whether the first row labels the columns beneath it.

    Decided by comparing row 0 against the column below it, not by how row 0
    looks on its own: a label row is *inconsistent* with its data (text above
    numbers, or a short label above long prose), while a data row is
    consistent with the rows that follow. This is the test `csv.Sniffer`
    uses, and it is a property of the table rather than a pattern borrowed
    from one document.
    """
    if len(rows) < 2:
        return False

    candidate, data = rows[0], rows[1:]
    if not all(cell for cell in candidate[: len(data[0])] if cell) and _looks_like_title_row(
        candidate
    ):
        return False
    if any(len(cell) > MAX_HEADER_CELL_LENGTH for cell in candidate):
        return False

    for index, label in enumerate(candidate):
        if not label:
            continue
        column = [row[index] for row in data if index < len(row) and row[index]]
        if not column:
            continue

        # A text label above a numeric column.
        if not _is_numeric(label) and all(_is_numeric(value) for value in column):
            return True

        # A short label above a column of much longer values.
        mean_length = sum(len(value) for value in column) / len(column)
        if mean_length > len(label) * _HEADER_LENGTH_RATIO:
            return True

    return False


def strip_title_rows(rows: Sequence[Row]) -> Sequence[Row]:
    """Drop leading merged-title and blank rows.

    Structural, not a content guess: these rows are the artifact of a merged
    range spanning the sheet width, so they carry one value across many cells.
    Only *leading* rows are dropped, so no data row is ever lost.
    """
    start = 0
    while start < len(rows) and _looks_like_title_row(rows[start]):
        start += 1
    return rows[start:]


def _column_name(header: Row, index: int) -> str:
    """Return a header label, falling back positionally past the header width."""
    if index < len(header) and header[index]:
        return header[index]
    return f"column_{index + 1}"


def _dedupe_horizontal_merge(row: Row) -> list[str]:
    """Collapse the repeats a horizontally merged cell produces.

    Both python-docx and openpyxl report a merged cell once per grid column it
    spans, so an unlabeled row would otherwise repeat the same value.
    """
    collapsed: list[str] = []
    for cell in row:
        if cell and (not collapsed or collapsed[-1] != cell):
            collapsed.append(cell)
    return collapsed


def render_rows(rows: Sequence[Row]) -> list[str]:
    """Render table rows as text, one string per row.

    With a provable header each row becomes `"{header}: {value}"` lines, which
    makes it self-describing. Without one, cells are joined with `" | "` --
    unlabeled, but never mislabeled.
    """
    rows = strip_title_rows(rows)
    if not rows:
        return []

    if not has_header(rows):
        return [text for row in rows if (text := " | ".join(_dedupe_horizontal_merge(row)))]

    header, data = rows[0], rows[1:]
    # A header merged vertically across two rows resolves to the same text in
    # the row below it; that duplicate is the header, not data.
    while data and list(data[0]) == list(header):
        data = data[1:]

    rendered: list[str] = []
    for row in data:
        lines = [
            f"{_column_name(header, index)}: {value}" for index, value in enumerate(row) if value
        ]
        if lines:
            rendered.append("\n".join(lines))
    return rendered


def clean_rows(rows: Iterable[Iterable[object]]) -> list[Row]:
    """Normalize every cell and drop rows that are entirely empty."""
    cleaned: list[Row] = []
    for row in rows:
        values = [clean_cell(cell) for cell in row]
        if any(values):
            cleaned.append(values)
    return cleaned