From b16620385a45a5fa3636f0bdcbc63630b06eaa0b Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Sat, 16 May 2026 21:05:02 -0700 Subject: [PATCH 1/4] feat: route pagepiper.rag_query through cf-orch task assignment layer (closes #7) When CF_ORCH_URL is set, chat now calls CFOrchClient.task_allocate("pagepiper", "rag_query") instead of routing through LLMRouter with an explicit model. The coordinator resolves the assignment (granite-4.1-8b via assignments.yaml) and returns an allocated URL; pagepiper wraps it in a minimal LLMRouter config for the Synthesizer. Falls back to LLMRouter on TaskNotFound or allocation failure, so standalone Ollama installs are unaffected. Extracts _run_chat() and _build_llm_for_alloc() helpers to keep the endpoint body readable regardless of which path fires. --- app/api/chat.py | 117 +++++++++++++++++++++++++++++++----------------- 1 file changed, 77 insertions(+), 40 deletions(-) diff --git a/app/api/chat.py b/app/api/chat.py index 0fe3815..9226578 100644 --- a/app/api/chat.py +++ b/app/api/chat.py @@ -47,7 +47,7 @@ class ChatFeedbackRequest(BaseModel): def _get_llm_router(): - """Return LLMRouter if Ollama configured, else None.""" + """Return LLMRouter if Ollama/cf-orch configured, else None.""" from app.config import get_llm_config cfg = get_llm_config() @@ -58,6 +58,65 @@ def _get_llm_router(): return LLMRouter(cfg) +def _build_llm_for_alloc(alloc) -> "LLMRouter": + """Wrap a cf-orch task allocation in a minimal LLMRouter for completion calls.""" + from circuitforge_core.llm import LLMRouter + + base_url = alloc.url.rstrip("/") + if not base_url.endswith("/v1"): + base_url += "/v1" + cfg = { + "fallback_order": ["orch_task"], + "backends": { + "orch_task": { + "type": "openai_compat", + "base_url": base_url, + "model": alloc.model or "default", + "supports_images": False, + } + }, + } + return LLMRouter(cfg) + + +def _run_chat(req: "ChatRequest", ctx: "UserCtx", llm) -> "ChatResponse": + retriever = Retriever(ctx.bm25) + chunks = retriever.hybrid_search( + query=req.message, + top_k=req.top_k, + doc_ids=req.doc_ids, + db_path=ctx.db_path, + vec_db_path=ctx.vec_db_path, + llm=llm, + ) + if not chunks: + return ChatResponse( + answer=( + "I couldn't find any relevant passages. " + "Try a different query or check which documents are indexed." + ), + citations=[], + ) + synth = Synthesizer(llm) + result = synth.synthesize( + message=req.message, + history=[t.model_dump() for t in req.history], + chunks=chunks, + ) + return ChatResponse( + answer=result.answer, + citations=[ + { + "doc_id": c.doc_id, + "page_number": c.page_number, + "snippet": c.snippet, + "bm25_score": c.bm25_score, + } + for c in result.citations + ], + ) + + def _require_llm(): """Return LLMRouter or raise 402.""" llm = _get_llm_router() @@ -81,46 +140,24 @@ def chat( ctx: UserCtx = Depends(get_user_ctx), _tier: str = Depends(require_paid_tier), ) -> ChatResponse: + orch_url = os.environ.get("CF_ORCH_URL", "").strip() + + if orch_url: + try: + from circuitforge_orch.client import CFOrchClient, TaskNotFound # type: ignore[import] + api_key = os.environ.get("CF_LICENSE_KEY", "") + client = CFOrchClient(orch_url, api_key=api_key) + with client.task_allocate("pagepiper", "rag_query") as alloc: + llm = _build_llm_for_alloc(alloc) + return _run_chat(req, ctx, llm) + except Exception as exc: + logger.warning( + "cf-orch task allocation for pagepiper.rag_query failed, falling back to LLMRouter: %s", + exc, + ) + llm = _require_llm() - - retriever = Retriever(ctx.bm25) - chunks = retriever.hybrid_search( - query=req.message, - top_k=req.top_k, - doc_ids=req.doc_ids, - db_path=ctx.db_path, - vec_db_path=ctx.vec_db_path, - llm=llm, - ) - - if not chunks: - return ChatResponse( - answer=( - "I couldn't find any relevant passages. " - "Try a different query or check which documents are indexed." - ), - citations=[], - ) - - synth = Synthesizer(llm) - result = synth.synthesize( - message=req.message, - history=[t.model_dump() for t in req.history], - chunks=chunks, - ) - - return ChatResponse( - answer=result.answer, - citations=[ - { - "doc_id": c.doc_id, - "page_number": c.page_number, - "snippet": c.snippet, - "bm25_score": c.bm25_score, - } - for c in result.citations - ], - ) + return _run_chat(req, ctx, llm) @router.get("/feedback/status") -- 2.45.2 From f941ebdeeb5a83d12d1aa57bc4a7842ecd0a0d1a Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Fri, 10 Jul 2026 13:58:43 -0700 Subject: [PATCH 2/4] feat: add ODT and Apple Pages document support, wire DOCX into UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends Pagepiper's document shelving pipeline (renamed from "ingest" — see below) to cover the formats most likely to appear in a real-world engineering document corpus, prompted by scoping a STERIS licensing pitch that needs DOCX/ODT coverage. - Rename the ingest pipeline to "shelve" throughout (scripts/, app/api, tests, docs, frontend). "Glean" (Turnstone's term) was considered and rejected — that's a harvest metaphor for log/knowledge extraction, not a fit for documents entering a library. Documented as a general CF naming principle in the org-level CLAUDE.md. - Wire DOCX into the upload/scan UI, README, and docs — the extraction logic (heading-based chunking, table serialization) already existed but wasn't exposed to users or covered by tests. - Add ODT support via odfpy, mirroring DOCX's chunking strategy. - Add Apple Pages support via headless LibreOffice conversion to ODT. No maintained Python library parses the IWA format directly; libreoffice bundles libetonyek, the only real open-source Pages parser. Adds libreoffice-writer to the Docker image (~300-400MB) for this. - 24 new/updated tests across shelve_docx, shelve_odt, and shelve_pages; full suite (72 tests) passing. Known gaps not addressed here: no Windchill/DocPortal connector exists yet (metadata-only PowerShell recon only), Excel/.xlsx is unsupported, and circuitforge_core.tasks.dispatch_task does not currently exist in circuitforge-core — cf-orch dispatch is dead code, always falling through to local BackgroundTasks. See circuitforge-plans/pagepiper/superpowers/plans/2026-07-10-steris-licensing-pitch.md for the full writeup. --- .env.example | 16 +- Dockerfile | 5 +- README.md | 21 +- app/api/library.py | 58 ++-- app/api/{ingest.py => shelve.py} | 8 +- app/main.py | 4 +- app/services/bm25_index.py | 2 +- app/services/synthesizer.py | 4 +- app/startup.py | 10 +- config/{ingest.yaml => shelve.yaml} | 0 docs/getting-started/ollama-setup.md | 2 +- docs/getting-started/quick-start.md | 4 +- docs/index.md | 10 +- docs/reference/architecture.md | 6 +- docs/reference/tier-system.md | 2 +- docs/user-guide/library.md | 10 +- mkdocs.yml | 13 +- pyproject.toml | 2 + scripts/shelve_docx.py | 280 ++++++++++++++++++ scripts/{ingest_epub.py => shelve_epub.py} | 40 +-- scripts/shelve_odt.py | 268 +++++++++++++++++ scripts/shelve_pages.py | 100 +++++++ scripts/{ingest_pdf.py => shelve_pdf.py} | 44 +-- scripts/text_clean.py | 2 +- tests/test_library_api.py | 10 +- tests/{test_ingest.py => test_shelve.py} | 36 +-- tests/test_shelve_docx.py | 129 ++++++++ tests/test_shelve_odt.py | 135 +++++++++ tests/test_shelve_pages.py | 122 ++++++++ web/src/api.ts | 6 +- web/src/components/DocumentCard.vue | 8 +- ...{IngestProgress.vue => ShelveProgress.vue} | 4 +- web/src/views/LibraryView.vue | 10 +- 33 files changed, 1206 insertions(+), 165 deletions(-) rename app/api/{ingest.py => shelve.py} (74%) rename config/{ingest.yaml => shelve.yaml} (100%) create mode 100644 scripts/shelve_docx.py rename scripts/{ingest_epub.py => shelve_epub.py} (84%) create mode 100644 scripts/shelve_odt.py create mode 100644 scripts/shelve_pages.py rename scripts/{ingest_pdf.py => shelve_pdf.py} (77%) rename tests/{test_ingest.py => test_shelve.py} (83%) create mode 100644 tests/test_shelve_docx.py create mode 100644 tests/test_shelve_odt.py create mode 100644 tests/test_shelve_pages.py rename web/src/components/{IngestProgress.vue => ShelveProgress.vue} (96%) diff --git a/.env.example b/.env.example index 473929f..565cfc4 100644 --- a/.env.example +++ b/.env.example @@ -6,10 +6,20 @@ 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 # 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/Dockerfile b/Dockerfile index 4971de2..a3dc1ad 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,10 +2,13 @@ FROM continuumio/miniconda3:latest WORKDIR /app -# System deps for pytesseract (OCR) and pdfplumber +# System deps for pytesseract (OCR), pdfplumber, and Apple Pages conversion +# (libreoffice-writer bundles libetonyek, the only maintained open-source +# .pages parser — shelve_pages.py shells out to headless soffice) RUN apt-get update && apt-get install -y --no-install-recommends \ tesseract-ocr \ libgl1 \ + libreoffice-writer \ && rm -rf /var/lib/apt/lists/* # Install circuitforge-core from sibling directory (compose sets context: ..) diff --git a/README.md b/README.md index 85e921b..e0f0b3e 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 PDF, EPUB, DOCX, ODT, and Apple Pages 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. 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 semantic (vector) search that finds relevant passages even when your question doesn't use the exact words in the text. -- **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,13 @@ 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) | --- @@ -120,10 +123,10 @@ Default ports: Web UI `8521`, API `8540`. | Feature | Free | Paid (BYOK) | |---------|------|-------------| -| PDF and EPUB upload | Yes | Yes | +| PDF, EPUB, DOCX, ODT, and Pages 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 +144,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, EPUB/DOCX/ODT/Pages support — 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 98c396d..3349f22 100644 --- a/app/api/library.py +++ b/app/api/library.py @@ -23,32 +23,36 @@ 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", } -_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", } -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 = { @@ -61,24 +65,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) @@ -86,7 +90,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)} @@ -128,7 +132,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}") @@ -137,6 +141,8 @@ def scan_library( list(watch.glob("**/*.pdf")) + list(watch.glob("**/*.epub")) + list(watch.glob("**/*.docx")) + + list(watch.glob("**/*.odt")) + + list(watch.glob("**/*.pages")) ) queued = [] @@ -159,7 +165,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, ctx.bm25.mark_dirty ) db.execute( @@ -172,8 +178,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), @@ -183,7 +189,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, ctx.bm25.mark_dirty ) db.execute( @@ -259,8 +265,8 @@ 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") content = file.file.read() if len(content) > _MAX_UPLOAD_BYTES: @@ -289,7 +295,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, ctx.bm25.mark_dirty ) 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/main.py b/app/main.py index 5d872d3..7a82f95 100644 --- a/app/main.py +++ b/app/main.py @@ -49,14 +49,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..b982ba7 100644 --- a/app/startup.py +++ b/app/startup.py @@ -63,11 +63,15 @@ 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 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/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..889150d 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 PDF / EPUB / DOCX / ODT / Pages** 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) — PDF/EPUB/DOCX/ODT/Pages 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..b1aed29 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, and Apple Pages 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 | +| PDF, EPUB, DOCX, ODT, and Pages 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..f39ad8e 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 +PDF / EPUB / DOCX / ODT / Pages 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/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..e8af94d 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 PDF / EPUB / DOCX / ODT / Pages** 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 PDF, EPUB, DOCX, ODT, or Pages 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..ae077e1 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 PDF, EPUB, DOCX, ODT, and Apple Pages library 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 884a2da..ce6a2de 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,8 @@ dependencies = [ "PyYAML>=6.0", "httpx>=0.27", "circuitforge-core[pdf,vector]>=0.19.0", + "python-docx>=1.0", + "odfpy>=1.4", ] [tool.setuptools.packages.find] 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_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/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_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_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/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 @@