Compare commits
10 commits
fix/gpu-se
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 059ad21ed3 | |||
| afdc211813 | |||
| b0df7ac7bc | |||
| f548c82c96 | |||
| c684f347e9 | |||
| 156f52d6ac | |||
| 89a58ec9b0 | |||
| d39cfbd87a | |||
| fccb30c79d | |||
| f941ebdeeb |
51 changed files with 2642 additions and 218 deletions
16
.env.example
16
.env.example
|
|
@ -6,10 +6,20 @@ PAGEPIPER_BOOKS_DIR=/path/to/your/pdfs
|
|||
# Data directory (SQLite + vector DB stored here)
|
||||
PAGEPIPER_DATA_DIR=data
|
||||
|
||||
# Ollama URL — set this to unlock semantic search and RAG chat (BYOK)
|
||||
# LLM backend — either option (or both) unlocks semantic search and RAG chat.
|
||||
#
|
||||
# Option A: direct Ollama URL
|
||||
# PAGEPIPER_OLLAMA_URL=http://localhost:11434
|
||||
# PAGEPIPER_CHAT_MODEL=mistral:7b
|
||||
# PAGEPIPER_EMBED_MODEL=nomic-embed-text
|
||||
#
|
||||
# Option B: cf-orch coordinator (resolves service URL via GPU allocation).
|
||||
# Set CF_ORCH_URL alone — no PAGEPIPER_OLLAMA_URL needed.
|
||||
# PAGEPIPER_OLLAMA_URL is used as a fallback if cf-orch is unreachable.
|
||||
# CF_ORCH_URL=http://localhost:7700
|
||||
# CF_APP_NAME=pagepiper
|
||||
# PAGEPIPER_ORCH_SERVICE=ollama # or cf-text for managed transformer inference
|
||||
#
|
||||
PAGEPIPER_CHAT_MODEL=mistral:7b
|
||||
PAGEPIPER_EMBED_MODEL=nomic-embed-text
|
||||
|
||||
# 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
3
.gitignore
vendored
|
|
@ -29,3 +29,6 @@ compose.override.yml
|
|||
# Logs and runtime files
|
||||
*.log
|
||||
*.db
|
||||
|
||||
# Playwright MCP scratch dir (test fixtures, screenshots — never commit)
|
||||
.playwright-mcp/
|
||||
|
|
|
|||
|
|
@ -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: ..)
|
||||
|
|
|
|||
30
README.md
30
README.md
|
|
@ -4,9 +4,9 @@
|
|||
|
||||
[](https://git.opensourcesolarpunk.com/Circuit-Forge/pagepiper)
|
||||
[](LICENSE)
|
||||
[](https://git.opensourcesolarpunk.com/Circuit-Forge/pagepiper/releases)
|
||||
[](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
|
||||
|
||||

|
||||

|
||||
|
||||
### 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)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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=?",
|
||||
|
|
|
|||
|
|
@ -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] = {}
|
||||
|
||||
|
||||
|
|
@ -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),
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
19
app/main.py
19
app/main.py
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
137
app/services/colbert_index.py
Normal file
137
app/services/colbert_index.py
Normal 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
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||

|
||||
{ .only-light }
|
||||
{ .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
|
||||
|
||||

|
||||
{ .only-light }
|
||||
{ .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.
|
||||
|
|
|
|||
|
|
@ -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) |
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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) |
|
||||
|
|
|
|||
BIN
docs/screenshots/01-library-dark.png
Normal file
BIN
docs/screenshots/01-library-dark.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 28 KiB |
BIN
docs/screenshots/02-chat-dark.png
Normal file
BIN
docs/screenshots/02-chat-dark.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 27 KiB |
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
15
mkdocs.yml
15
mkdocs.yml
|
|
@ -1,5 +1,5 @@
|
|||
site_name: Pagepiper
|
||||
site_description: Self-hosted PDF and EPUB library with BM25 full-text search, hybrid vector retrieval, and LLM-powered RAG chat.
|
||||
site_description: Self-hosted document and spreadsheet library (PDF, EPUB, DOCX, ODT, Apple Pages, XLSX, ODS, Apple Numbers) with BM25 full-text search, hybrid vector retrieval, and LLM-powered RAG chat.
|
||||
site_author: Circuit Forge LLC
|
||||
site_url: https://docs.circuitforge.tech/pagepiper
|
||||
repo_url: https://git.opensourcesolarpunk.com/Circuit-Forge/pagepiper
|
||||
|
|
@ -9,14 +9,14 @@ theme:
|
|||
name: material
|
||||
palette:
|
||||
- scheme: default
|
||||
primary: deep purple
|
||||
accent: purple
|
||||
primary: brown
|
||||
accent: orange
|
||||
toggle:
|
||||
icon: material/brightness-7
|
||||
name: Switch to dark mode
|
||||
- scheme: slate
|
||||
primary: deep purple
|
||||
accent: purple
|
||||
primary: brown
|
||||
accent: amber
|
||||
toggle:
|
||||
icon: material/brightness-4
|
||||
name: Switch to light mode
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
280
scripts/shelve_docx.py
Normal 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,
|
||||
)
|
||||
|
|
@ -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
101
scripts/shelve_numbers.py
Normal 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
211
scripts/shelve_ods.py
Normal 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
268
scripts/shelve_odt.py
Normal 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
100
scripts/shelve_pages.py
Normal 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,
|
||||
)
|
||||
|
|
@ -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
208
scripts/shelve_xlsx.py
Normal 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,
|
||||
)
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
168
tests/test_colbert_index.py
Normal 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"
|
||||
|
|
@ -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
113
tests/test_retriever.py
Normal 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 == []
|
||||
|
|
@ -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
129
tests/test_shelve_docx.py
Normal 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"
|
||||
118
tests/test_shelve_numbers.py
Normal file
118
tests/test_shelve_numbers.py
Normal 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
135
tests/test_shelve_ods.py
Normal 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
135
tests/test_shelve_odt.py
Normal 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
122
tests/test_shelve_pages.py
Normal 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
125
tests/test_shelve_xlsx.py
Normal 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"
|
||||
|
|
@ -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()
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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; }
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Reference in a new issue