# 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, )