diff --git a/.env.cloud.example b/.env.cloud.example index 0063c7c..1d1f31a 100644 --- a/.env.cloud.example +++ b/.env.cloud.example @@ -23,10 +23,13 @@ HEIMDALL_ADMIN_TOKEN= # Must match COORDINATOR_PRODUCT_KEYS["pagepiper"] in cf-orch.env on the coordinator COORDINATOR_PAGEPIPER_KEY= -# cf-orch coordinator URL — routes chat/embed calls through managed GPU allocation -# CF_LICENSE_KEY is the auth token sent to the coordinator (same value as COORDINATOR_PAGEPIPER_KEY) -# Leave CF_ORCH_URL blank to skip allocation and hit PAGEPIPER_OLLAMA_URL directly -CF_ORCH_URL= +# GPU server / cf-orch coordinator URL — routes chat/embed calls through managed +# GPU allocation. CF_LICENSE_KEY is the auth token sent to the coordinator (same +# value as COORDINATOR_PAGEPIPER_KEY). Leave blank to skip allocation and hit +# PAGEPIPER_OLLAMA_URL directly. CF_ORCH_URL is still honoured as a legacy alias +# if GPU_SERVER_URL is unset — Paid+ tiers auto-default to orch.circuitforge.tech +# when CF_LICENSE_KEY is present. +GPU_SERVER_URL= CF_LICENSE_KEY= CF_APP_NAME=pagepiper diff --git a/.env.example b/.env.example index 473929f..c975313 100644 --- a/.env.example +++ b/.env.example @@ -6,10 +6,26 @@ PAGEPIPER_BOOKS_DIR=/path/to/your/pdfs # Data directory (SQLite + vector DB stored here) PAGEPIPER_DATA_DIR=data -# Ollama URL — set this to unlock semantic search and RAG chat (BYOK) +# LLM backend — either option (or both) unlocks semantic search and RAG chat. +# +# Option A: direct Ollama URL # PAGEPIPER_OLLAMA_URL=http://localhost:11434 -# PAGEPIPER_CHAT_MODEL=mistral:7b -# PAGEPIPER_EMBED_MODEL=nomic-embed-text +# +# Option B: cf-orch coordinator (resolves service URL via GPU allocation). +# Set CF_ORCH_URL alone — no PAGEPIPER_OLLAMA_URL needed. +# PAGEPIPER_OLLAMA_URL is used as a fallback if cf-orch is unreachable. +# CF_ORCH_URL=http://localhost:7700 +# CF_APP_NAME=pagepiper +# PAGEPIPER_ORCH_SERVICE=ollama # or cf-text for managed transformer inference +# +PAGEPIPER_CHAT_MODEL=mistral:7b +PAGEPIPER_EMBED_MODEL=nomic-embed-text + +# Self-hosted GPU rig URL — set this instead of PAGEPIPER_OLLAMA_URL if you're +# running a cf-orch coordinator (e.g. a home GPU rack). The coordinator resolves +# the actual service URL at allocation time. (CF_ORCH_URL is still honoured as +# a legacy alias if you already have it set.) +# GPU_SERVER_URL=http://localhost:7700 # Forgejo API token — enables the in-app feedback button (files Forgejo issues) # Create a token at https://git.opensourcesolarpunk.com/user/settings/applications diff --git a/.gitignore b/.gitignore index 5908ea1..a79ba22 100644 --- a/.gitignore +++ b/.gitignore @@ -29,3 +29,6 @@ compose.override.yml # Logs and runtime files *.log *.db + +# Playwright MCP scratch dir (test fixtures, screenshots — never commit) +.playwright-mcp/ diff --git a/Dockerfile b/Dockerfile index 4971de2..a66ed9e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,10 +2,15 @@ FROM continuumio/miniconda3:latest WORKDIR /app -# System deps for pytesseract (OCR) and pdfplumber +# System deps for pytesseract (OCR), pdfplumber, and Apple Pages/Numbers +# conversion (libreoffice bundles libetonyek, the only maintained open-source +# parser for Apple's iWork formats — shelve_pages.py / shelve_numbers.py +# shell out to headless soffice for .pages / .numbers respectively) RUN apt-get update && apt-get install -y --no-install-recommends \ tesseract-ocr \ libgl1 \ + libreoffice-writer \ + libreoffice-calc \ && rm -rf /var/lib/apt/lists/* # Install circuitforge-core from sibling directory (compose sets context: ..) diff --git a/README.md b/README.md index 063a321..d0eda93 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ [![License: MIT / BSL 1.1](https://img.shields.io/badge/license-MIT%20%2F%20BSL%201.1-blue)](LICENSE) [![Version](https://img.shields.io/badge/version-v0.1.0-orange)](https://git.opensourcesolarpunk.com/Circuit-Forge/pagepiper/releases) -Self-hosted PDF and EPUB search with BM25 (Best Match 25) full-text indexing and LLM (large language model) synthesis. Drop your documents in, ask a question, get an answer that tells you exactly which page to turn to. +Self-hosted document and spreadsheet search — PDF, EPUB, DOCX, ODT, Apple Pages, XLSX, ODS, and Apple Numbers — with BM25 (Best Match 25) full-text indexing and LLM (large language model) synthesis. Drop your documents in, ask a question, get an answer that tells you exactly which page to turn to. Built for TTRPG (tabletop roleplaying game) players who are tired of ctrl-F'ing through six-hundred-page rulebooks. Works equally well for legal research, technical manuals, academic papers, or any personal document library you want to query in plain language. @@ -18,7 +18,7 @@ No cloud required. Your files stay on your machine. ### Library -![Library view — documents listed with ingest status and page counts](docs/screenshots/01-library.png) +![Library view — documents listed with shelving status and page counts](docs/screenshots/01-library.png) ### Chat with citations @@ -32,7 +32,7 @@ No cloud required. Your files stay on your machine. - **Works without an LLM.** BM25 full-text search runs entirely inside the Docker container. No Ollama, no API key, no GPU required for keyword search. - **Answers cite their sources.** Every LLM response includes the document name and page number it drew from. You can verify or dispute every answer. - **Hybrid search when you want it.** Connect a local Ollama instance to unlock hybrid search — BM25 merged with Agent-ModernColBERT, a late-interaction retriever that scores passages by token-level relevance instead of collapsing your whole question into one vector, so multi-part questions find the right passage even when it doesn't use your exact words. -- **Open ingest pipeline.** The indexing and search layer is MIT-licensed. Add support for new formats, improve the PDF parser, contribute — the community benefits directly. +- **Open shelve pipeline.** The indexing and search layer is MIT-licensed. Add support for new formats, improve the PDF parser, contribute — the community benefits directly. --- @@ -79,10 +79,16 @@ PAGEPIPER_EMBED_MODEL=nomic-embed-text ## Supported Formats -| Format | Ingest | Page-level citations | +| Format | Shelve | Page-level citations | |--------|--------|----------------------| | PDF | Yes | Yes | | EPUB | Yes | Yes (chapter/location) | +| DOCX | Yes | Yes (section/heading) | +| ODT | Yes | Yes (section/heading) | +| Pages | Yes | Yes (section/heading, via LibreOffice) | +| XLSX | Yes | Yes (sheet/row-window) | +| ODS | Yes | Yes (sheet/row-window) | +| Numbers | Yes | Yes (sheet/row-window, via LibreOffice) | --- @@ -120,10 +126,10 @@ Default ports: Web UI `8521`, API `8540`. | Feature | Free | Paid (BYOK) | |---------|------|-------------| -| PDF and EPUB upload | Yes | Yes | +| All supported document/spreadsheet formats upload | Yes | Yes | | Directory scan | Yes | Yes | | BM25 full-text search | Yes | Yes | -| Unlimited local ingestion | Yes | Yes | +| Unlimited local shelving | Yes | Yes | | Hybrid BM25 + vector search | — | Yes (local Ollama) | | LLM synthesis with page citations | — | Yes (local Ollama) | @@ -141,9 +147,13 @@ Pagepiper is developed and hosted at [git.opensourcesolarpunk.com/Circuit-Forge/ Pagepiper uses a split license: -- **MIT:** Document ingest pipeline, BM25 full-text index, library management, EPUB support — the core discovery and retrieval layer. +- **MIT:** Document shelve pipeline, BM25 full-text index, library management, all format support (EPUB/DOCX/ODT/Pages/XLSX/ODS/Numbers) — the core discovery and retrieval layer. - **BSL 1.1 (Business Source License):** Hybrid vector search, LLM synthesis, RAG (retrieval-augmented generation) chat interface — free for personal non-commercial self-hosting; commercial use or SaaS re-hosting requires a license. Converts to MIT after four years. --- *A [Circuit Forge LLC](https://circuitforge.tech) product. Privacy · Safety · Accessibility — co-equal, non-negotiable.* + +--- + +Humans own design, architecture, code review, testing, and verification. LLMs are part of our development workflow. [Our positions on LLM use →](https://circuitforge.tech/positions) diff --git a/app/api/library.py b/app/api/library.py index e3ab7fe..1eec5a8 100644 --- a/app/api/library.py +++ b/app/api/library.py @@ -23,16 +23,26 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/library", tags=["library"]) -_INGEST_TASKS = { - ".pdf": "pagepiper/ingest_pdf", - ".epub": "pagepiper/ingest_epub", - ".docx": "pagepiper/ingest_docx", +_SHELVE_TASKS = { + ".pdf": "pagepiper/shelve_pdf", + ".epub": "pagepiper/shelve_epub", + ".docx": "pagepiper/shelve_docx", + ".odt": "pagepiper/shelve_odt", + ".pages": "pagepiper/shelve_pages", + ".xlsx": "pagepiper/shelve_xlsx", + ".ods": "pagepiper/shelve_ods", + ".numbers": "pagepiper/shelve_numbers", } -_INGEST_RUNNERS = { - ".pdf": "scripts.ingest_pdf", - ".epub": "scripts.ingest_epub", - ".docx": "scripts.ingest_docx", +_SHELVE_RUNNERS = { + ".pdf": "scripts.shelve_pdf", + ".epub": "scripts.shelve_epub", + ".docx": "scripts.shelve_docx", + ".odt": "scripts.shelve_odt", + ".pages": "scripts.shelve_pages", + ".xlsx": "scripts.shelve_xlsx", + ".ods": "scripts.shelve_ods", + ".numbers": "scripts.shelve_numbers", } @@ -42,19 +52,19 @@ def _mark_indexes_dirty(ctx: UserCtx) -> None: ctx.colbert.mark_dirty() -def _dispatch_ingest( +def _dispatch_shelve( doc_id: str, file_path: str, background_tasks: BackgroundTasks, data_dir: Path, mark_dirty_fn: Callable[[], None], ) -> str: - """Dispatch an ingest task. Tries cf-orch; falls back to BackgroundTasks.""" + """Dispatch a shelve task. Tries cf-orch; falls back to BackgroundTasks.""" import importlib suffix = Path(file_path).suffix.lower() - task_name = _INGEST_TASKS.get(suffix, "pagepiper/ingest_pdf") - runner_module = _INGEST_RUNNERS.get(suffix, "scripts.ingest_pdf") + task_name = _SHELVE_TASKS.get(suffix, "pagepiper/shelve_pdf") + runner_module = _SHELVE_RUNNERS.get(suffix, "scripts.shelve_pdf") task_id = str(uuid.uuid4()) args = { @@ -67,24 +77,24 @@ def _dispatch_ingest( try: from circuitforge_core.tasks import dispatch_task # type: ignore[import] task_id = dispatch_task(caller=task_name, args=args) - logger.info("Dispatched cf-orch ingest task %s for doc %s", task_id, doc_id) + logger.info("Dispatched cf-orch shelve task %s for doc %s", task_id, doc_id) except Exception: mod = importlib.import_module(runner_module) - background_tasks.add_task(_run_ingest_background, mod.run, args, task_id, mark_dirty_fn) + background_tasks.add_task(_run_shelve_background, mod.run, args, task_id, mark_dirty_fn) logger.info( - "cf-orch unavailable — running ingest in background thread (task %s)", task_id + "cf-orch unavailable — running shelve in background thread (task %s)", task_id ) return task_id -def _run_ingest_background( +def _run_shelve_background( run_fn: Callable[..., None], args: dict, task_id: str, mark_dirty_fn: Callable[[], None] | None = None, ) -> None: - from app.api.ingest import _task_registry + from app.api.shelve import _task_registry _task_registry[task_id] = {"status": "running", "progress": 0} try: run_fn(**args) @@ -92,7 +102,7 @@ def _run_ingest_background( if mark_dirty_fn: mark_dirty_fn() except Exception as exc: - logger.exception("Ingest task %s failed", task_id) + logger.exception("Shelve task %s failed", task_id) _task_registry[task_id] = {"status": "error", "error": str(exc)} @@ -134,7 +144,7 @@ def scan_library( db: sqlite3.Connection = Depends(get_db), ctx: UserCtx = Depends(get_user_ctx), ) -> dict: - """Scan the watched directory and queue ingest for any new PDFs.""" + """Scan the watched directory and queue shelving for any new PDFs.""" watch = ctx.watch_dir if not watch.exists(): raise HTTPException(status_code=404, detail=f"Watch directory not found: {watch}") @@ -143,6 +153,11 @@ def scan_library( list(watch.glob("**/*.pdf")) + list(watch.glob("**/*.epub")) + list(watch.glob("**/*.docx")) + + list(watch.glob("**/*.odt")) + + list(watch.glob("**/*.pages")) + + list(watch.glob("**/*.xlsx")) + + list(watch.glob("**/*.ods")) + + list(watch.glob("**/*.numbers")) ) queued = [] @@ -165,7 +180,7 @@ def scan_library( ).fetchone()[0] db.commit() - task_id = _dispatch_ingest( + task_id = _dispatch_shelve( doc_id, path_str, background_tasks, ctx.data_dir, lambda: _mark_indexes_dirty(ctx) ) db.execute( @@ -178,8 +193,8 @@ def scan_library( return {"discovered": len(pdfs), "queued": len(queued), "tasks": queued} -@router.post("/{doc_id}/reingest", status_code=202) -def reingest_document( +@router.post("/{doc_id}/reshelve", status_code=202) +def reshelve_document( doc_id: str, background_tasks: BackgroundTasks, db: sqlite3.Connection = Depends(get_db), @@ -189,7 +204,7 @@ def reingest_document( if not row: raise HTTPException(status_code=404, detail="Document not found") - task_id = _dispatch_ingest( + task_id = _dispatch_shelve( doc_id, row["file_path"], background_tasks, ctx.data_dir, lambda: _mark_indexes_dirty(ctx) ) db.execute( @@ -265,8 +280,11 @@ def upload_document( """Accept a PDF/EPUB upload, save to data/uploads/, and queue for indexing.""" name = Path(file.filename or "").name suffix = Path(name).suffix.lower() - if suffix not in _INGEST_TASKS: - raise HTTPException(status_code=400, detail="Supported formats: PDF, EPUB, DOCX") + if suffix not in _SHELVE_TASKS: + raise HTTPException( + status_code=400, + detail="Supported formats: PDF, EPUB, DOCX, ODT, Pages, XLSX, ODS, Numbers", + ) content = file.file.read() if len(content) > _MAX_UPLOAD_BYTES: @@ -295,7 +313,7 @@ def upload_document( ).fetchone()[0] db.commit() - task_id = _dispatch_ingest( + task_id = _dispatch_shelve( doc_id, path_str, background_tasks, ctx.data_dir, lambda: _mark_indexes_dirty(ctx) ) db.execute( diff --git a/app/api/ingest.py b/app/api/shelve.py similarity index 74% rename from app/api/ingest.py rename to app/api/shelve.py index 318c948..a07ebfe 100644 --- a/app/api/ingest.py +++ b/app/api/shelve.py @@ -1,12 +1,12 @@ -# app/api/ingest.py -"""Ingest job status polling (proxies cf-orch or checks in-memory registry).""" +# app/api/shelve.py +"""Shelve job status polling (proxies cf-orch or checks in-memory registry).""" from __future__ import annotations from fastapi import APIRouter, HTTPException -router = APIRouter(prefix="/api/ingest", tags=["ingest"]) +router = APIRouter(prefix="/api/shelve", tags=["shelve"]) -# Populated by _run_ingest_background when cf-orch is unavailable +# Populated by _run_shelve_background when cf-orch is unavailable _task_registry: dict[str, dict] = {} diff --git a/app/config.py b/app/config.py index 88c0a71..a689541 100644 --- a/app/config.py +++ b/app/config.py @@ -14,6 +14,21 @@ VEC_DIMENSIONS = int(os.environ.get("PAGEPIPER_EMBED_DIMS", "1024")) LOCAL_USER_ID = "__local__" +CF_LICENSE_KEY = os.environ.get("CF_LICENSE_KEY") + +# Priority: GPU_SERVER_URL env var -> CF_ORCH_URL env var (backward compat) +# -> https://orch.circuitforge.tech when CF_LICENSE_KEY is present (Paid+) +# Resolved value is written back to os.environ["CF_ORCH_URL"] so every existing +# caller that reads os.environ.get("CF_ORCH_URL") directly (get_llm_config below, +# app/api/chat.py) sees the right URL without any further changes. +GPU_SERVER_URL = ( + os.environ.get("GPU_SERVER_URL") + or os.environ.get("CF_ORCH_URL") + or ("https://orch.circuitforge.tech" if CF_LICENSE_KEY else None) +) +if GPU_SERVER_URL: + os.environ["CF_ORCH_URL"] = GPU_SERVER_URL + def user_data_dir(user_id: str) -> Path: """Return (and create) the per-user data directory under DATA_DIR/users/.""" diff --git a/app/main.py b/app/main.py index 9d4ba7f..61f8b20 100644 --- a/app/main.py +++ b/app/main.py @@ -64,14 +64,14 @@ app = FastAPI(title="Pagepiper", lifespan=lifespan) # Register routers from app.api.library import router as library_router # noqa: E402 -from app.api.ingest import router as ingest_router # noqa: E402 +from app.api.shelve import router as shelve_router # noqa: E402 from app.api.search import router as search_router # noqa: E402 from app.api.chat import router as chat_router # noqa: E402 from app.api.feedback import router as feedback_router # noqa: E402 from app.api.feedback_attach import router as feedback_attach_router # noqa: E402 app.include_router(library_router) -app.include_router(ingest_router) +app.include_router(shelve_router) app.include_router(search_router) app.include_router(chat_router) app.include_router(feedback_router, prefix="/api/v1/feedback") diff --git a/app/services/bm25_index.py b/app/services/bm25_index.py index 5aee25e..0206c89 100644 --- a/app/services/bm25_index.py +++ b/app/services/bm25_index.py @@ -40,7 +40,7 @@ class BM25Index: self._dirty: bool = True def mark_dirty(self) -> None: - """Signal that the index needs rebuilding (call after any ingest completes).""" + """Signal that the index needs rebuilding (call after any document is shelved).""" self._dirty = True def ensure_fresh(self, db_path: str) -> None: diff --git a/app/services/synthesizer.py b/app/services/synthesizer.py index 80346e5..06d7cb5 100644 --- a/app/services/synthesizer.py +++ b/app/services/synthesizer.py @@ -19,7 +19,7 @@ _SYSTEM_PROMPT = ( _NO_RESULTS_ANSWER = ( "I could not find any relevant passages in the indexed documents for that question. " - "Try rephrasing, or check that the relevant document has been ingested." + "Try rephrasing, or check that the relevant document has been shelved." ) # Phrases the model uses when it escapes the provided context and pulls from @@ -53,7 +53,7 @@ def _strip_escape(response: str) -> str: if any(phrase in lower for phrase in _ESCAPE_PHRASES): return ( "I could not find an answer to that question in the indexed documents. " - "The answer may be in a document that has not been ingested yet." + "The answer may be in a document that has not been shelved yet." ) return response diff --git a/app/startup.py b/app/startup.py index 9736f2a..38e0d8f 100644 --- a/app/startup.py +++ b/app/startup.py @@ -63,11 +63,21 @@ def reembed_docs(docs: list[tuple[str, str]], db_path: str, vec_db_path: str) -> suffix = os.path.splitext(file_path)[1].lower() try: if suffix == ".epub": - from scripts.ingest_epub import run + from scripts.shelve_epub import run elif suffix == ".docx": - from scripts.ingest_docx import run + from scripts.shelve_docx import run + elif suffix == ".odt": + from scripts.shelve_odt import run + elif suffix == ".pages": + from scripts.shelve_pages import run + elif suffix == ".xlsx": + from scripts.shelve_xlsx import run + elif suffix == ".ods": + from scripts.shelve_ods import run + elif suffix == ".numbers": + from scripts.shelve_numbers import run else: - from scripts.ingest_pdf import run + from scripts.shelve_pdf import run logger.info("Auto re-embed: starting %s", os.path.basename(file_path)) run(doc_id=doc_id, file_path=file_path, db_path=db_path, vec_db_path=vec_db_path) except Exception as exc: diff --git a/compose.cloud.yml b/compose.cloud.yml index 2708b71..93a4700 100644 --- a/compose.cloud.yml +++ b/compose.cloud.yml @@ -17,8 +17,10 @@ services: PAGEPIPER_BOOKS_DIR: /devl/pagepiper-cloud-data/books # PAGEPIPER_OLLAMA_URL — set in .env (BYOK gate for hybrid search + RAG) # HEIMDALL_URL, HEIMDALL_ADMIN_TOKEN — set in .env for license validation - # cf-orch: route LLM inference through coordinator for managed GPU access - CF_ORCH_URL: http://host.docker.internal:7700 + # GPU server: route LLM inference through the cf-orch coordinator for + # managed GPU access (app/config.py normalizes this into CF_ORCH_URL + # internally, so no other service code needs to change) + GPU_SERVER_URL: http://host.docker.internal:7700 CF_APP_NAME: pagepiper # CF_LICENSE_KEY is the auth token CFOrchClient sends to the coordinator CF_LICENSE_KEY: ${COORDINATOR_PAGEPIPER_KEY:-} diff --git a/config/ingest.yaml b/config/shelve.yaml similarity index 100% rename from config/ingest.yaml rename to config/shelve.yaml diff --git a/docs/getting-started/ollama-setup.md b/docs/getting-started/ollama-setup.md index 9661f0f..6c72200 100644 --- a/docs/getting-started/ollama-setup.md +++ b/docs/getting-started/ollama-setup.md @@ -39,7 +39,7 @@ Restart Pagepiper: ## Verify -Upload or re-index a document. The document card should show **Embedding N / M pages** during ingest. Once complete, the Chat tab becomes active. +Upload or re-index a document. The document card should show **Embedding N / M pages** while shelving. Once complete, the Chat tab becomes active. ## Changing embedding models diff --git a/docs/getting-started/quick-start.md b/docs/getting-started/quick-start.md index e6fc786..4803d99 100644 --- a/docs/getting-started/quick-start.md +++ b/docs/getting-started/quick-start.md @@ -14,7 +14,7 @@ Open `http://localhost:8521` in your browser. You have two options: -**Upload directly** — click **Upload PDF / EPUB** in the library header and pick a file from your computer. +**Upload directly** — click **Upload Document or Spreadsheet** in the library header and pick a file from your computer. **Scan a directory** — set `PAGEPIPER_WATCH_DIR` in your `.env` to a folder of PDFs or EPUBs, then click **Scan for PDFs**. Pagepiper indexes every file it finds. @@ -22,7 +22,7 @@ You have two options: The document card shows progress while text is being extracted and embedded: -- **Extracting text...** (animated bar) — PDF/EPUB is being parsed into page chunks +- **Extracting text...** (animated bar) — the file is being parsed into page chunks - **Embedding N / M pages (X%)** (filling bar) — vectors are being written to the vector store (only when Ollama is configured) Once the badge shows **READY**, the document is searchable. diff --git a/docs/index.md b/docs/index.md index 42f0fc9..b7470cc 100644 --- a/docs/index.md +++ b/docs/index.md @@ -1,6 +1,6 @@ # Pagepiper -Self-hosted document search with BM25 full-text indexing and (with local Ollama) hybrid vector search and LLM-powered chat. Supports PDF and EPUB files. +Self-hosted document search with BM25 full-text indexing and (with local Ollama) hybrid vector search and LLM-powered chat. Supports PDF, EPUB, DOCX, ODT, Apple Pages, XLSX, ODS, and Apple Numbers files. ## Demo @@ -12,7 +12,7 @@ Try it: [pagepiper.circuitforge.tech](https://pagepiper.circuitforge.tech) ![Library view](screenshots/01-library.png) -Scan your PDF directory to index documents, or upload individual PDFs directly. Each document shows page count and ingest status. +Scan your PDF directory to index documents, or upload individual PDFs directly. Each document shows page count and shelving status. ### Chat @@ -25,8 +25,8 @@ Ask questions across your indexed documents. Results cite the source document an | Feature | Free | Paid (BYOK) | |---------|------|-------------| | BM25 full-text search | Yes | Yes | -| PDF and EPUB upload via browser | Yes | Yes | -| Unlimited local ingestion | Yes | Yes | +| All supported formats upload via browser | Yes | Yes | +| Unlimited local shelving | Yes | Yes | | Hybrid vector search | No | Yes (local Ollama) | | LLM chat over documents | No | Yes (local Ollama) | @@ -137,6 +137,6 @@ docker compose up -d --build ## Notes -- Pagepiper indexes PDFs at ingest time. Changes to the source file require a re-index (use the re-index button on the document card). +- Pagepiper indexes PDFs at shelve time. Changes to the source file require a re-index (use the re-index button on the document card). - The `data/` directory contains the SQLite index database and any uploaded files. Back it up to preserve your index. - Large PDFs (hundreds of pages) can take a few minutes to index. Watch the status badge on the document card. diff --git a/docs/reference/architecture.md b/docs/reference/architecture.md index 4069ec1..581780e 100644 --- a/docs/reference/architecture.md +++ b/docs/reference/architecture.md @@ -16,10 +16,10 @@ Browser (Vue 3 SPA) sqlite-vec (vectors) ``` -## Ingest pipeline +## Shelve pipeline ``` -PDF / EPUB file +any supported document or spreadsheet file │ ├─ PDFExtractor (pdfminer + OCR fallback) ← circuitforge_core │ or @@ -56,5 +56,5 @@ The vector database stores one row per page chunk. If the embedding model change | Component | License | |-----------|---------| -| BM25 search, ingest pipeline, library API | MIT | +| BM25 search, shelve pipeline, library API | MIT | | Hybrid vector search, RAG chat, embedding | BSL 1.1 (BYOK unlocked on Free tier) | diff --git a/docs/reference/environment-variables.md b/docs/reference/environment-variables.md index f1ec664..cd2697c 100644 --- a/docs/reference/environment-variables.md +++ b/docs/reference/environment-variables.md @@ -27,12 +27,13 @@ Copy `.env.example` to `.env` and configure as needed. **Note:** `page_vecs` (the sqlite-vec table populated at shelve time using `PAGEPIPER_EMBED_MODEL`) is no longer read by search or chat — retrieval was switched to the ColBERT index above (pagepiper#8). Shelving still computes and stores those embeddings for now; removing that redundant work is tracked as a follow-up. -## cf-orch (managed deployments) +## GPU server / cf-orch (managed deployments) | Variable | Default | Description | |----------|---------|-------------| -| `CF_ORCH_URL` | _(unset)_ | cf-orch coordinator URL for GPU allocation | -| `CF_LICENSE_KEY` | _(unset)_ | License key for cf-orch authentication | +| `GPU_SERVER_URL` | _(unset)_ | Self-hosted GPU rig / cf-orch coordinator URL for GPU allocation. Preferred over `CF_ORCH_URL`. | +| `CF_ORCH_URL` | _(unset)_ | Legacy alias for `GPU_SERVER_URL` — still honoured if `GPU_SERVER_URL` is unset. | +| `CF_LICENSE_KEY` | _(unset)_ | License key for cf-orch authentication. When set and neither `GPU_SERVER_URL` nor `CF_ORCH_URL` is configured, defaults `GPU_SERVER_URL` to `https://orch.circuitforge.tech` (Paid+ tiers). | | `CF_APP_NAME` | `pagepiper` | Application identifier sent to cf-orch | ## License (cloud tier) diff --git a/docs/reference/tier-system.md b/docs/reference/tier-system.md index 2a36347..e483be3 100644 --- a/docs/reference/tier-system.md +++ b/docs/reference/tier-system.md @@ -4,7 +4,7 @@ |---------|------|-------------| | BM25 full-text search | Yes | Yes | | PDF and EPUB upload | Yes | Yes | -| Unlimited local ingestion | Yes | Yes | +| Unlimited local shelving | Yes | Yes | | Directory scan | Yes | Yes | | Hybrid vector search | No | Yes (local Ollama) | | RAG chat with page citations | No | Yes (local Ollama) | diff --git a/docs/user-guide/library.md b/docs/user-guide/library.md index 00957ca..77131e6 100644 --- a/docs/user-guide/library.md +++ b/docs/user-guide/library.md @@ -4,9 +4,9 @@ The library is the home screen. It shows all indexed documents and lets you add ## Adding documents -**Upload** — click **Upload PDF / EPUB** and select a file. Files up to 200 MB are accepted. The document is saved to `data/uploads/` and queued for indexing immediately. +**Upload** — click **Upload Document or Spreadsheet** and select a file. Files up to 200 MB are accepted. The document is saved to `data/uploads/` and queued for indexing immediately. -**Scan** — set `PAGEPIPER_WATCH_DIR` to a directory in your `.env`, then click **Scan for PDFs**. Any PDF or EPUB not already in the library is queued. Re-scanning is safe; already-indexed documents are skipped. +**Scan** — set `PAGEPIPER_WATCH_DIR` to a directory in your `.env`, then click **Scan for PDFs**. Any supported document or spreadsheet file not already in the library is queued. Re-scanning is safe; already-indexed documents are skipped. ## Document states @@ -16,7 +16,7 @@ The library is the home screen. It shows all indexed documents and lets you add | READY | Fully indexed and searchable | | ERROR | Indexing failed — see the error message on the card | -## Ingestion progress +## Shelving progress While a document is processing, its card shows a live progress bar: @@ -27,10 +27,10 @@ The card refreshes automatically and emits a library reload when indexing comple ## Re-indexing -Click **Re-index** on any document card to re-run the full ingest pipeline. This is useful after: +Click **Re-index** on any document card to re-run the full shelve pipeline. This is useful after: - Changing the `PAGEPIPER_EMBED_MODEL` (dimension mismatch auto-detected at startup, but you can also trigger manually) -- A failed ingest you want to retry +- A failed shelve you want to retry - Updating to a new version of Pagepiper with an improved extractor ## Removing a document diff --git a/mkdocs.yml b/mkdocs.yml index a9367db..66e7443 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -1,5 +1,5 @@ site_name: Pagepiper -site_description: Self-hosted PDF and EPUB library with BM25 full-text search, hybrid vector retrieval, and LLM-powered RAG chat. +site_description: Self-hosted document and spreadsheet library (PDF, EPUB, DOCX, ODT, Apple Pages, XLSX, ODS, Apple Numbers) with BM25 full-text search, hybrid vector retrieval, and LLM-powered RAG chat. site_author: Circuit Forge LLC site_url: https://docs.circuitforge.tech/pagepiper repo_url: https://git.opensourcesolarpunk.com/Circuit-Forge/pagepiper @@ -9,14 +9,14 @@ theme: name: material palette: - scheme: default - primary: deep purple - accent: purple + primary: brown + accent: orange toggle: icon: material/brightness-7 name: Switch to dark mode - scheme: slate - primary: deep purple - accent: purple + primary: brown + accent: amber toggle: icon: material/brightness-4 name: Switch to light mode @@ -60,5 +60,8 @@ nav: - Tier System: reference/tier-system.md - Environment Variables: reference/environment-variables.md +extra_css: + - stylesheets/theme.css + extra_javascript: - plausible.js diff --git a/pyproject.toml b/pyproject.toml index e3202bf..d841540 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,9 @@ dependencies = [ "PyYAML>=6.0", "httpx>=0.27", "circuitforge-core[pdf,vector]>=0.19.0", + "python-docx>=1.0", + "odfpy>=1.4", + "openpyxl>=3.1", "pylate[voyager]>=1.6", ] diff --git a/scripts/shelve_docx.py b/scripts/shelve_docx.py new file mode 100644 index 0000000..7588874 --- /dev/null +++ b/scripts/shelve_docx.py @@ -0,0 +1,280 @@ +# scripts/shelve_docx.py +""" +cf-orch task: pagepiper/shelve_docx + +Extracts text from a Word .docx file, stores section chunks in SQLite, and +(if Ollama is configured) generates embeddings in the sqlite-vec store. + +Chunking strategy: + - If the document has >=2 Heading-style paragraphs: split at each heading + (one chunk per section, heading text included). + - Otherwise: accumulate blocks into ~WORDS_PER_CHUNK rolling windows. + +Tables are serialised as pipe-delimited rows and included in the surrounding +section chunk, preserving document order via XML tree traversal. + +Entry point: + python scripts/shelve_docx.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, field +from pathlib import Path + +logger = logging.getLogger("pagepiper.shelve_docx") + +EMBED_BATCH_SIZE = 64 +_WORDS_PER_CHUNK = 500 + + +@dataclass +class _Chunk: + page_number: int + text: str + source: str + word_count: int + + +def _table_to_text(table) -> str: + """Serialise a DOCX table as pipe-delimited rows.""" + lines = [] + for row in table.rows: + cells = [c.text.strip().replace("\n", " ") for c in row.cells] + if any(cells): + lines.append(" | ".join(cells)) + return "\n".join(lines) + + +def _iter_blocks(doc): + """ + Yield (kind, obj) pairs in document body order, where kind is + 'paragraph' or 'table'. Walks the raw XML so that tables and + paragraphs appear in the correct interleaved sequence. + """ + import docx.text.paragraph as _p_mod + import docx.table as _t_mod + from docx.oxml.ns import qn + + for child in doc.element.body.iterchildren(): + if child.tag == qn("w:p"): + yield "paragraph", _p_mod.Paragraph(child, doc) + elif child.tag == qn("w:tbl"): + yield "table", _t_mod.Table(child, doc) + + +def _is_heading(para) -> bool: + return para.style.name.startswith("Heading") + + +def _extract_chunks(file_path: str) -> list[_Chunk]: + import docx + from scripts.text_clean import clean_line, is_artifact_line + + doc = docx.Document(file_path) + + # Count headings to decide strategy + heading_count = sum(1 for p in doc.paragraphs if _is_heading(p)) + + blocks: list[tuple[str, object]] = list(_iter_blocks(doc)) + + if heading_count >= 2: + return _heading_chunks(blocks) + else: + return _wordcount_chunks(blocks) + + +def _heading_chunks(blocks: list[tuple[str, object]]) -> list[_Chunk]: + """One chunk per heading section; tables included inline.""" + from scripts.text_clean import clean_line, is_artifact_line + + chunks: list[_Chunk] = [] + current_parts: list[str] = [] + + def _flush(parts: list[str]) -> None: + text = "\n".join(parts).strip() + if text: + n = len(chunks) + 1 + chunks.append(_Chunk(n, text, "section", len(text.split()))) + + for kind, obj in blocks: + if kind == "paragraph": + if _is_heading(obj): + _flush(current_parts) + current_parts = [] + t = obj.text.strip() + if t: + current_parts.append(t) + else: + t = clean_line(obj.text.strip()) + if t and not is_artifact_line(t): + current_parts.append(t) + elif kind == "table": + table_text = _table_to_text(obj) + if table_text: + current_parts.append(table_text) + + _flush(current_parts) + return chunks + + +def _wordcount_chunks(blocks: list[tuple[str, object]]) -> list[_Chunk]: + """Accumulate blocks into ~WORDS_PER_CHUNK rolling windows.""" + from scripts.text_clean import clean_line, is_artifact_line + + chunks: list[_Chunk] = [] + current: list[str] = [] + current_count = 0 + + def _flush(parts: list[str]) -> None: + text = "\n".join(parts).strip() + if text: + n = len(chunks) + 1 + chunks.append(_Chunk(n, text, "text", len(text.split()))) + + for kind, obj in blocks: + if kind == "paragraph": + t = clean_line(obj.text.strip()) + if not t or is_artifact_line(t): + continue + else: # table + t = _table_to_text(obj) + if not t: + continue + + words = t.split() + if current_count + len(words) > _WORDS_PER_CHUNK and current: + _flush(current) + current, current_count = [], 0 + current.append(t) + current_count += len(words) + + if current: + _flush(current) + + return chunks + + +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 DOCX. 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 sections 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 a Word .docx (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, + ) diff --git a/scripts/ingest_epub.py b/scripts/shelve_epub.py similarity index 84% rename from scripts/ingest_epub.py rename to scripts/shelve_epub.py index 8984726..1f70e7f 100644 --- a/scripts/ingest_epub.py +++ b/scripts/shelve_epub.py @@ -1,6 +1,6 @@ -# scripts/ingest_epub.py +# scripts/shelve_epub.py """ -cf-orch task: pagepiper/ingest_epub +cf-orch task: pagepiper/shelve_epub Extracts text from an EPUB file, stores chapter chunks in SQLite, and (if Ollama is configured) generates embeddings and stores them in the sqlite-vec store. @@ -8,7 +8,7 @@ configured) generates embeddings and stores them in the sqlite-vec store. Each EPUB chapter becomes one chunk (equivalent to a PDF page). Entry point: - python scripts/ingest_epub.py --doc-id X --file-path Y --db-path Z --vec-db-path W + python scripts/shelve_epub.py --doc-id X --file-path Y --db-path Z --vec-db-path W """ from __future__ import annotations @@ -18,7 +18,7 @@ import sqlite3 from dataclasses import dataclass from pathlib import Path -logger = logging.getLogger("pagepiper.ingest_epub") +logger = logging.getLogger("pagepiper.shelve_epub") EMBED_BATCH_SIZE = 64 _WORDS_PER_CHUNK = 500 # target chunk size for word-count fallback @@ -131,7 +131,7 @@ def _update_status( def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: - """Run the full ingest pipeline for one EPUB. Called by cf-orch or BackgroundTasks.""" + """Run the full shelve pipeline for one EPUB. Called by cf-orch or BackgroundTasks.""" conn: sqlite3.Connection | None = None try: conn = sqlite3.connect(db_path, timeout=30) @@ -155,29 +155,15 @@ def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: conn.commit() # Embedding failure is non-fatal: document remains BM25-searchable. - ollama_url = os.environ.get("PAGEPIPER_OLLAMA_URL", "").strip() - if ollama_url and chunks: + from app.config import get_llm_config + llm_cfg = get_llm_config() + if llm_cfg and chunks: try: - logger.info("Embedding %d chapters via Ollama at %s", len(chunks), ollama_url) + logger.info("Embedding %d chapters", len(chunks)) from circuitforge_core.llm import LLMRouter from circuitforge_core.vector.sqlite_vec import LocalSQLiteVecStore - _clean = ollama_url.rstrip("/") - base_url = _clean if _clean.endswith("/v1") else _clean + "/v1" - router = LLMRouter({ - "fallback_order": ["ollama"], - "backends": { - "ollama": { - "type": "openai_compat", - "base_url": base_url, - "model": os.environ.get("PAGEPIPER_CHAT_MODEL", "mistral:7b"), - "embedding_model": os.environ.get( - "PAGEPIPER_EMBED_MODEL", "nomic-embed-text" - ), - "supports_images": False, - } - }, - }) + 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 @@ -203,10 +189,10 @@ def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: ) _update_status(conn, doc_id, "ready", page_count=len(chunks)) - logger.info("Ingest complete for doc %s (%d chapters)", doc_id, len(chunks)) + logger.info("Shelve complete for doc %s (%d chapters)", doc_id, len(chunks)) except Exception as exc: - logger.error("Ingest failed for doc %s: %s", doc_id, exc, exc_info=True) + 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)) @@ -224,7 +210,7 @@ if __name__ == "__main__": logging.basicConfig(level=logging.INFO) parser = argparse.ArgumentParser( - description="Ingest an EPUB (cf-orch task entry point)" + description="Shelve an EPUB (cf-orch task entry point)" ) parser.add_argument("--doc-id", required=True) parser.add_argument("--file-path", required=True) diff --git a/scripts/shelve_numbers.py b/scripts/shelve_numbers.py new file mode 100644 index 0000000..9accb30 --- /dev/null +++ b/scripts/shelve_numbers.py @@ -0,0 +1,101 @@ +# scripts/shelve_numbers.py +""" +cf-orch task: pagepiper/shelve_numbers + +Converts an Apple Numbers (.numbers) file to XLSX via headless LibreOffice +(soffice), then delegates extraction/chunking/embedding to shelve_xlsx's +pipeline. + +Like .pages, .numbers is IWA internally (Snappy-compressed Protocol +Buffers) — no maintained Python library parses it directly. LibreOffice +bundles libetonyek, the only mature open-source parser for Apple's iWork +formats, so shelling out to headless LibreOffice is the only realistic +full-fidelity extraction path. + +Entry point: + python scripts/shelve_numbers.py --doc-id X --file-path Y --db-path Z --vec-db-path W +""" +from __future__ import annotations + +import logging +import shutil +import sqlite3 +import subprocess +import tempfile +from pathlib import Path + +logger = logging.getLogger("pagepiper.shelve_numbers") + +_CONVERT_TIMEOUT_SECONDS = 120 + + +def _update_status_error(db_path: str, doc_id: str, error_msg: str) -> None: + conn = sqlite3.connect(db_path, timeout=30) + try: + conn.execute( + "UPDATE documents SET status='error', error_msg=?, updated_at=datetime('now') WHERE id=?", + [error_msg, doc_id], + ) + conn.commit() + finally: + conn.close() + + +def _convert_to_xlsx(numbers_path: str, out_dir: str) -> str: + """Shell out to headless LibreOffice to convert .numbers -> .xlsx. Returns the output path.""" + result = subprocess.run( + ["soffice", "--headless", "--convert-to", "xlsx", "--outdir", out_dir, numbers_path], + capture_output=True, text=True, timeout=_CONVERT_TIMEOUT_SECONDS, + ) + if result.returncode != 0: + raise RuntimeError( + f"LibreOffice conversion failed: {result.stderr.strip() or result.stdout.strip()}" + ) + + stem = Path(numbers_path).stem + xlsx_path = Path(out_dir) / f"{stem}.xlsx" + if not xlsx_path.exists(): + raise RuntimeError(f"LibreOffice reported success but produced no output for {numbers_path}") + return str(xlsx_path) + + +def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: + """Convert a .numbers file to XLSX, then run the XLSX shelve pipeline against it.""" + from scripts.shelve_xlsx import run as run_xlsx + + tmp_dir = tempfile.mkdtemp(prefix="pagepiper_numbers_") + try: + logger.info("Converting %s to XLSX via headless LibreOffice", file_path) + xlsx_path = _convert_to_xlsx(file_path, tmp_dir) + logger.info("Converted to %s — handing off to XLSX shelver", xlsx_path) + run_xlsx(doc_id=doc_id, file_path=xlsx_path, db_path=db_path, vec_db_path=vec_db_path) + except Exception as exc: + logger.error("Shelve failed for doc %s: %s", doc_id, exc, exc_info=True) + try: + _update_status_error(db_path, doc_id, str(exc)) + except Exception: + logger.warning("Could not write error status for doc %s", doc_id) + raise + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +if __name__ == "__main__": + import argparse + + logging.basicConfig(level=logging.INFO) + + parser = argparse.ArgumentParser( + description="Shelve an Apple Numbers .numbers file (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, + ) diff --git a/scripts/shelve_ods.py b/scripts/shelve_ods.py new file mode 100644 index 0000000..9a4560c --- /dev/null +++ b/scripts/shelve_ods.py @@ -0,0 +1,211 @@ +# scripts/shelve_ods.py +""" +cf-orch task: pagepiper/shelve_ods + +Extracts rows from an OpenDocument Spreadsheet (.ods), stores chunks in +SQLite, and (if Ollama is configured) generates embeddings in the +sqlite-vec store. + +Chunking strategy identical to shelve_xlsx.py: one chunk per sheet (a + element), row-windowed for large sheets with the header row +repeated in every window. + +Entry point: + python scripts/shelve_ods.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_ods") + +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(cells: list[str]) -> str: + cells = [c.strip() for c in cells] + return " | ".join(cells) if any(cells) else "" + + +def _sheet_to_chunks(sheet_name: str, rows: list[list[str]], 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]: + from odf.opendocument import load + from odf.table import Table, TableRow, TableCell + from odf import teletype + + doc = load(file_path) + chunks: list[_Chunk] = [] + for element in doc.spreadsheet.childNodes: + if element.qname[1] != "table": + continue + sheet_name = element.getAttribute("name") or "Sheet" + rows: list[list[str]] = [] + for row in element.getElementsByType(TableRow): + cells = [ + teletype.extractText(c).replace("\n", " ") + for c in row.getElementsByType(TableCell) + ] + rows.append(cells) + next_page = len(chunks) + 1 + chunks.extend(_sheet_to_chunks(sheet_name, rows, next_page)) + return chunks + + +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 ODS. 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 OpenDocument Spreadsheet .ods (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, + ) diff --git a/scripts/shelve_odt.py b/scripts/shelve_odt.py new file mode 100644 index 0000000..35cec9d --- /dev/null +++ b/scripts/shelve_odt.py @@ -0,0 +1,268 @@ +# scripts/shelve_odt.py +""" +cf-orch task: pagepiper/shelve_odt + +Extracts text from an OpenDocument Text (.odt) file, stores section chunks in +SQLite, and (if Ollama is configured) generates embeddings in the sqlite-vec +store. + +Chunking strategy: + - If the document has >=2 heading paragraphs (): split at each + heading (one chunk per section, heading text included). + - Otherwise: accumulate blocks into ~WORDS_PER_CHUNK rolling windows. + +Tables are serialised as pipe-delimited rows and included in the surrounding +section chunk. odfpy already yields body children in document order, so no +raw XML tree-walk is needed (unlike the DOCX shelver). + +Entry point: + python scripts/shelve_odt.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 +from pathlib import Path + +logger = logging.getLogger("pagepiper.shelve_odt") + +EMBED_BATCH_SIZE = 64 +_WORDS_PER_CHUNK = 500 + + +@dataclass +class _Chunk: + page_number: int + text: str + source: str + word_count: int + + +def _table_to_text(table) -> str: + """Serialise an ODT table as pipe-delimited rows.""" + from odf.table import TableCell, TableRow + from odf import teletype + + lines = [] + for row in table.getElementsByType(TableRow): + cells = [teletype.extractText(c).strip().replace("\n", " ") for c in row.getElementsByType(TableCell)] + if any(cells): + lines.append(" | ".join(cells)) + return "\n".join(lines) + + +def _extract_chunks(file_path: str) -> list[_Chunk]: + from odf.opendocument import load + from odf import teletype + from scripts.text_clean import clean_line, is_artifact_line + + doc = load(file_path) + blocks = list(doc.text.childNodes) + + heading_count = sum(1 for b in blocks if b.qname[1] == "h") + + if heading_count >= 2: + return _heading_chunks(blocks) + else: + return _wordcount_chunks(blocks) + + +def _heading_chunks(blocks: list) -> list[_Chunk]: + """One chunk per heading section; tables included inline.""" + from odf import teletype + from scripts.text_clean import clean_line, is_artifact_line + + chunks: list[_Chunk] = [] + current_parts: list[str] = [] + + def _flush(parts: list[str]) -> None: + text = "\n".join(parts).strip() + if text: + n = len(chunks) + 1 + chunks.append(_Chunk(n, text, "section", len(text.split()))) + + for block in blocks: + kind = block.qname[1] + if kind == "h": + _flush(current_parts) + current_parts = [] + t = teletype.extractText(block).strip() + if t: + current_parts.append(t) + elif kind == "p": + t = clean_line(teletype.extractText(block).strip()) + if t and not is_artifact_line(t): + current_parts.append(t) + elif kind == "table": + table_text = _table_to_text(block) + if table_text: + current_parts.append(table_text) + + _flush(current_parts) + return chunks + + +def _wordcount_chunks(blocks: list) -> list[_Chunk]: + """Accumulate blocks into ~WORDS_PER_CHUNK rolling windows.""" + from odf import teletype + from scripts.text_clean import clean_line, is_artifact_line + + chunks: list[_Chunk] = [] + current: list[str] = [] + current_count = 0 + + def _flush(parts: list[str]) -> None: + text = "\n".join(parts).strip() + if text: + n = len(chunks) + 1 + chunks.append(_Chunk(n, text, "text", len(text.split()))) + + for block in blocks: + kind = block.qname[1] + if kind in ("p", "h"): + t = clean_line(teletype.extractText(block).strip()) + if not t or is_artifact_line(t): + continue + elif kind == "table": + t = _table_to_text(block) + if not t: + continue + else: + continue + + words = t.split() + if current_count + len(words) > _WORDS_PER_CHUNK and current: + _flush(current) + current, current_count = [], 0 + current.append(t) + current_count += len(words) + + if current: + _flush(current) + + return chunks + + +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 ODT. 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 sections 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 OpenDocument Text .odt (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, + ) diff --git a/scripts/shelve_pages.py b/scripts/shelve_pages.py new file mode 100644 index 0000000..ac6a357 --- /dev/null +++ b/scripts/shelve_pages.py @@ -0,0 +1,100 @@ +# scripts/shelve_pages.py +""" +cf-orch task: pagepiper/shelve_pages + +Converts an Apple Pages (.pages) file to ODT via headless LibreOffice +(soffice), then delegates extraction/chunking/embedding to shelve_odt's +pipeline. + +.pages is IWA internally (Snappy-compressed Protocol Buffers) — no +maintained Python library parses it directly. LibreOffice bundles +libetonyek, the only mature open-source Pages parser, so shelling out to +headless LibreOffice is the only realistic full-fidelity extraction path. + +Entry point: + python scripts/shelve_pages.py --doc-id X --file-path Y --db-path Z --vec-db-path W +""" +from __future__ import annotations + +import logging +import shutil +import sqlite3 +import subprocess +import tempfile +from pathlib import Path + +logger = logging.getLogger("pagepiper.shelve_pages") + +_CONVERT_TIMEOUT_SECONDS = 120 + + +def _update_status_error(db_path: str, doc_id: str, error_msg: str) -> None: + conn = sqlite3.connect(db_path, timeout=30) + try: + conn.execute( + "UPDATE documents SET status='error', error_msg=?, updated_at=datetime('now') WHERE id=?", + [error_msg, doc_id], + ) + conn.commit() + finally: + conn.close() + + +def _convert_to_odt(pages_path: str, out_dir: str) -> str: + """Shell out to headless LibreOffice to convert .pages -> .odt. Returns the output path.""" + result = subprocess.run( + ["soffice", "--headless", "--convert-to", "odt:writer8", "--outdir", out_dir, pages_path], + capture_output=True, text=True, timeout=_CONVERT_TIMEOUT_SECONDS, + ) + if result.returncode != 0: + raise RuntimeError( + f"LibreOffice conversion failed: {result.stderr.strip() or result.stdout.strip()}" + ) + + stem = Path(pages_path).stem + odt_path = Path(out_dir) / f"{stem}.odt" + if not odt_path.exists(): + raise RuntimeError(f"LibreOffice reported success but produced no output for {pages_path}") + return str(odt_path) + + +def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: + """Convert a .pages file to ODT, then run the ODT shelve pipeline against it.""" + from scripts.shelve_odt import run as run_odt + + tmp_dir = tempfile.mkdtemp(prefix="pagepiper_pages_") + try: + logger.info("Converting %s to ODT via headless LibreOffice", file_path) + odt_path = _convert_to_odt(file_path, tmp_dir) + logger.info("Converted to %s — handing off to ODT shelver", odt_path) + run_odt(doc_id=doc_id, file_path=odt_path, db_path=db_path, vec_db_path=vec_db_path) + except Exception as exc: + logger.error("Shelve failed for doc %s: %s", doc_id, exc, exc_info=True) + try: + _update_status_error(db_path, doc_id, str(exc)) + except Exception: + logger.warning("Could not write error status for doc %s", doc_id) + raise + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +if __name__ == "__main__": + import argparse + + logging.basicConfig(level=logging.INFO) + + parser = argparse.ArgumentParser( + description="Shelve an Apple Pages .pages file (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, + ) diff --git a/scripts/ingest_pdf.py b/scripts/shelve_pdf.py similarity index 77% rename from scripts/ingest_pdf.py rename to scripts/shelve_pdf.py index 65671de..5ddfaed 100644 --- a/scripts/ingest_pdf.py +++ b/scripts/shelve_pdf.py @@ -1,12 +1,12 @@ -# scripts/ingest_pdf.py +# scripts/shelve_pdf.py """ -cf-orch task: pagepiper/ingest_pdf +cf-orch task: pagepiper/shelve_pdf Extracts text from a PDF, stores page chunks in SQLite, and (if Ollama is configured) generates embeddings and stores them in the sqlite-vec store. Entry point: - python scripts/ingest_pdf.py --doc-id X --file-path Y --db-path Z --vec-db-path W + python scripts/shelve_pdf.py --doc-id X --file-path Y --db-path Z --vec-db-path W """ from __future__ import annotations @@ -15,7 +15,7 @@ import os import sqlite3 from pathlib import Path -logger = logging.getLogger("pagepiper.ingest") +logger = logging.getLogger("pagepiper.shelve_pdf") # Pages to embed per Ollama API call — avoids hitting request size limits on large PDFs EMBED_BATCH_SIZE = 64 @@ -47,7 +47,7 @@ def _update_status( def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: - """Run the full ingest pipeline for one PDF. Called by cf-orch or BackgroundTasks.""" + """Run the full shelve pipeline for one PDF. Called by cf-orch or BackgroundTasks.""" from circuitforge_core.documents.pdf import PDFExtractor conn: sqlite3.Connection | None = None @@ -79,37 +79,23 @@ def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: chunk_rows.append((row[0], chunk.page_number, cleaned_text)) conn.commit() - # Step 3: Embed and store vectors if Ollama is configured (BYOK gate) + # Step 3: Embed and store vectors if LLM is configured (BYOK gate). # Embedding failure is non-fatal: document remains BM25-searchable. - ollama_url = os.environ.get("PAGEPIPER_OLLAMA_URL", "").strip() - if ollama_url and chunks: + from app.config import get_llm_config + llm_cfg = get_llm_config() + if llm_cfg and chunks: try: - logger.info("Embedding %d pages via Ollama at %s", len(chunks), ollama_url) + logger.info("Embedding %d pages", len(chunks)) from circuitforge_core.llm import LLMRouter from circuitforge_core.vector.sqlite_vec import LocalSQLiteVecStore - _clean = ollama_url.rstrip("/") - base_url = _clean if _clean.endswith("/v1") else _clean + "/v1" - router = LLMRouter({ - "fallback_order": ["ollama"], - "backends": { - "ollama": { - "type": "openai_compat", - "base_url": base_url, - "model": os.environ.get("PAGEPIPER_CHAT_MODEL", "mistral:7b"), - "embedding_model": os.environ.get( - "PAGEPIPER_EMBED_MODEL", "nomic-embed-text" - ), - "supports_images": False, - } - }, - }) + 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 ) # Remove old vectors before re-inserting. If embedding fails mid-way, - # old vectors are gone but new ones are partial — re-ingest recovers. + # old vectors are gone but new ones are partial — re-shelving recovers. vec_store.delete_where({"doc_id": doc_id}) texts = [text for _, _, text in chunk_rows] @@ -131,10 +117,10 @@ def run(doc_id: str, file_path: str, db_path: str, vec_db_path: str) -> None: ) _update_status(conn, doc_id, "ready", page_count=len(chunks)) - logger.info("Ingest complete for doc %s (%d pages)", doc_id, len(chunks)) + logger.info("Shelve complete for doc %s (%d pages)", doc_id, len(chunks)) except Exception as exc: - logger.error("Ingest failed for doc %s: %s", doc_id, exc, exc_info=True) + 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)) @@ -152,7 +138,7 @@ if __name__ == "__main__": logging.basicConfig(level=logging.INFO) parser = argparse.ArgumentParser( - description="Ingest a PDF (cf-orch task entry point)" + description="Shelve a PDF (cf-orch task entry point)" ) parser.add_argument("--doc-id", required=True) parser.add_argument("--file-path", required=True) diff --git a/scripts/shelve_xlsx.py b/scripts/shelve_xlsx.py new file mode 100644 index 0000000..4389917 --- /dev/null +++ b/scripts/shelve_xlsx.py @@ -0,0 +1,208 @@ +# 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, + ) diff --git a/scripts/text_clean.py b/scripts/text_clean.py index 26f1f48..2737d9f 100644 --- a/scripts/text_clean.py +++ b/scripts/text_clean.py @@ -1,6 +1,6 @@ # scripts/text_clean.py """ -Shared text-cleaning utilities for ingest pipelines. +Shared text-cleaning utilities for shelve pipelines. Removes boilerplate lines injected by ebook converters, piracy watermarks, and other non-content artifacts before chunks are stored or embedded. diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..1593817 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,94 @@ +# tests/test_config.py +"""Unit tests for the GPU_SERVER_URL / CF_ORCH_URL resolution in app/config.py. + +app/config.py resolves GPU_SERVER_URL at *module import time* (module-level +code, not a function), so each test reloads the module after adjusting env +vars via monkeypatch, then reloads it again afterward with the real +environment restored so later tests see normal module state. +""" +from __future__ import annotations + +import importlib +import os + +import pytest + +import app.config as config + + +@pytest.fixture(autouse=True) +def _restore_config_module(): + yield + # config.py's write-back (`os.environ["CF_ORCH_URL"] = GPU_SERVER_URL`) is a + # module-level side effect monkeypatch never sees, so it can't auto-revert + # it — without this explicit cleanup, a value set here leaks into every + # other test in the session (e.g. shelve tests' get_llm_config() checks). + os.environ.pop("CF_ORCH_URL", None) + os.environ.pop("GPU_SERVER_URL", None) + os.environ.pop("CF_LICENSE_KEY", None) + importlib.reload(config) + + +def _clear_orch_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("GPU_SERVER_URL", raising=False) + monkeypatch.delenv("CF_ORCH_URL", raising=False) + monkeypatch.delenv("CF_LICENSE_KEY", raising=False) + + +def test_gpu_server_url_prefers_explicit_env_var(monkeypatch): + _clear_orch_env(monkeypatch) + monkeypatch.setenv("GPU_SERVER_URL", "http://my-rig:7700") + monkeypatch.setenv("CF_ORCH_URL", "http://should-be-ignored:7700") + + importlib.reload(config) + + assert config.GPU_SERVER_URL == "http://my-rig:7700" + + +def test_gpu_server_url_falls_back_to_cf_orch_url_legacy_alias(monkeypatch): + _clear_orch_env(monkeypatch) + monkeypatch.setenv("CF_ORCH_URL", "http://legacy-coordinator:7700") + + importlib.reload(config) + + assert config.GPU_SERVER_URL == "http://legacy-coordinator:7700" + + +def test_gpu_server_url_defaults_when_license_key_present(monkeypatch): + _clear_orch_env(monkeypatch) + monkeypatch.setenv("CF_LICENSE_KEY", "test-key-123") + + importlib.reload(config) + + assert config.GPU_SERVER_URL == "https://orch.circuitforge.tech" + + +def test_gpu_server_url_none_when_nothing_set(monkeypatch): + _clear_orch_env(monkeypatch) + + importlib.reload(config) + + assert config.GPU_SERVER_URL is None + + +def test_gpu_server_url_writes_back_to_cf_orch_url_for_legacy_callers(monkeypatch): + """get_llm_config() and app/api/chat.py read os.environ["CF_ORCH_URL"] directly — + the write-back must happen so they keep working without code changes.""" + import os + + _clear_orch_env(monkeypatch) + monkeypatch.setenv("GPU_SERVER_URL", "http://my-rig:7700") + + importlib.reload(config) + + assert os.environ.get("CF_ORCH_URL") == "http://my-rig:7700" + + +def test_gpu_server_url_unset_does_not_write_back(monkeypatch): + import os + + _clear_orch_env(monkeypatch) + + importlib.reload(config) + + assert "CF_ORCH_URL" not in os.environ diff --git a/tests/test_library_api.py b/tests/test_library_api.py index 7171406..cc77924 100644 --- a/tests/test_library_api.py +++ b/tests/test_library_api.py @@ -45,25 +45,25 @@ def test_delete_nonexistent_returns_404(client): assert resp.status_code == 404 -def test_reingest_returns_task_id(client, test_db, tmp_path): +def test_reshelve_returns_task_id(client, test_db, tmp_path): pdf_path = str(tmp_path / "books" / "test.pdf") open(pdf_path, "wb").write(b"%PDF-1.4") doc_id = _add_doc(test_db, "Test Book", pdf_path) - resp = client.post(f"/api/library/{doc_id}/reingest") + resp = client.post(f"/api/library/{doc_id}/reshelve") assert resp.status_code == 202 assert "task_id" in resp.json() -def test_reingest_updates_status_to_processing(client, test_db, tmp_path): +def test_reshelve_updates_status_to_processing(client, test_db, tmp_path): from pathlib import Path pdf_path = str(tmp_path / "books" / "dm_guide.pdf") Path(pdf_path).write_bytes(b"%PDF-1.4 empty fixture") doc_id = _add_doc(test_db, "DM Guide", pdf_path) - resp = client.post(f"/api/library/{doc_id}/reingest") + resp = client.post(f"/api/library/{doc_id}/reshelve") assert resp.status_code == 202 - # Document should be in processing state (or beyond if stub ingest ran instantly) + # Document should be in processing state (or beyond if stub shelve ran instantly) status_resp = client.get(f"/api/library/{doc_id}/status") assert status_resp.json()["status"] in ("processing", "error", "ready") diff --git a/tests/test_ingest.py b/tests/test_shelve.py similarity index 83% rename from tests/test_ingest.py rename to tests/test_shelve.py index 2e42dac..7266320 100644 --- a/tests/test_ingest.py +++ b/tests/test_shelve.py @@ -1,5 +1,5 @@ -# tests/test_ingest.py -"""Unit tests for scripts/ingest_pdf.py.""" +# tests/test_shelve.py +"""Unit tests for scripts/shelve_pdf.py.""" from __future__ import annotations import sqlite3 @@ -8,11 +8,11 @@ from unittest.mock import MagicMock, patch import pytest -from scripts.ingest_pdf import run +from scripts.shelve_pdf import run @pytest.fixture -def ingest_db(tmp_path) -> tuple[str, str]: +def shelve_db(tmp_path) -> tuple[str, str]: db_path = str(tmp_path / "test.db") schema = Path("migrations/001_initial_schema.sql").read_text() conn = sqlite3.connect(db_path) @@ -35,8 +35,8 @@ def _make_mock_chunk(page_number: int = 1, text: str = "Some page text about rul return chunk -def test_ingest_sets_status_ready_on_success(ingest_db): - db_path, vec_db_path = ingest_db +def test_shelve_sets_status_ready_on_success(shelve_db): + db_path, vec_db_path = shelve_db mock_extractor = MagicMock() mock_extractor.chunk_pages.return_value = [_make_mock_chunk()] @@ -51,8 +51,8 @@ def test_ingest_sets_status_ready_on_success(ingest_db): assert row[1] == 1 -def test_ingest_stores_page_chunks(ingest_db): - db_path, vec_db_path = ingest_db +def test_shelve_stores_page_chunks(shelve_db): + db_path, vec_db_path = shelve_db mock_extractor = MagicMock() chunks = [_make_mock_chunk(page_number=i + 1, text=f"Page {i+1} text content.") for i in range(3)] @@ -71,11 +71,11 @@ def test_ingest_stores_page_chunks(ingest_db): assert "Page 1" in rows[0][1] -def test_ingest_sets_error_status_on_failure(ingest_db): - db_path, vec_db_path = ingest_db +def test_shelve_sets_error_status_on_failure(shelve_db): + db_path, vec_db_path = shelve_db with patch("circuitforge_core.documents.pdf.PDFExtractor", side_effect=RuntimeError("PDF corrupt")): - from scripts.ingest_pdf import run + from scripts.shelve_pdf import run with pytest.raises(RuntimeError): run(doc_id="d1", file_path="bad.pdf", db_path=db_path, vec_db_path=vec_db_path) @@ -86,9 +86,9 @@ def test_ingest_sets_error_status_on_failure(ingest_db): assert "PDF corrupt" in row[1] -def test_ingest_skips_embeddings_without_ollama_url(ingest_db, monkeypatch): +def test_shelve_skips_embeddings_without_ollama_url(shelve_db, monkeypatch): """When PAGEPIPER_OLLAMA_URL is unset, no vec DB file should be created.""" - db_path, vec_db_path = ingest_db + db_path, vec_db_path = shelve_db monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False) mock_extractor = MagicMock() @@ -111,20 +111,20 @@ def test_ingest_skips_embeddings_without_ollama_url(ingest_db, monkeypatch): assert chunk_count == 1 -def test_ingest_replaces_existing_chunks_on_reingest(ingest_db): - """Re-running ingest for the same doc_id replaces old page_chunks.""" - db_path, vec_db_path = ingest_db +def test_shelve_replaces_existing_chunks_on_reshelve(shelve_db): + """Re-running shelve for the same doc_id replaces old page_chunks.""" + db_path, vec_db_path = shelve_db mock_extractor = MagicMock() - # First ingest: 3 pages + # First shelve: 3 pages mock_extractor.chunk_pages.return_value = [ _make_mock_chunk(page_number=i + 1, text=f"Original page {i+1}.") for i in range(3) ] with patch("circuitforge_core.documents.pdf.PDFExtractor", return_value=mock_extractor): run(doc_id="d1", file_path="test.pdf", db_path=db_path, vec_db_path=vec_db_path) - # Second ingest: 1 page (simulating a re-ingest after file change) + # Second shelve: 1 page (simulating a re-shelve after file change) mock_extractor.chunk_pages.return_value = [_make_mock_chunk(text="Updated single page.")] with patch("circuitforge_core.documents.pdf.PDFExtractor", return_value=mock_extractor): run(doc_id="d1", file_path="test.pdf", db_path=db_path, vec_db_path=vec_db_path) diff --git a/tests/test_shelve_docx.py b/tests/test_shelve_docx.py new file mode 100644 index 0000000..f11c1cf --- /dev/null +++ b/tests/test_shelve_docx.py @@ -0,0 +1,129 @@ +# tests/test_shelve_docx.py +"""Unit tests for scripts/shelve_docx.py.""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from scripts.shelve_docx import run + + +@pytest.fixture +def shelve_db(tmp_path) -> tuple[str, str]: + db_path = str(tmp_path / "test.db") + schema = Path("migrations/001_initial_schema.sql").read_text() + conn = sqlite3.connect(db_path) + conn.executescript(schema) + conn.execute( + "INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Test','test.docx','pending')" + ) + conn.commit() + conn.close() + vec_db_path = str(tmp_path / "vecs.db") + return db_path, vec_db_path + + +def _make_docx(path: Path, with_headings: bool = True, with_table: bool = False) -> None: + import docx + + doc = docx.Document() + if with_headings: + doc.add_heading("Setting the IP", level=1) + doc.add_paragraph("Connect to the device over the service port.") + doc.add_heading("Verifying the Change", level=1) + doc.add_paragraph("Ping the new address to confirm.") + else: + doc.add_paragraph("Some unstructured procedure text with no headings at all.") + if with_table: + table = doc.add_table(rows=2, cols=2) + table.rows[0].cells[0].text = "Field" + table.rows[0].cells[1].text = "Value" + table.rows[1].cells[0].text = "IP Address" + table.rows[1].cells[1].text = "10.0.0.5" + doc.save(str(path)) + + +def test_shelve_docx_sets_status_ready_on_success(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + docx_path = tmp_path / "test.docx" + _make_docx(docx_path) + + run(doc_id="d1", file_path=str(docx_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, page_count FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "ready" + assert row[1] == 2 # one chunk per heading section + + +def test_shelve_docx_splits_by_heading(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + docx_path = tmp_path / "test.docx" + _make_docx(docx_path) + + run(doc_id="d1", file_path=str(docx_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + assert len(rows) == 2 + assert "Setting the IP" in rows[0][0] + assert "Verifying the Change" in rows[1][0] + + +def test_shelve_docx_falls_back_to_wordcount_chunks_without_headings(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + docx_path = tmp_path / "test.docx" + _make_docx(docx_path, with_headings=False) + + run(doc_id="d1", file_path=str(docx_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute("SELECT text FROM page_chunks WHERE doc_id='d1'").fetchall() + conn.close() + assert len(rows) == 1 + assert "unstructured procedure text" in rows[0][0] + + +def test_shelve_docx_serializes_tables_inline(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + docx_path = tmp_path / "test.docx" + _make_docx(docx_path, with_table=True) + + run(doc_id="d1", file_path=str(docx_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute("SELECT text FROM page_chunks WHERE doc_id='d1'").fetchall() + conn.close() + joined = "\n".join(r[0] for r in rows) + assert "IP Address | 10.0.0.5" in joined + + +def test_shelve_docx_sets_error_status_on_failure(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + missing_path = tmp_path / "does-not-exist.docx" + + with pytest.raises(Exception): + run(doc_id="d1", file_path=str(missing_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, error_msg FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "error" + assert row[1] + + +def test_shelve_docx_skips_embeddings_without_ollama_url(shelve_db, tmp_path, monkeypatch): + db_path, vec_db_path = shelve_db + monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False) + docx_path = tmp_path / "test.docx" + _make_docx(docx_path) + + run(doc_id="d1", file_path=str(docx_path), db_path=db_path, vec_db_path=vec_db_path) + + assert not Path(vec_db_path).exists(), "vec DB should not be created without OLLAMA_URL" diff --git a/tests/test_shelve_numbers.py b/tests/test_shelve_numbers.py new file mode 100644 index 0000000..b193da8 --- /dev/null +++ b/tests/test_shelve_numbers.py @@ -0,0 +1,118 @@ +# tests/test_shelve_numbers.py +"""Unit tests for scripts/shelve_numbers.py. + +soffice isn't available in the test environment, so _convert_to_xlsx is +mocked to copy a pre-built XLSX fixture into the expected output path — +everything downstream (extraction/chunking/storage) runs for real via +scripts.shelve_xlsx.run. +""" +from __future__ import annotations + +import shutil +import sqlite3 +import subprocess +from pathlib import Path +from unittest.mock import patch + +import pytest + +from scripts.shelve_numbers import run + + +@pytest.fixture +def shelve_db(tmp_path) -> tuple[str, str]: + db_path = str(tmp_path / "test.db") + schema = Path("migrations/001_initial_schema.sql").read_text() + conn = sqlite3.connect(db_path) + conn.executescript(schema) + conn.execute( + "INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Test','test.numbers','pending')" + ) + conn.commit() + conn.close() + vec_db_path = str(tmp_path / "vecs.db") + return db_path, vec_db_path + + +def _make_fixture_xlsx(path: Path) -> None: + import openpyxl + + wb = openpyxl.Workbook() + ws = wb.active + ws.title = "Config" + ws.append(["Field", "Value"]) + ws.append(["Default IP", "192.168.1.50"]) + wb.save(str(path)) + + +def _fake_soffice_convert(fixture_xlsx: Path): + def _fake_run(cmd, **kwargs): + out_dir = Path(cmd[cmd.index("--outdir") + 1]) + numbers_path = Path(cmd[-1]) + shutil.copy(fixture_xlsx, out_dir / f"{numbers_path.stem}.xlsx") + return subprocess.CompletedProcess(cmd, returncode=0, stdout="", stderr="") + + return _fake_run + + +def test_shelve_numbers_converts_and_shelves_successfully(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + numbers_path = tmp_path / "test.numbers" + numbers_path.write_bytes(b"not a real numbers bundle, just needs to exist") + + fixture_xlsx = tmp_path / "fixture.xlsx" + _make_fixture_xlsx(fixture_xlsx) + + with patch("scripts.shelve_numbers.subprocess.run", side_effect=_fake_soffice_convert(fixture_xlsx)): + run(doc_id="d1", file_path=str(numbers_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, page_count FROM documents WHERE id='d1'").fetchone() + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + assert row[0] == "ready" + assert row[1] == 1 + assert "Default IP | 192.168.1.50" in rows[0][0] + + +def test_shelve_numbers_sets_error_status_on_conversion_failure(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + numbers_path = tmp_path / "test.numbers" + numbers_path.write_bytes(b"not a real numbers bundle") + + fake_result = subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr="soffice: unsupported document format" + ) + with patch("scripts.shelve_numbers.subprocess.run", return_value=fake_result): + with pytest.raises(RuntimeError, match="LibreOffice conversion failed"): + run(doc_id="d1", file_path=str(numbers_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, error_msg FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "error" + assert "unsupported document format" in row[1] + + +def test_shelve_numbers_cleans_up_temp_dir(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + numbers_path = tmp_path / "test.numbers" + numbers_path.write_bytes(b"not a real numbers bundle") + + fixture_xlsx = tmp_path / "fixture.xlsx" + _make_fixture_xlsx(fixture_xlsx) + + captured_tmp_dirs: list[str] = [] + real_convert = _fake_soffice_convert(fixture_xlsx) + + def _tracking_run(cmd, **kwargs): + captured_tmp_dirs.append(cmd[cmd.index("--outdir") + 1]) + return real_convert(cmd, **kwargs) + + with patch("scripts.shelve_numbers.subprocess.run", side_effect=_tracking_run): + run(doc_id="d1", file_path=str(numbers_path), db_path=db_path, vec_db_path=vec_db_path) + + assert captured_tmp_dirs + assert not Path(captured_tmp_dirs[0]).exists(), "temp conversion dir should be cleaned up" diff --git a/tests/test_shelve_ods.py b/tests/test_shelve_ods.py new file mode 100644 index 0000000..788465e --- /dev/null +++ b/tests/test_shelve_ods.py @@ -0,0 +1,135 @@ +# tests/test_shelve_ods.py +"""Unit tests for scripts/shelve_ods.py.""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from scripts.shelve_ods import run, ROWS_PER_CHUNK + + +@pytest.fixture +def shelve_db(tmp_path) -> tuple[str, str]: + db_path = str(tmp_path / "test.db") + schema = Path("migrations/001_initial_schema.sql").read_text() + conn = sqlite3.connect(db_path) + conn.executescript(schema) + conn.execute( + "INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Test','test.ods','pending')" + ) + conn.commit() + conn.close() + vec_db_path = str(tmp_path / "vecs.db") + return db_path, vec_db_path + + +def _make_ods(path: Path, big_sheet: bool = False) -> None: + from odf.opendocument import OpenDocumentSpreadsheet + from odf.table import Table, TableRow, TableCell + from odf.text import P + + def _row(vals): + row = TableRow() + for v in vals: + cell = TableCell() + cell.addElement(P(text=v)) + row.addElement(cell) + return row + + doc = OpenDocumentSpreadsheet() + + t1 = Table(name="Config") + t1.addElement(_row(["Field", "Value"])) + t1.addElement(_row(["Default IP", "192.168.1.50"])) + t1.addElement(_row(["AVC-X Model", "AVC-X200"])) + doc.spreadsheet.addElement(t1) + + t2 = Table(name="Parts List") + t2.addElement(_row(["Part Number", "Description"])) + if big_sheet: + for i in range(ROWS_PER_CHUNK + 50): + t2.addElement(_row([f"P{i:04d}", f"Widget {i}"])) + else: + t2.addElement(_row(["P0001", "Filter cartridge"])) + doc.spreadsheet.addElement(t2) + + doc.save(str(path)) + + +def test_shelve_ods_sets_status_ready_on_success(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + ods_path = tmp_path / "test.ods" + _make_ods(ods_path) + + run(doc_id="d1", file_path=str(ods_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, page_count FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "ready" + assert row[1] == 2 + + +def test_shelve_ods_serializes_rows_with_header(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + ods_path = tmp_path / "test.ods" + _make_ods(ods_path) + + run(doc_id="d1", file_path=str(ods_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + assert "Sheet: Config" in rows[0][0] + assert "Field | Value" in rows[0][0] + assert "Default IP | 192.168.1.50" in rows[0][0] + assert "Sheet: Parts List" in rows[1][0] + assert "P0001 | Filter cartridge" in rows[1][0] + + +def test_shelve_ods_splits_large_sheet_into_row_windows(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + ods_path = tmp_path / "test.ods" + _make_ods(ods_path, big_sheet=True) + + run(doc_id="d1", file_path=str(ods_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + assert len(rows) >= 3 + parts_list_chunks = [r[0] for r in rows if "Sheet: Parts List" in r[0]] + assert len(parts_list_chunks) >= 2 + for chunk_text in parts_list_chunks: + assert "Part Number | Description" in chunk_text + + +def test_shelve_ods_sets_error_status_on_failure(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + missing_path = tmp_path / "does-not-exist.ods" + + with pytest.raises(Exception): + run(doc_id="d1", file_path=str(missing_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, error_msg FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "error" + assert row[1] + + +def test_shelve_ods_skips_embeddings_without_ollama_url(shelve_db, tmp_path, monkeypatch): + db_path, vec_db_path = shelve_db + monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False) + ods_path = tmp_path / "test.ods" + _make_ods(ods_path) + + run(doc_id="d1", file_path=str(ods_path), db_path=db_path, vec_db_path=vec_db_path) + + assert not Path(vec_db_path).exists(), "vec DB should not be created without OLLAMA_URL" diff --git a/tests/test_shelve_odt.py b/tests/test_shelve_odt.py new file mode 100644 index 0000000..c66e7b6 --- /dev/null +++ b/tests/test_shelve_odt.py @@ -0,0 +1,135 @@ +# tests/test_shelve_odt.py +"""Unit tests for scripts/shelve_odt.py.""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from scripts.shelve_odt import run + + +@pytest.fixture +def shelve_db(tmp_path) -> tuple[str, str]: + db_path = str(tmp_path / "test.db") + schema = Path("migrations/001_initial_schema.sql").read_text() + conn = sqlite3.connect(db_path) + conn.executescript(schema) + conn.execute( + "INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Test','test.odt','pending')" + ) + conn.commit() + conn.close() + vec_db_path = str(tmp_path / "vecs.db") + return db_path, vec_db_path + + +def _make_odt(path: Path, with_headings: bool = True, with_table: bool = False) -> None: + from odf.opendocument import OpenDocumentText + from odf.text import H, P + from odf.table import Table, TableRow, TableCell + + doc = OpenDocumentText() + if with_headings: + doc.text.addElement(H(outlinelevel=1, text="Setting the IP")) + doc.text.addElement(P(text="Connect to the device over the service port.")) + doc.text.addElement(H(outlinelevel=1, text="Verifying the Change")) + doc.text.addElement(P(text="Ping the new address to confirm.")) + else: + doc.text.addElement(P(text="Some unstructured procedure text with no headings at all.")) + if with_table: + table = Table(name="T1") + for rowvals in [["Field", "Value"], ["IP Address", "10.0.0.5"]]: + row = TableRow() + for v in rowvals: + cell = TableCell() + cell.addElement(P(text=v)) + row.addElement(cell) + table.addElement(row) + doc.text.addElement(table) + doc.save(str(path)) + + +def test_shelve_odt_sets_status_ready_on_success(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + odt_path = tmp_path / "test.odt" + _make_odt(odt_path) + + run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, page_count FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "ready" + assert row[1] == 2 # one chunk per heading section + + +def test_shelve_odt_splits_by_heading(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + odt_path = tmp_path / "test.odt" + _make_odt(odt_path) + + run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + assert len(rows) == 2 + assert "Setting the IP" in rows[0][0] + assert "Verifying the Change" in rows[1][0] + + +def test_shelve_odt_falls_back_to_wordcount_chunks_without_headings(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + odt_path = tmp_path / "test.odt" + _make_odt(odt_path, with_headings=False) + + run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute("SELECT text FROM page_chunks WHERE doc_id='d1'").fetchall() + conn.close() + assert len(rows) == 1 + assert "unstructured procedure text" in rows[0][0] + + +def test_shelve_odt_serializes_tables_inline(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + odt_path = tmp_path / "test.odt" + _make_odt(odt_path, with_table=True) + + run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute("SELECT text FROM page_chunks WHERE doc_id='d1'").fetchall() + conn.close() + joined = "\n".join(r[0] for r in rows) + assert "IP Address | 10.0.0.5" in joined + + +def test_shelve_odt_sets_error_status_on_failure(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + missing_path = tmp_path / "does-not-exist.odt" + + with pytest.raises(Exception): + run(doc_id="d1", file_path=str(missing_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, error_msg FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "error" + assert row[1] + + +def test_shelve_odt_skips_embeddings_without_ollama_url(shelve_db, tmp_path, monkeypatch): + db_path, vec_db_path = shelve_db + monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False) + odt_path = tmp_path / "test.odt" + _make_odt(odt_path) + + run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path) + + assert not Path(vec_db_path).exists(), "vec DB should not be created without OLLAMA_URL" diff --git a/tests/test_shelve_pages.py b/tests/test_shelve_pages.py new file mode 100644 index 0000000..2f5bc52 --- /dev/null +++ b/tests/test_shelve_pages.py @@ -0,0 +1,122 @@ +# tests/test_shelve_pages.py +"""Unit tests for scripts/shelve_pages.py. + +soffice isn't available in the test environment, so _convert_to_odt is +mocked to copy a pre-built ODT fixture into the expected output path — +everything downstream (extraction/chunking/storage) runs for real via +scripts.shelve_odt.run. +""" +from __future__ import annotations + +import shutil +import sqlite3 +import subprocess +from pathlib import Path +from unittest.mock import patch + +import pytest + +from scripts.shelve_pages import run + + +@pytest.fixture +def shelve_db(tmp_path) -> tuple[str, str]: + db_path = str(tmp_path / "test.db") + schema = Path("migrations/001_initial_schema.sql").read_text() + conn = sqlite3.connect(db_path) + conn.executescript(schema) + conn.execute( + "INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Test','test.pages','pending')" + ) + conn.commit() + conn.close() + vec_db_path = str(tmp_path / "vecs.db") + return db_path, vec_db_path + + +def _make_fixture_odt(path: Path) -> None: + from odf.opendocument import OpenDocumentText + from odf.text import H, P + + doc = OpenDocumentText() + doc.text.addElement(H(outlinelevel=1, text="Setting the IP")) + doc.text.addElement(P(text="Connect to the device over the service port.")) + doc.text.addElement(H(outlinelevel=1, text="Verifying the Change")) + doc.text.addElement(P(text="Ping the new address to confirm.")) + doc.save(str(path)) + + +def _fake_soffice_convert(fixture_odt: Path): + """Return a fake subprocess.run that copies fixture_odt into --outdir.""" + + def _fake_run(cmd, **kwargs): + out_dir = Path(cmd[cmd.index("--outdir") + 1]) + pages_path = Path(cmd[-1]) + shutil.copy(fixture_odt, out_dir / f"{pages_path.stem}.odt") + return subprocess.CompletedProcess(cmd, returncode=0, stdout="", stderr="") + + return _fake_run + + +def test_shelve_pages_converts_and_shelves_successfully(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + pages_path = tmp_path / "test.pages" + pages_path.write_bytes(b"not a real pages bundle, just needs to exist") + + fixture_odt = tmp_path / "fixture.odt" + _make_fixture_odt(fixture_odt) + + with patch("scripts.shelve_pages.subprocess.run", side_effect=_fake_soffice_convert(fixture_odt)): + run(doc_id="d1", file_path=str(pages_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, page_count FROM documents WHERE id='d1'").fetchone() + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + assert row[0] == "ready" + assert row[1] == 2 + assert "Setting the IP" in rows[0][0] + assert "Verifying the Change" in rows[1][0] + + +def test_shelve_pages_sets_error_status_on_conversion_failure(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + pages_path = tmp_path / "test.pages" + pages_path.write_bytes(b"not a real pages bundle") + + fake_result = subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr="soffice: unsupported document format" + ) + with patch("scripts.shelve_pages.subprocess.run", return_value=fake_result): + with pytest.raises(RuntimeError, match="LibreOffice conversion failed"): + run(doc_id="d1", file_path=str(pages_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, error_msg FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "error" + assert "unsupported document format" in row[1] + + +def test_shelve_pages_cleans_up_temp_dir(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + pages_path = tmp_path / "test.pages" + pages_path.write_bytes(b"not a real pages bundle") + + fixture_odt = tmp_path / "fixture.odt" + _make_fixture_odt(fixture_odt) + + captured_tmp_dirs: list[str] = [] + real_convert = _fake_soffice_convert(fixture_odt) + + def _tracking_run(cmd, **kwargs): + captured_tmp_dirs.append(cmd[cmd.index("--outdir") + 1]) + return real_convert(cmd, **kwargs) + + with patch("scripts.shelve_pages.subprocess.run", side_effect=_tracking_run): + run(doc_id="d1", file_path=str(pages_path), db_path=db_path, vec_db_path=vec_db_path) + + assert captured_tmp_dirs + assert not Path(captured_tmp_dirs[0]).exists(), "temp conversion dir should be cleaned up" diff --git a/tests/test_shelve_xlsx.py b/tests/test_shelve_xlsx.py new file mode 100644 index 0000000..494aea4 --- /dev/null +++ b/tests/test_shelve_xlsx.py @@ -0,0 +1,125 @@ +# tests/test_shelve_xlsx.py +"""Unit tests for scripts/shelve_xlsx.py.""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from scripts.shelve_xlsx import run, ROWS_PER_CHUNK + + +@pytest.fixture +def shelve_db(tmp_path) -> tuple[str, str]: + db_path = str(tmp_path / "test.db") + schema = Path("migrations/001_initial_schema.sql").read_text() + conn = sqlite3.connect(db_path) + conn.executescript(schema) + conn.execute( + "INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Test','test.xlsx','pending')" + ) + conn.commit() + conn.close() + vec_db_path = str(tmp_path / "vecs.db") + return db_path, vec_db_path + + +def _make_xlsx(path: Path, big_sheet: bool = False) -> None: + import openpyxl + + wb = openpyxl.Workbook() + ws1 = wb.active + ws1.title = "Config" + ws1.append(["Field", "Value"]) + ws1.append(["Default IP", "192.168.1.50"]) + ws1.append(["AVC-X Model", "AVC-X200"]) + + ws2 = wb.create_sheet("Parts List") + ws2.append(["Part Number", "Description"]) + if big_sheet: + for i in range(ROWS_PER_CHUNK + 50): + ws2.append([f"P{i:04d}", f"Widget {i}"]) + else: + ws2.append(["P0001", "Filter cartridge"]) + + wb.save(str(path)) + + +def test_shelve_xlsx_sets_status_ready_on_success(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + xlsx_path = tmp_path / "test.xlsx" + _make_xlsx(xlsx_path) + + run(doc_id="d1", file_path=str(xlsx_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, page_count FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "ready" + assert row[1] == 2 # one chunk per sheet, both small + + +def test_shelve_xlsx_serializes_rows_with_header(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + xlsx_path = tmp_path / "test.xlsx" + _make_xlsx(xlsx_path) + + run(doc_id="d1", file_path=str(xlsx_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + assert "Sheet: Config" in rows[0][0] + assert "Field | Value" in rows[0][0] + assert "Default IP | 192.168.1.50" in rows[0][0] + assert "Sheet: Parts List" in rows[1][0] + assert "P0001 | Filter cartridge" in rows[1][0] + + +def test_shelve_xlsx_splits_large_sheet_into_row_windows(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + xlsx_path = tmp_path / "test.xlsx" + _make_xlsx(xlsx_path, big_sheet=True) + + run(doc_id="d1", file_path=str(xlsx_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + rows = conn.execute( + "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number" + ).fetchall() + conn.close() + # Config sheet (1 chunk) + Parts List sheet split across >=2 row-window chunks + assert len(rows) >= 3 + parts_list_chunks = [r[0] for r in rows if "Sheet: Parts List" in r[0]] + assert len(parts_list_chunks) >= 2 + # header row repeated in every window + for chunk_text in parts_list_chunks: + assert "Part Number | Description" in chunk_text + + +def test_shelve_xlsx_sets_error_status_on_failure(shelve_db, tmp_path): + db_path, vec_db_path = shelve_db + missing_path = tmp_path / "does-not-exist.xlsx" + + with pytest.raises(Exception): + run(doc_id="d1", file_path=str(missing_path), db_path=db_path, vec_db_path=vec_db_path) + + conn = sqlite3.connect(db_path) + row = conn.execute("SELECT status, error_msg FROM documents WHERE id='d1'").fetchone() + conn.close() + assert row[0] == "error" + assert row[1] + + +def test_shelve_xlsx_skips_embeddings_without_ollama_url(shelve_db, tmp_path, monkeypatch): + db_path, vec_db_path = shelve_db + monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False) + xlsx_path = tmp_path / "test.xlsx" + _make_xlsx(xlsx_path) + + run(doc_id="d1", file_path=str(xlsx_path), db_path=db_path, vec_db_path=vec_db_path) + + assert not Path(vec_db_path).exists(), "vec DB should not be created without OLLAMA_URL" diff --git a/web/src/api.ts b/web/src/api.ts index b7f62f5..4ff418c 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -62,8 +62,8 @@ export const api = { if (!r.ok) throw new Error(await r.text()) return r.json() }, - async reingestDocument(docId: string): Promise<{ task_id: string }> { - const r = await fetch(`${BASE}/api/library/${docId}/reingest`, { method: "POST" }) + async reshelveDocument(docId: string): Promise<{ task_id: string }> { + const r = await fetch(`${BASE}/api/library/${docId}/reshelve`, { method: "POST" }) if (!r.ok) throw new Error(await r.text()) return r.json() }, @@ -79,7 +79,7 @@ export const api = { return r.json() }, async getTaskStatus(taskId: string): Promise { - const r = await fetch(`${BASE}/api/ingest/${taskId}`) + const r = await fetch(`${BASE}/api/shelve/${taskId}`) if (!r.ok) throw new Error(await r.text()) return r.json() }, diff --git a/web/src/components/DocumentCard.vue b/web/src/components/DocumentCard.vue index 488c376..31a5462 100644 --- a/web/src/components/DocumentCard.vue +++ b/web/src/components/DocumentCard.vue @@ -5,7 +5,7 @@
{{ displayPageCount }} pages
{{ shortPath }}
-
+
{{ progressLabel }} {{ progressPct }}% @@ -18,7 +18,7 @@

{{ errorMsg ?? 'Indexing failed.' }}

- @@ -32,7 +32,7 @@ import type { Document } from "@/api" import { api } from "@/api" const props = defineProps<{ doc: Document }>() -const emit = defineEmits<{ reingest: [id: string]; delete: [id: string]; refresh: [] }>() +const emit = defineEmits<{ reshelve: [id: string]; delete: [id: string]; refresh: [] }>() const shortPath = computed(() => { const parts = props.doc.file_path.split("/") @@ -126,7 +126,7 @@ onUnmounted(stopPoll) .doc-error { color: var(--color-error); font-size: 0.8rem; } /* Progress bar */ -.ingest-progress { margin-top: 0.25rem; } +.shelve-progress { margin-top: 0.25rem; } .progress-label { display: flex; justify-content: space-between; font-size: 0.78rem; color: var(--color-text-muted); margin-bottom: 4px; diff --git a/web/src/components/IngestProgress.vue b/web/src/components/ShelveProgress.vue similarity index 96% rename from web/src/components/IngestProgress.vue rename to web/src/components/ShelveProgress.vue index 650717c..7698c61 100644 --- a/web/src/components/IngestProgress.vue +++ b/web/src/components/ShelveProgress.vue @@ -1,5 +1,5 @@