Compare commits

..

10 commits

Author SHA1 Message Date
059ad21ed3 docs: add product docs and screenshots 2026-07-11 22:40:42 -07:00
afdc211813 chore: bump version to 0.2.0
Multi-format shelve support (DOCX/ODT/Pages/XLSX/ODS/Numbers), the
ingest→shelve terminology rename, GPU_SERVER_URL config rename, and the
Agent-ModernColBERT retriever swap — see release notes.
2026-07-10 19:38:24 -07:00
b0df7ac7bc Merge pull request 'feat: replace nomic-embed-text retriever with Agent-ModernColBERT' (#15) from feat/colbert-retriever into main 2026-07-10 19:37:10 -07:00
f548c82c96 Merge branch 'main' into feat/colbert-retriever 2026-07-10 19:36:11 -07:00
c684f347e9 Merge pull request 'feat: rename CF_ORCH_URL to GPU_SERVER_URL for self-hoster clarity' (#14) from fix/gpu-server-url-rename into main 2026-07-10 19:34:19 -07:00
156f52d6ac Merge pull request 'feat: shelve rename + DOCX/ODT/Pages/XLSX/ODS/Numbers support' (#12) from feat/shelve-multi-format-support into main 2026-07-10 19:33:59 -07:00
89a58ec9b0 feat: replace nomic-embed-text retriever with Agent-ModernColBERT
Bi-encoder embeddings collapse a whole query into one vector, losing
multi-part reasoning structure — queries like "the procedure for setting
an IP on an AVC-X" or "what is the action economy for a fighter casting
a spell while prone" lose nuance. Agent-ModernColBERT is a late-interaction
retriever: per-token embeddings, scored via MaxSim at query time, built
specifically for agentic/multi-hop queries.

Implements Option A from the issue (in-process, via `pylate`) rather than
Option B (managed cf-orch service) — cf-orch already has `agent-moderncolbert`
registered in model_registry.yaml with a `pagepiper/retrieve` assignment
in assignments.yaml pointing at it and referencing this issue directly,
someone had already pre-wired that side.

- app/services/colbert_index.py: new ColBERTIndex class, mirrors
  BM25Index's dirty-flag/rebuild-from-SQLite pattern exactly — no
  separate per-shelve indexing step needed, just mark_dirty() on the
  same callback that already marks BM25 dirty.
- app/services/retriever.py: hybrid_search's semantic half now merges
  BM25 with ColBERT MaxSim scores (min-max normalized per-batch, since
  MaxSim is unbounded unlike the old sqlite-vec L2-distance path) instead
  of Ollama-embed + sqlite-vec cosine. BM25 merge/rank/per-doc-cap/
  adjacent-chunk-window logic is unchanged.
- app/main.py / app/deps.py: per-user ColBERTIndex registry, same
  pattern as the existing per-user BM25Index registry.
- Existing BYOK tier gate preserved exactly (llm is None check) — this
  is a retrieval-technology swap, not a tier/licensing change. The
  ColBERT model runs locally via pylate with no Ollama dependency, but
  gating still follows product tiering.
- 12 new tests. pylate is intentionally NOT installed in the dev/test
  env — see the cf-sysadmin skill's "Known Gotchas" for why (installing
  it directly into the shared `cf` conda env broke several other
  services' torch/transformers pins on 2026-07-10). Tests inject fake
  pylate modules via sys.modules instead.

Known follow-up (not addressed here): shelve scripts still compute and
store Ollama embeddings into `page_vecs` at shelve time — that table is
no longer read by search/chat now that retrieval uses the ColBERT index.
Removing the now-redundant embedding step is separate cleanup.

Closes: #8
2026-07-10 19:02:12 -07:00
d39cfbd87a feat: add XLSX, ODS, and Apple Numbers spreadsheet support
Extends the shelve pipeline to cover spreadsheets, closing the Excel gap
called out in the PR's original "known gaps" list — Windchill/DocPortal
corpora commonly include parts lists and spec sheets as spreadsheets, not
just prose documents.

- scripts/shelve_xlsx.py — openpyxl, chunked by sheet with row-window
  splitting for large sheets (header row repeated in every window so
  each chunk stays self-describing for retrieval).
- scripts/shelve_ods.py — same chunking strategy via odfpy (already a
  dependency from ODT support), OpenDocumentSpreadsheet's Table/TableRow/
  TableCell.
- scripts/shelve_numbers.py — converts via headless LibreOffice to XLSX
  and delegates to shelve_xlsx, mirroring shelve_pages.py's pattern for
  .pages. Adds libreoffice-calc to the Docker image alongside the
  existing libreoffice-writer.
- Upload button text changed from an ever-growing format list to
  "Upload Document or Spreadsheet" — the Supported Formats table in
  README/docs is now the source of truth for the full list.
- 13 new tests (XLSX, ODS, Numbers); full suite (85 tests) passing.

Manually verified via Playwright against an isolated test instance:
XLSX and ODS both upload, shelve to "ready", and store correctly
row-serialized, header-repeated chunks (confirmed via sample-chunks).
BM25 search against a 2-chunk toy corpus returned no hits for terms
split 1-vs-1 across the two chunks — traced to Okapi BM25's IDF formula
giving an exact 0 for terms in exactly half a tiny corpus
(log((N-n+0.5)/(n+0.5)) = log(1.0) = 0, filtered by `score <= 0`), not a
defect in the new shelvers. The earlier DOCX/ODT/PDF Playwright pass
(5 chunks total) diluted this enough to return real results.
2026-07-10 15:06:16 -07:00
fccb30c79d chore: gitignore Playwright MCP scratch directory
Discovered while manually verifying the DOCX/ODT/Pages upload flow —
.playwright-mcp/ (fixtures, screenshots) wasn't ignored and risked
accidental commits of test artifacts.
2026-07-10 14:18:52 -07:00
f941ebdeeb feat: add ODT and Apple Pages document support, wire DOCX into UI
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.
2026-07-10 13:58:43 -07:00
51 changed files with 2642 additions and 218 deletions

View file

@ -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
# 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

3
.gitignore vendored
View file

@ -29,3 +29,6 @@ compose.override.yml
# Logs and runtime files
*.log
*.db
# Playwright MCP scratch dir (test fixtures, screenshots — never commit)
.playwright-mcp/

View file

@ -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: ..)

View file

@ -4,9 +4,9 @@
[![Status](https://img.shields.io/badge/status-beta-blue)](https://git.opensourcesolarpunk.com/Circuit-Forge/pagepiper)
[![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)
[![Version](https://img.shields.io/badge/version-v0.2.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
@ -31,8 +31,8 @@ No cloud required. Your files stay on your machine.
- **Your library, not ours.** Documents are indexed and stored locally. Nothing is sent to a third-party service unless you explicitly configure a cloud LLM.
- **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.
- **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 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) |
---
@ -92,7 +98,7 @@ PAGEPIPER_EMBED_MODEL=nomic-embed-text
|-------|-----------|
| Backend API | FastAPI + SQLite |
| Full-text search | BM25 (custom index, no external service) |
| Vector search | sqlite-vec + Ollama embeddings (optional) |
| Semantic search | Agent-ModernColBERT late-interaction retrieval, via `pylate` (optional, BYOK-gated) |
| LLM synthesis | Ollama (local, any model) |
| Frontend | Vue 3 SPA served by nginx |
| Deployment | Docker Compose |
@ -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)

View file

@ -80,7 +80,7 @@ def _build_llm_for_alloc(alloc) -> "LLMRouter":
def _run_chat(req: "ChatRequest", ctx: "UserCtx", llm) -> "ChatResponse":
retriever = Retriever(ctx.bm25)
retriever = Retriever(ctx.bm25, ctx.colbert)
chunks = retriever.hybrid_search(
query=req.message,
top_k=req.top_k,

View file

@ -23,32 +23,48 @@ 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",
}
def _dispatch_ingest(
def _mark_indexes_dirty(ctx: UserCtx) -> None:
"""Mark both BM25 and ColBERT indexes dirty — call after any document is shelved."""
ctx.bm25.mark_dirty()
ctx.colbert.mark_dirty()
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 +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)
@ -86,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)}
@ -128,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}")
@ -137,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 = []
@ -159,8 +180,8 @@ def scan_library(
).fetchone()[0]
db.commit()
task_id = _dispatch_ingest(
doc_id, path_str, background_tasks, ctx.data_dir, ctx.bm25.mark_dirty
task_id = _dispatch_shelve(
doc_id, path_str, background_tasks, ctx.data_dir, lambda: _mark_indexes_dirty(ctx)
)
db.execute(
"UPDATE documents SET status='processing', task_id=? WHERE id=?",
@ -172,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),
@ -183,8 +204,8 @@ def reingest_document(
if not row:
raise HTTPException(status_code=404, detail="Document not found")
task_id = _dispatch_ingest(
doc_id, row["file_path"], background_tasks, ctx.data_dir, ctx.bm25.mark_dirty
task_id = _dispatch_shelve(
doc_id, row["file_path"], background_tasks, ctx.data_dir, lambda: _mark_indexes_dirty(ctx)
)
db.execute(
"UPDATE documents SET status='processing', task_id=?, error_msg=NULL WHERE id=?",
@ -216,7 +237,7 @@ def delete_document(
except Exception as exc:
logger.warning("Could not remove vectors for doc %s: %s", doc_id, exc)
ctx.bm25.mark_dirty()
_mark_indexes_dirty(ctx)
def _get_vec_count(doc_id: str, vec_db_path: str) -> int:
@ -259,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:
@ -289,8 +313,8 @@ def upload_document(
).fetchone()[0]
db.commit()
task_id = _dispatch_ingest(
doc_id, path_str, background_tasks, ctx.data_dir, ctx.bm25.mark_dirty
task_id = _dispatch_shelve(
doc_id, path_str, background_tasks, ctx.data_dir, lambda: _mark_indexes_dirty(ctx)
)
db.execute(
"UPDATE documents SET status='processing', task_id=? WHERE id=?",

View file

@ -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] = {}

View file

@ -11,11 +11,12 @@ from fastapi import Depends, Request
from app.config import DATA_DIR, LOCAL_USER_ID
from app.services.bm25_index import BM25Index
from app.services.colbert_index import ColBERTIndex
@dataclass
class UserCtx:
"""Per-request context routing DB paths and BM25 to the right user."""
"""Per-request context routing DB paths, BM25, and ColBERT to the right user."""
user_id: str
db_path: str
@ -23,6 +24,7 @@ class UserCtx:
data_dir: Path
watch_dir: Path
bm25: BM25Index
colbert: ColBERTIndex
_user_startup_done: set[str] = set()
@ -67,6 +69,7 @@ def get_user_ctx(request: Request) -> UserCtx:
data_dir=user_dir,
watch_dir=watch_dir,
bm25=_main._get_bm25_for(user_id),
colbert=_main._get_colbert_for(user_id, user_dir),
)

View file

@ -10,12 +10,17 @@ from fastapi import FastAPI
from app.config import DB_PATH, VEC_DB_PATH, VEC_DIMENSIONS
from app.services.bm25_index import BM25Index
from app.services.colbert_index import ColBERTIndex
logger = logging.getLogger("pagepiper")
# Per-user BM25 registry — keyed by user_id; "__local__" for single-user mode
_bm25_map: dict[str, BM25Index] = {}
# Per-user ColBERT registry — keyed by user_id; index files live under
# <user_dir>/colbert_index/ (see _get_colbert_for)
_colbert_map: dict[str, ColBERTIndex] = {}
def _get_bm25_for(user_id: str) -> BM25Index:
if user_id not in _bm25_map:
@ -23,6 +28,13 @@ def _get_bm25_for(user_id: str) -> BM25Index:
return _bm25_map[user_id]
def _get_colbert_for(user_id: str, user_dir) -> ColBERTIndex:
if user_id not in _colbert_map:
index_dir = str(user_dir / "colbert_index")
_colbert_map[user_id] = ColBERTIndex(index_dir=index_dir)
return _colbert_map[user_id]
@asynccontextmanager
async def lifespan(app: FastAPI):
from app.cloud_session import CLOUD_MODE
@ -38,9 +50,12 @@ async def lifespan(app: FastAPI):
warn_if_unencrypted(str(DATA_DIR))
else:
# In cloud mode, per-user migration and vec schema check run on first request (deps.py).
from app.config import DATA_DIR
apply_migrations(DB_PATH)
check_and_rebuild_vec_schema(VEC_DB_PATH, VEC_DIMENSIONS, DB_PATH)
_get_bm25_for(LOCAL_USER_ID).mark_dirty()
_get_colbert_for(LOCAL_USER_ID, DATA_DIR).mark_dirty()
yield
@ -49,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")

View file

@ -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:

View file

@ -0,0 +1,137 @@
# app/services/colbert_index.py
"""
ColBERT late-interaction index (pagepiper#8).
Replaces nomic-embed-text bi-encoder + cosine similarity for the semantic
half of hybrid search with Agent-ModernColBERT (lightonai/Agent-ModernColBERT),
a late-interaction retriever that keeps per-token embeddings and scores via
MaxSim at query time better suited to multi-part rulebook questions than a
single collapsed query vector.
BSL 1.1 same BYOK gate as the rest of hybrid search (Retriever only reaches
this index when an LLM is configured). The model itself runs locally via
`pylate`, no Ollama call required the gate matches product tiering, not a
technical dependency on Ollama.
Mirrors BM25Index's dirty-flag, rebuild-from-SQLite pattern: no separate
per-shelve indexing step is needed. `mark_dirty()` is called by the same
callback that already marks BM25 dirty on shelve completion; the next query
triggers a full rebuild from `page_chunks`.
"""
from __future__ import annotations
import logging
import os
import sqlite3
import threading
logger = logging.getLogger(__name__)
_DEFAULT_MODEL = "lightonai/Agent-ModernColBERT"
class ColBERTIndex:
def __init__(self, index_dir: str, model_name: str | None = None) -> None:
self._index_dir = index_dir
self._model_name = model_name or os.environ.get("PAGEPIPER_COLBERT_MODEL", _DEFAULT_MODEL)
self._model = None
self._index = None
self._chunks: dict[str, dict] = {}
self._dirty = True
self._lock = threading.Lock()
def mark_dirty(self) -> None:
"""Signal that the index needs rebuilding (call after any document is shelved)."""
self._dirty = True
def _get_model(self):
if self._model is None:
from pylate import models
logger.info("Loading ColBERT model %s", self._model_name)
self._model = models.ColBERT(model_name_or_path=self._model_name)
return self._model
def ensure_fresh(self, db_path: str) -> None:
"""Rebuild from SQLite if dirty."""
if not self._dirty:
return
with self._lock:
if not self._dirty:
return
try:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
try:
rows = conn.execute(
"SELECT id, doc_id, page_number, text FROM page_chunks ORDER BY doc_id, page_number"
).fetchall()
finally:
conn.close()
except sqlite3.Error as exc:
logger.error("ColBERT index rebuild failed: %s", exc)
return
self._chunks = {str(r["id"]): dict(r) for r in rows}
if not rows:
self._index = None
self._dirty = False
return
from pylate import indexes
model = self._get_model()
ids = [str(r["id"]) for r in rows]
texts = [r["text"] for r in rows]
embeddings = model.encode(texts, is_query=False, show_progress_bar=False)
os.makedirs(self._index_dir, exist_ok=True)
index = indexes.Voyager(
index_folder=self._index_dir, index_name="colbert", override=True
)
index.add_documents(documents_ids=ids, documents_embeddings=embeddings)
self._index = index
self._dirty = False
logger.info("ColBERT index rebuilt: %d chunks", len(rows))
def query(
self,
query_text: str,
top_k: int = 10,
doc_ids: list[str] | None = None,
) -> list[dict]:
"""Search the corpus. Returns results sorted by descending MaxSim score."""
if self._index is None:
return []
from pylate import retrieve
model = self._get_model()
query_embeddings = model.encode([query_text], is_query=True, show_progress_bar=False)
retriever = retrieve.ColBERT(index=self._index)
# Oversample when filtering to a doc subset — same pattern as the
# sqlite-vec path this replaces (see app/services/retriever.py).
k = top_k * 20 if doc_ids else top_k * 2
results = retriever.retrieve(queries_embeddings=query_embeddings, k=k)[0]
hits: list[dict] = []
for r in results:
chunk = self._chunks.get(str(r["id"]))
if not chunk:
continue
if doc_ids is not None and chunk["doc_id"] not in doc_ids:
continue
hits.append(
{
"chunk_id": chunk["id"],
"doc_id": chunk["doc_id"],
"page_number": chunk["page_number"],
"text": chunk["text"],
"score": r["score"],
}
)
if len(hits) >= top_k:
break
return hits

View file

@ -4,6 +4,11 @@ Hybrid BM25 + semantic retriever.
BSL 1.1 semantic path requires PAGEPIPER_OLLAMA_URL (BYOK gate).
BM25-only path is MIT and has no gate.
The semantic half uses Agent-ModernColBERT (pagepiper#8) — a late-interaction
retriever scored via MaxSim, replacing the earlier nomic-embed-text bi-encoder
+ cosine similarity approach. The model runs locally via `pylate`; the BYOK
gate below matches product tiering, not a technical dependency on Ollama.
"""
from __future__ import annotations
@ -12,6 +17,7 @@ import sqlite3
from dataclasses import dataclass
from app.services.bm25_index import BM25Index
from app.services.colbert_index import ColBERTIndex
logger = logging.getLogger(__name__)
@ -85,8 +91,9 @@ class RetrievedChunk:
class Retriever:
def __init__(self, bm25: BM25Index) -> None:
def __init__(self, bm25: BM25Index, colbert: ColBERTIndex | None = None) -> None:
self._bm25 = bm25
self._colbert = colbert
def hybrid_search(
self,
@ -98,14 +105,12 @@ class Retriever:
llm, # LLMRouter | None — caller must pass
) -> list[RetrievedChunk]:
"""
Merge BM25 and semantic results.
Falls back to BM25-only if llm is None.
Merge BM25 and semantic (ColBERT) results.
Falls back to BM25-only if llm is None or no ColBERT index is configured.
"""
if llm is None:
if llm is None or self._colbert is None:
return self._bm25_only(query, top_k, doc_ids, db_path)
from circuitforge_core.vector.sqlite_vec import LocalSQLiteVecStore
self._bm25.ensure_fresh(db_path)
bm25_hits = {
r.chunk_id: r
@ -113,24 +118,15 @@ class Retriever:
}
try:
vec = llm.embed([query])[0]
self._colbert.ensure_fresh(db_path)
# ColBERTIndex.query already oversamples internally when doc_ids is set —
# see app/services/colbert_index.py.
colbert_hits = self._colbert.query(query, top_k=top_k, doc_ids=doc_ids)
except Exception as exc:
logger.warning("Embed failed, falling back to BM25-only: %s", exc)
logger.warning("ColBERT retrieval failed, falling back to BM25-only: %s", exc)
return self._bm25_only(query, top_k, doc_ids, db_path)
from app.config import VEC_DIMENSIONS
store = LocalSQLiteVecStore(db_path=vec_db_path, table="page_vecs", dimensions=VEC_DIMENSIONS)
# sqlite-vec applies filter_metadata as a Python post-filter after fetching k
# nearest globally. When the corpus spans many documents and only a subset is
# selected, most of those k candidates are from non-target docs and get dropped,
# leaving too few vector hits. Oversample heavily and filter in Python instead.
if doc_ids:
vec_candidates = store.query(vec, top_k=top_k * 20)
vec_hits = [h for h in vec_candidates if h.metadata.get("doc_id") in doc_ids]
else:
vec_hits = store.query(vec, top_k=top_k * 2)
# Merge: BM25 hits take priority; vector hits fill in additional results
# Merge: BM25 hits take priority; ColBERT hits fill in additional results
merged: dict[str, RetrievedChunk] = {}
for cid, r in bm25_hits.items():
merged[cid] = RetrievedChunk(
@ -141,33 +137,37 @@ class Retriever:
bm25_score=r.score,
vector_score=None,
)
for vh in vec_hits:
# _chunks is the loaded list of dicts from BM25Index; no public accessor exists
text = next((c["text"] for c in self._bm25._chunks if c["id"] == vh.entry_id), "")
if vh.entry_id in merged:
existing = merged[vh.entry_id]
merged[vh.entry_id] = RetrievedChunk(
# ColBERT MaxSim scores are unbounded (roughly num_query_tokens *
# max_per_token_similarity), unlike BM25's already-comparable range —
# min-max normalize within this result batch before combining.
max_colbert_score = max((h["score"] for h in colbert_hits), default=0.0)
for h in colbert_hits:
cid = h["chunk_id"]
norm_score = (h["score"] / max_colbert_score) if max_colbert_score > 0 else 0.0
if cid in merged:
existing = merged[cid]
merged[cid] = RetrievedChunk(
chunk_id=existing.chunk_id,
doc_id=existing.doc_id,
page_number=existing.page_number,
text=existing.text,
bm25_score=existing.bm25_score,
vector_score=vh.score,
vector_score=norm_score,
)
else:
merged[vh.entry_id] = RetrievedChunk(
chunk_id=vh.entry_id,
doc_id=vh.metadata.get("doc_id", ""),
page_number=int(vh.metadata.get("page_number", 0)),
text=text,
merged[cid] = RetrievedChunk(
chunk_id=cid,
doc_id=h["doc_id"],
page_number=h["page_number"],
text=h["text"],
bm25_score=0.0,
vector_score=vh.score,
vector_score=norm_score,
)
def _combined(r: RetrievedChunk) -> float:
bm25 = r.bm25_score
# sqlite-vec returns L2 distance (lower=better); invert to [0,1] higher-is-better
vec = (1.0 / (1.0 + r.vector_score)) if r.vector_score is not None else 0.0
vec = r.vector_score if r.vector_score is not None else 0.0
return bm25 * 0.5 + vec * 0.5
all_ranked = sorted(merged.values(), key=_combined, reverse=True)

View file

@ -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

View file

@ -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:

View file

@ -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

View file

@ -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.

View file

@ -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
@ -10,13 +10,15 @@ Try it: [pagepiper.circuitforge.tech](https://pagepiper.circuitforge.tech)
### Library
![Library view](screenshots/01-library.png)
![Library view](screenshots/01-library.png){ .only-light }
![Library view](screenshots/01-library-dark.png){ .only-dark }
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
![Chat view](screenshots/02-chat.png)
![Chat view](screenshots/02-chat.png){ .only-light }
![Chat view](screenshots/02-chat-dark.png){ .only-dark }
Ask questions across your indexed documents. Results cite the source document and page number.
@ -25,8 +27,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 +139,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.

View file

@ -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) |

View file

@ -14,10 +14,18 @@ Copy `.env.example` to `.env` and configure as needed.
| Variable | Default | Description |
|----------|---------|-------------|
| `PAGEPIPER_OLLAMA_URL` | _(unset)_ | Ollama base URL, e.g. `http://localhost:11434`. Enables hybrid search and chat. |
| `PAGEPIPER_EMBED_MODEL` | `nomic-embed-text` | Ollama embedding model |
| `PAGEPIPER_EMBED_DIMS` | `1024` | Embedding dimensions (must match the model) |
| `PAGEPIPER_OLLAMA_URL` | _(unset)_ | Ollama base URL, e.g. `http://localhost:11434`. Enables hybrid search and chat (BYOK gate — see below). |
| `PAGEPIPER_CHAT_MODEL` | `mistral:7b` | Ollama chat/completion model |
| `PAGEPIPER_EMBED_MODEL` | `nomic-embed-text` | Ollama embedding model — used for shelve-time embeddings only (`page_vecs`), not for search retrieval (see ColBERT below) |
| `PAGEPIPER_EMBED_DIMS` | `1024` | Embedding dimensions (must match `PAGEPIPER_EMBED_MODEL`) |
## Semantic search (ColBERT)
| Variable | Default | Description |
|----------|---------|-------------|
| `PAGEPIPER_COLBERT_MODEL` | `lightonai/Agent-ModernColBERT` | HuggingFace model used for hybrid search's semantic half — a late-interaction retriever, runs locally via `pylate`, no Ollama call required. Gated behind the same BYOK check as the rest of hybrid search (`PAGEPIPER_OLLAMA_URL` or `CF_ORCH_URL`/`GPU_SERVER_URL` must be set). |
**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.
## GPU server / cf-orch (managed deployments)

View file

@ -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) |

Binary file not shown.

After

Width:  |  Height:  |  Size: 28 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 27 KiB

View file

@ -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

View file

@ -21,4 +21,4 @@ BM25 (Best Match 25) ranks pages by term frequency weighted against how rare eac
## Hybrid search (requires Ollama)
When Ollama is configured, the Chat endpoint uses hybrid search behind the scenes: BM25 results are merged with semantic vector results using a 50/50 score blend. The Search page always uses BM25 only.
When Ollama is configured, the Chat endpoint uses hybrid search behind the scenes: BM25 results are merged with Agent-ModernColBERT late-interaction results using a 50/50 score blend. The Search page always uses BM25 only.

View file

@ -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
@ -31,6 +31,7 @@ theme:
markdown_extensions:
- admonition
- attr_list
- pymdownx.details
- pymdownx.superfences:
custom_fences:
@ -59,6 +60,10 @@ nav:
- Architecture: reference/architecture.md
- Tier System: reference/tier-system.md
- Environment Variables: reference/environment-variables.md
- All CF Docs: https://docs.circuitforge.tech
extra_css:
- stylesheets/theme.css
extra_javascript:
- plausible.js

View file

@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
[project]
name = "pagepiper"
version = "0.1.0"
description = "Self-hosted PDF library manager with RAG chat and page-level citations"
version = "0.2.0"
description = "Self-hosted document and spreadsheet library manager with RAG chat and page-level citations"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
@ -16,6 +16,10 @@ 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",
]
[tool.setuptools.packages.find]

280
scripts/shelve_docx.py Normal file
View file

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

View file

@ -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)

101
scripts/shelve_numbers.py Normal file
View file

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

211
scripts/shelve_ods.py Normal file
View file

@ -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
<table:table> 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,
)

268
scripts/shelve_odt.py Normal file
View file

@ -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 (<text:h>): 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,
)

100
scripts/shelve_pages.py Normal file
View file

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

View file

@ -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)

208
scripts/shelve_xlsx.py Normal file
View file

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

View file

@ -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.

View file

@ -31,6 +31,7 @@ def client(test_db, tmp_path, monkeypatch):
from app.deps import UserCtx, get_db, get_user_ctx
from app.main import app
from app.services.bm25_index import BM25Index
from app.services.colbert_index import ColBERTIndex
from app.startup import apply_migrations, check_and_rebuild_vec_schema
monkeypatch.setattr(_main_module, "_apply_migrations", lambda: None, raising=False)
@ -43,6 +44,7 @@ def client(test_db, tmp_path, monkeypatch):
test_bm25 = BM25Index()
test_bm25.mark_dirty()
test_colbert = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
def override_user_ctx():
return UserCtx(
@ -52,6 +54,7 @@ def client(test_db, tmp_path, monkeypatch):
data_dir=Path(tmp_path),
watch_dir=Path(tmp_path) / "books",
bm25=test_bm25,
colbert=test_colbert,
)
def override_db():

168
tests/test_colbert_index.py Normal file
View file

@ -0,0 +1,168 @@
# tests/test_colbert_index.py
"""Tests for app.services.colbert_index.
pylate is NOT installed in the dev/test env by design (see cf-sysadmin skill's
"Known Gotchas" installing it directly into the shared `cf` conda env broke
several other services' pinned torch/transformers versions on 2026-07-10).
These tests inject fake `pylate`/`pylate.models`/`pylate.indexes`/`pylate.retrieve`
modules via sys.modules so ColBERTIndex's lazy imports resolve to mocks without
pylate ever needing to be installed here.
"""
from __future__ import annotations
import sqlite3
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from app.services.colbert_index import ColBERTIndex
@pytest.fixture
def fake_pylate(monkeypatch):
fake_models_mod = types.ModuleType("pylate.models")
fake_indexes_mod = types.ModuleType("pylate.indexes")
fake_retrieve_mod = types.ModuleType("pylate.retrieve")
fake_pylate_mod = types.ModuleType("pylate")
fake_pylate_mod.models = fake_models_mod
fake_pylate_mod.indexes = fake_indexes_mod
fake_pylate_mod.retrieve = fake_retrieve_mod
mock_model = MagicMock()
mock_model.encode.return_value = [[0.1, 0.2], [0.3, 0.4]]
fake_models_mod.ColBERT = MagicMock(return_value=mock_model)
mock_index = MagicMock()
fake_indexes_mod.Voyager = MagicMock(return_value=mock_index)
mock_retriever = MagicMock()
fake_retrieve_mod.ColBERT = MagicMock(return_value=mock_retriever)
monkeypatch.setitem(sys.modules, "pylate", fake_pylate_mod)
monkeypatch.setitem(sys.modules, "pylate.models", fake_models_mod)
monkeypatch.setitem(sys.modules, "pylate.indexes", fake_indexes_mod)
monkeypatch.setitem(sys.modules, "pylate.retrieve", fake_retrieve_mod)
return types.SimpleNamespace(
model_cls=fake_models_mod.ColBERT,
model=mock_model,
index_cls=fake_indexes_mod.Voyager,
index=mock_index,
retriever_cls=fake_retrieve_mod.ColBERT,
retriever=mock_retriever,
)
@pytest.fixture
def seeded_db(tmp_path) -> 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.pdf','ready')"
)
conn.execute(
"INSERT INTO page_chunks(id, doc_id, page_number, text, source, word_count) "
"VALUES ('c1','d1',1,'Setting the IP on the AVC-X','text',6)"
)
conn.execute(
"INSERT INTO page_chunks(id, doc_id, page_number, text, source, word_count) "
"VALUES ('c2','d1',2,'Filter cartridge replacement steps','text',5)"
)
conn.commit()
conn.close()
return db_path
def test_ensure_fresh_builds_index_from_sqlite(fake_pylate, seeded_db, tmp_path):
idx = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
idx.ensure_fresh(seeded_db)
fake_pylate.model.encode.assert_called_once()
call_args = fake_pylate.model.encode.call_args
assert set(call_args[0][0]) == {"Setting the IP on the AVC-X", "Filter cartridge replacement steps"}
assert call_args[1]["is_query"] is False
fake_pylate.index.add_documents.assert_called_once()
add_kwargs = fake_pylate.index.add_documents.call_args[1]
assert set(add_kwargs["documents_ids"]) == {"c1", "c2"}
def test_ensure_fresh_skips_rebuild_when_not_dirty(fake_pylate, seeded_db, tmp_path):
idx = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
idx.ensure_fresh(seeded_db)
idx.ensure_fresh(seeded_db)
fake_pylate.model.encode.assert_called_once()
def test_mark_dirty_forces_rebuild(fake_pylate, seeded_db, tmp_path):
idx = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
idx.ensure_fresh(seeded_db)
idx.mark_dirty()
idx.ensure_fresh(seeded_db)
assert fake_pylate.model.encode.call_count == 2
def test_ensure_fresh_with_empty_corpus_leaves_index_none(fake_pylate, tmp_path):
db_path = str(tmp_path / "empty.db")
schema = Path("migrations/001_initial_schema.sql").read_text()
conn = sqlite3.connect(db_path)
conn.executescript(schema)
conn.commit()
conn.close()
idx = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
idx.ensure_fresh(db_path)
fake_pylate.index_cls.assert_not_called()
assert idx.query("anything") == []
def test_query_maps_results_back_to_chunks(fake_pylate, seeded_db, tmp_path):
idx = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
idx.ensure_fresh(seeded_db)
fake_pylate.retriever.retrieve.return_value = [
[{"id": "c1", "score": 13.5}, {"id": "c2", "score": 9.2}]
]
results = idx.query("how do I set the IP on the AVC-X", top_k=10)
assert len(results) == 2
assert results[0]["chunk_id"] == "c1"
assert results[0]["doc_id"] == "d1"
assert results[0]["score"] == 13.5
assert results[1]["chunk_id"] == "c2"
def test_query_filters_by_doc_ids(fake_pylate, seeded_db, tmp_path):
idx = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
idx.ensure_fresh(seeded_db)
fake_pylate.retriever.retrieve.return_value = [
[{"id": "c1", "score": 13.5}, {"id": "c2", "score": 9.2}]
]
results = idx.query("query", top_k=10, doc_ids=["other-doc"])
assert results == []
def test_query_respects_top_k(fake_pylate, seeded_db, tmp_path):
idx = ColBERTIndex(index_dir=str(tmp_path / "colbert_index"))
idx.ensure_fresh(seeded_db)
fake_pylate.retriever.retrieve.return_value = [
[{"id": "c1", "score": 13.5}, {"id": "c2", "score": 9.2}]
]
results = idx.query("query", top_k=1)
assert len(results) == 1
assert results[0]["chunk_id"] == "c1"

View file

@ -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")

113
tests/test_retriever.py Normal file
View file

@ -0,0 +1,113 @@
# tests/test_retriever.py
"""Tests for app.services.retriever.Retriever.hybrid_search — the BM25 + ColBERT merge."""
from __future__ import annotations
import sqlite3
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from app.services.bm25_index import BM25Index
from app.services.retriever import Retriever
@pytest.fixture
def seeded_db(tmp_path) -> 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.pdf','ready')"
)
conn.execute(
"INSERT INTO page_chunks(id, doc_id, page_number, text, source, word_count) "
"VALUES ('c1','d1',1,'Setting the IP address on the AVC-X device','text',7)"
)
conn.execute(
"INSERT INTO page_chunks(id, doc_id, page_number, text, source, word_count) "
"VALUES ('c2','d1',2,'Filter cartridge replacement procedure','text',4)"
)
# Third, unrelated chunk — with only 2 chunks total, a term appearing in
# exactly one of them gets an Okapi BM25 IDF of exactly log(1.0) == 0
# (log((N-n+0.5)/(n+0.5)) with N=2, n=1), silently zeroing every score.
# A third chunk dilutes N enough for real term-overlap scores to surface.
conn.execute(
"INSERT INTO page_chunks(id, doc_id, page_number, text, source, word_count) "
"VALUES ('c3','d1',3,'Warranty terms and annual maintenance schedule','text',5)"
)
conn.commit()
conn.close()
return db_path
def _seeded_bm25() -> BM25Index:
idx = BM25Index()
idx._dirty = True
return idx
def test_hybrid_search_falls_back_to_bm25_only_without_llm(seeded_db):
retriever = Retriever(_seeded_bm25(), colbert=MagicMock())
results = retriever.hybrid_search(
query="IP address", top_k=5, doc_ids=None,
db_path=seeded_db, vec_db_path="unused", llm=None,
)
assert any(r.chunk_id == "c1" for r in results)
def test_hybrid_search_falls_back_to_bm25_only_without_colbert(seeded_db):
retriever = Retriever(_seeded_bm25(), colbert=None)
results = retriever.hybrid_search(
query="IP address", top_k=5, doc_ids=None,
db_path=seeded_db, vec_db_path="unused", llm=MagicMock(),
)
assert any(r.chunk_id == "c1" for r in results)
def test_hybrid_search_merges_bm25_and_colbert_hits(seeded_db):
fake_colbert = MagicMock()
fake_colbert.query.return_value = [
{"chunk_id": "c1", "doc_id": "d1", "page_number": 1, "text": "Setting the IP address on the AVC-X device", "score": 15.0},
{"chunk_id": "c2", "doc_id": "d1", "page_number": 2, "text": "Filter cartridge replacement procedure", "score": 5.0},
]
retriever = Retriever(_seeded_bm25(), colbert=fake_colbert)
results = retriever.hybrid_search(
query="IP address AVC-X", top_k=5, doc_ids=None,
db_path=seeded_db, vec_db_path="unused", llm=MagicMock(),
)
fake_colbert.ensure_fresh.assert_called_once_with(seeded_db)
result_ids = {r.chunk_id for r in results}
assert "c1" in result_ids
c1 = next(r for r in results if r.chunk_id == "c1")
assert c1.vector_score == 1.0 # highest colbert score, normalized to max
def test_hybrid_search_falls_back_when_colbert_raises(seeded_db):
fake_colbert = MagicMock()
fake_colbert.query.side_effect = RuntimeError("model not loaded")
retriever = Retriever(_seeded_bm25(), colbert=fake_colbert)
results = retriever.hybrid_search(
query="IP address", top_k=5, doc_ids=None,
db_path=seeded_db, vec_db_path="unused", llm=MagicMock(),
)
assert any(r.chunk_id == "c1" for r in results)
def test_hybrid_search_discards_pure_noise(seeded_db):
fake_colbert = MagicMock()
fake_colbert.query.return_value = []
retriever = Retriever(_seeded_bm25(), colbert=fake_colbert)
results = retriever.hybrid_search(
query="completely unrelated gibberish xyzzy",
top_k=5, doc_ids=None,
db_path=seeded_db, vec_db_path="unused", llm=MagicMock(),
)
assert results == []

View file

@ -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)

129
tests/test_shelve_docx.py Normal file
View file

@ -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"

View file

@ -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"

135
tests/test_shelve_ods.py Normal file
View file

@ -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"

135
tests/test_shelve_odt.py Normal file
View file

@ -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"

122
tests/test_shelve_pages.py Normal file
View file

@ -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"

125
tests/test_shelve_xlsx.py Normal file
View file

@ -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"

View file

@ -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<TaskStatus> {
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()
},

View file

@ -5,7 +5,7 @@
<div class="doc-meta" v-if="displayPageCount != null">{{ displayPageCount }} pages</div>
<div class="doc-meta path">{{ shortPath }}</div>
<div class="ingest-progress" v-if="isProcessing">
<div class="shelve-progress" v-if="isProcessing">
<div class="progress-label">
<span>{{ progressLabel }}</span>
<span class="progress-pct" v-if="progressPct != null">{{ progressPct }}%</span>
@ -18,7 +18,7 @@
<p class="doc-error" v-if="currentStatus === 'error'">{{ errorMsg ?? 'Indexing failed.' }}</p>
<div class="doc-actions">
<button class="btn-sm" @click="emit('reingest', doc.id)" :disabled="isProcessing">
<button class="btn-sm" @click="emit('reshelve', doc.id)" :disabled="isProcessing">
Re-index
</button>
<button class="btn-sm danger" @click="emit('delete', doc.id)">Remove</button>
@ -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;

View file

@ -1,5 +1,5 @@
<template>
<div class="ingest-progress" v-if="visible">
<div class="shelve-progress" v-if="visible">
<div class="progress-label">
<span>{{ statusLabel }}</span>
<span class="progress-pct" v-if="status?.progress != null">{{ status.progress }}%</span>
@ -74,7 +74,7 @@ onUnmounted(stopPoll)
</script>
<style scoped>
.ingest-progress { margin-top: 0.5rem; }
.shelve-progress { margin-top: 0.5rem; }
.progress-label { display: flex; justify-content: space-between; font-size: 0.8rem; color: var(--color-text-muted); margin-bottom: 4px; }
.progress-bar { height: 4px; background: var(--color-border); border-radius: 2px; overflow: hidden; }
.progress-fill { height: 100%; background: var(--color-accent); transition: width 0.3s ease; }

View file

@ -4,9 +4,9 @@
<h1>Library</h1>
<div class="header-actions">
<button class="btn-secondary" @click="triggerUpload" :disabled="uploading">
{{ uploading ? "Uploading..." : "Upload PDF / EPUB" }}
{{ uploading ? "Uploading..." : "Upload Document or Spreadsheet" }}
</button>
<input ref="fileInput" type="file" accept=".pdf,.epub" style="display:none" @change="handleUpload">
<input ref="fileInput" type="file" accept=".pdf,.epub,.docx,.odt,.pages,.xlsx,.ods,.numbers" style="display:none" @change="handleUpload">
<button class="btn-primary" @click="scan" :disabled="scanning">
{{ scanning ? "Scanning..." : "Scan for PDFs" }}
</button>
@ -26,7 +26,7 @@
v-for="doc in docs"
:key="doc.id"
:doc="doc"
@reingest="reingest"
@reshelve="reshelve"
@delete="remove"
@refresh="load"
/>
@ -76,10 +76,10 @@ async function scan() {
}
}
async function reingest(id: string) {
async function reshelve(id: string) {
error.value = null
try {
await api.reingestDocument(id)
await api.reshelveDocument(id)
await load()
} catch (e) {
error.value = e instanceof Error ? e.message : "Re-index failed"