pagepiper/scripts/shelve_xlsx.py

# scripts/shelve_xlsx.py
"""
cf-orch task: pagepiper/shelve_xlsx

Extracts rows from an Excel .xlsx workbook, stores chunks in SQLite, and
(if Ollama is configured) generates embeddings in the sqlite-vec store.

Chunking strategy:
  - One chunk per sheet if the sheet has <= ROWS_PER_CHUNK rows.
  - Larger sheets split into row-window chunks, with the header row (the
    first row) repeated at the top of every window so each chunk stays
    self-describing for BM25/embedding retrieval.

Rows are serialised as pipe-delimited cells, matching the table-serialization
style used by the DOCX/ODT shelvers.

Entry point:
    python scripts/shelve_xlsx.py --doc-id X --file-path Y --db-path Z --vec-db-path W
"""
from __future__ import annotations

import logging
import os
import sqlite3
from dataclasses import dataclass

logger = logging.getLogger("pagepiper.shelve_xlsx")

EMBED_BATCH_SIZE = 64
ROWS_PER_CHUNK = 200


@dataclass
class _Chunk:
    page_number: int
    text: str
    source: str
    word_count: int


def _row_to_text(row: tuple) -> str:
    cells = [str(c).strip() if c is not None else "" for c in row]
    return " | ".join(cells) if any(cells) else ""


def _sheet_to_chunks(sheet_name: str, rows: list[tuple], start_page: int) -> list[_Chunk]:
    """Chunk one sheet's rows into one-or-more page-numbered chunks."""
    text_rows = [_row_to_text(r) for r in rows]
    text_rows = [r for r in text_rows if r]
    if not text_rows:
        return []

    header = text_rows[0]
    body = text_rows[1:]

    chunks: list[_Chunk] = []
    if len(body) == 0:
        lines = [f"Sheet: {sheet_name}", header]
        text = "\n".join(lines)
        chunks.append(_Chunk(start_page, text, "sheet", len(text.split())))
        return chunks

    for i in range(0, len(body), ROWS_PER_CHUNK):
        window = body[i : i + ROWS_PER_CHUNK]
        lines = [f"Sheet: {sheet_name}", header] + window
        text = "\n".join(lines)
        chunks.append(_Chunk(start_page + len(chunks), text, "sheet", len(text.split())))

    return chunks


def _extract_chunks(file_path: str) -> list[_Chunk]:
    import openpyxl

    wb = openpyxl.load_workbook(file_path, read_only=True, data_only=True)
    try:
        chunks: list[_Chunk] = []
        for sheet_name in wb.sheetnames:
            ws = wb[sheet_name]
            rows = list(ws.iter_rows(values_only=True))
            next_page = len(chunks) + 1
            chunks.extend(_sheet_to_chunks(sheet_name, rows, next_page))
        return chunks
    finally:
        wb.close()


def _update_status(
    conn: sqlite3.Connection,
    doc_id: str,
    status: str,
    page_count: int | None = None,
    error_msg: str | None = None,
) -> None:
    if page_count is not None:
        conn.execute(
            "UPDATE documents SET status=?, page_count=?, updated_at=datetime('now') WHERE id=?",
            [status, page_count, doc_id],
        )
    elif error_msg is not None:
        conn.execute(
            "UPDATE documents SET status=?, error_msg=?, updated_at=datetime('now') WHERE id=?",
            [status, error_msg, doc_id],
        )
    else:
        conn.execute(
            "UPDATE documents SET status=?, updated_at=datetime('now') WHERE id=?",
            [status, doc_id],
        )
    conn.commit()


def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None:
    """Run the full shelve pipeline for one XLSX. Called by cf-orch or BackgroundTasks."""
    conn: sqlite3.Connection | None = None
    try:
        conn = sqlite3.connect(db_path, timeout=30)
        conn.execute("PRAGMA journal_mode = WAL")
        conn.execute("PRAGMA foreign_keys = ON")
        _update_status(conn, doc_id, "processing")

        logger.info("Extracting sheets from %s", file_path)
        chunks = _extract_chunks(file_path)
        logger.info("Extracted %d chunks", len(chunks))

        from scripts.text_clean import clean_paragraph
        conn.execute("DELETE FROM page_chunks WHERE doc_id=?", [doc_id])
        chunk_rows: list[tuple[str, int, str]] = []
        for chunk in chunks:
            cleaned = clean_paragraph(chunk.text)
            if not cleaned:
                continue
            row = conn.execute(
                """INSERT INTO page_chunks(doc_id, page_number, text, source, word_count)
                   VALUES (?,?,?,?,?) RETURNING id""",
                [doc_id, chunk.page_number, cleaned, chunk.source, len(cleaned.split())],
            ).fetchone()
            chunk_rows.append((row[0], chunk.page_number, cleaned))
        conn.commit()

        from app.config import get_llm_config
        llm_cfg = get_llm_config()
        if llm_cfg and chunks:
            try:
                logger.info("Embedding %d chunks", len(chunks))
                from circuitforge_core.llm import LLMRouter
                from circuitforge_core.vector.sqlite_vec import LocalSQLiteVecStore

                router = LLMRouter(llm_cfg)
                embed_dims = int(os.environ.get("PAGEPIPER_EMBED_DIMS", "1024"))
                vec_store = LocalSQLiteVecStore(
                    db_path=vec_db_path, table="page_vecs", dimensions=embed_dims
                )
                vec_store.delete_where({"doc_id": doc_id})

                texts = [text for _, _, text in chunk_rows]
                vectors: list[list[float]] = []
                for i in range(0, len(texts), EMBED_BATCH_SIZE):
                    vectors.extend(router.embed(texts[i : i + EMBED_BATCH_SIZE]))

                for (chunk_id, page_number, _), vector in zip(chunk_rows, vectors):
                    vec_store.upsert(
                        entry_id=chunk_id,
                        vector=vector,
                        metadata={"doc_id": doc_id, "page_number": page_number},
                    )
                logger.info("Stored %d embeddings", len(vectors))
            except Exception as embed_exc:
                logger.warning(
                    "Embedding skipped for doc %s — BM25 only (reason: %s)",
                    doc_id, embed_exc,
                )

        _update_status(conn, doc_id, "ready", page_count=len(chunks))
        logger.info("Shelve complete for doc %s (%d chunks)", doc_id, len(chunks))

    except Exception as exc:
        logger.error("Shelve failed for doc %s: %s", doc_id, exc, exc_info=True)
        if conn is not None:
            try:
                _update_status(conn, doc_id, "error", error_msg=str(exc))
            except Exception:
                logger.warning("Could not write error status for doc %s", doc_id)
        raise
    finally:
        if conn is not None:
            conn.close()


if __name__ == "__main__":
    import argparse

    logging.basicConfig(level=logging.INFO)

    parser = argparse.ArgumentParser(
        description="Shelve an Excel .xlsx workbook (cf-orch task entry point)"
    )
    parser.add_argument("--doc-id", required=True)
    parser.add_argument("--file-path", required=True)
    parser.add_argument("--db-path", required=True)
    parser.add_argument("--vec-db-path", required=True)
    a = parser.parse_args()
    run(
        doc_id=a.doc_id,
        file_path=a.file_path,
        db_path=a.db_path,
        vec_db_path=a.vec_db_path,
    )