Feat: Add scripts for file processing and vector embeddings

.gitignore

@@ -51,4 +51,5 @@ docker-compose.override.yml
 # OS files
 # ======================
 .DS_Store
-Thumbs.db
+Thumbs.db
+test_data/

app/example_usage.py

@@ -0,0 +1,126 @@
+"""
+example_usage.py
+----------------
+Shows how a FastAPI background task / CLI script would call the
+processing package and persist results using SQLAlchemy.
+
+This file is NOT part of the processing package itself — it lives
+alongside it as reference for the backend team.
+"""
+
+# ── SQLAlchemy models (abbreviated) ─────────────────────────────────────────
+# from app.models import File, FileContent
+# from app.database import SessionLocal
+
+# ── Processing package ───────────────────────────────────────────────────────
+from processing import process
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 1.  Basic full-directory scan
+# ─────────────────────────────────────────────────────────────────────────────
+
+def index_directory(root: str, db) -> None:
+    """
+    Index every supported file under *root* and persist to the database.
+
+    Parameters
+    ----------
+    root : str
+        Folder or single file path to scan.
+    db   : SQLAlchemy Session
+        Caller is responsible for lifecycle (commit / rollback / close).
+    """
+    file_rows, content_rows = process(root)
+
+    # ── Insert File rows ──────────────────────────────────────────────────────
+    for fr in file_rows:
+        # file_rows keys: file_path, file_hash, mime_type, last_modified, tags
+        existing = db.query(File).filter_by(file_path=fr["file_path"]).first()
+        if existing:
+            # Update hash / timestamp in case the file changed
+            for k, v in fr.items():
+                setattr(existing, k, v)
+        else:
+            db.add(File(**fr))
+
+    db.flush()   # ensure File PKs exist before FK inserts
+
+    # ── Insert FileContent rows ───────────────────────────────────────────────
+    for cr in content_rows:
+        file_path = cr.pop("file_path")   # resolve FK
+        file_obj  = db.query(File).filter_by(file_path=file_path).one()
+
+        db.add(FileContent(
+            file_id       = file_obj.id,
+            chunk_index   = cr["chunk_index"],
+            content_text  = cr["content_text"],
+            embedding     = cr["embedding"],
+            # keyword_tokens is a server-generated TSVECTOR column;
+            # PostgreSQL fills it automatically — do not pass it here.
+        ))
+
+    db.commit()
+    print(f"Indexed {len(file_rows)} file(s), {len(content_rows)} chunk(s).")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 2.  Incremental re-index (skip unchanged files)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def reindex_directory(root: str, db) -> None:
+    """Like index_directory but skips files whose SHA-256 hash is unchanged."""
+    existing_hashes = {row.file_hash for row in db.query(File.file_hash).all()}
+    file_rows, content_rows = process(root, skip_hashes=existing_hashes)
+
+    for fr in file_rows:
+        db.merge(File(**fr))   # upsert on unique file_path
+
+    db.flush()
+
+    for cr in content_rows:
+        file_path = cr.pop("file_path")
+        file_obj  = db.query(File).filter_by(file_path=file_path).one()
+        db.add(FileContent(
+            file_id      = file_obj.id,
+            chunk_index  = cr["chunk_index"],
+            content_text = cr["content_text"],
+            embedding    = cr["embedding"],
+        ))
+
+    db.commit()
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 3.  FastAPI background task example
+# ─────────────────────────────────────────────────────────────────────────────
+
+# from fastapi import BackgroundTasks
+#
+# @app.post("/index")
+# async def trigger_index(path: str, background_tasks: BackgroundTasks):
+#     background_tasks.add_task(index_directory, path, SessionLocal())
+#     return {"status": "indexing started", "path": path}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 4.  CLI quick-test
+# ─────────────────────────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    import sys, pprint
+    root = sys.argv[1] if len(sys.argv) > 1 else "."
+    file_rows, content_rows = process(root)
+
+    print(f"\n{'='*60}")
+    print(f"Files found   : {len(file_rows)}")
+    print(f"Chunks created: {len(content_rows)}")
+    print(f"{'='*60}\n")
+
+    for cr in content_rows[:3]:
+        pprint.pprint({
+            "file_path":    cr["file_path"],
+            "chunk_index":  cr["chunk_index"],
+            "content_text": cr["content_text"][:120] + "…",
+            "embedding_dim": len(cr["embedding"]),
+        })

app/processing/__init__.py

@@ -0,0 +1,27 @@
+"""
+processing
+----------
+WhereTF local-search processing system.
+
+Public surface
+~~~~~~~~~~~~~~
+
+    from processing import process
+
+    file_rows, content_rows = process("/path/to/scan")
+
+    # file_rows  → list of dicts matching the `File` SQLAlchemy model
+    # content_rows → list of dicts matching the `FileContent` model
+
+Lower-level modules
+~~~~~~~~~~~~~~~~~~~
+* ``traversal``  – recursive file walker
+* ``extractors`` – per-format text + embedded-OCR extractors
+* ``ocr``        – EasyOCR singleton wrapper
+* ``embeddings`` – batched sentence-transformer inference
+* ``pipeline``   – top-level orchestrator (re-exported as ``process``)
+"""
+
+from .pipeline import process
+
+__all__ = ["process"]

app/processing/embeddings.py

@@ -0,0 +1,109 @@
+"""
+embeddings.py
+-------------
+Batched embedding pipeline for WhereTF.
+
+Takes a flat list of ``ChunkData`` dicts (produced by the extractors) and
+mutates each one in-place by adding an ``"embedding"`` key whose value is a
+plain Python ``list[float]`` of 384 dimensions — ready to be passed straight
+into the ``FileContent`` SQLAlchemy model.
+
+The model is loaded once and cached for the lifetime of the process.
+"""
+
+from __future__ import annotations
+
+import logging
+from functools import lru_cache
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .extractors import ChunkData
+
+logger = logging.getLogger(__name__)
+
+MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
+DEFAULT_BATCH_SIZE = 64
+
+
+# ---------------------------------------------------------------------------
+# Lazy model singleton
+# ---------------------------------------------------------------------------
+
+@lru_cache(maxsize=1)
+def _get_model():
+    """Load and cache the SentenceTransformer model (once per process)."""
+    try:
+        from sentence_transformers import SentenceTransformer  # type: ignore
+    except ImportError as exc:
+        raise RuntimeError(
+            "sentence-transformers is required. "
+            "Install it with:  pip install sentence-transformers"
+        ) from exc
+
+    logger.info("[embeddings] Loading model '%s' …", MODEL_NAME)
+    model = SentenceTransformer(MODEL_NAME)
+    logger.info("[embeddings] Model loaded.")
+    return model
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def embed_chunks(
+    chunks: list["ChunkData"],
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> list["ChunkData"]:
+    """
+    Add an ``"embedding"`` field to every chunk dict **in-place** and
+    return the same list for convenient chaining.
+
+    Parameters
+    ----------
+    chunks:
+        Output of :func:`~processing.extractors.extract` — one dict per
+        text chunk.  Dicts that already have an ``"embedding"`` key are
+        skipped so this function is safe to call multiple times.
+    batch_size:
+        Number of texts to encode per forward pass.  64 is a good default
+        for CPU; increase to 256+ when a GPU is available.
+
+    Returns
+    -------
+    list[ChunkData]
+        The same list, each dict now containing:
+        ``{"chunk_index": int, "content_text": str, "embedding": list[float]}``
+    """
+    if not chunks:
+        return chunks
+
+    # Separate chunks that still need embeddings
+    pending_indices = [i for i, c in enumerate(chunks) if "embedding" not in c]
+    if not pending_indices:
+        return chunks
+
+    model = _get_model()
+    texts = [chunks[i]["content_text"] for i in pending_indices]
+
+    logger.info(
+        "[embeddings] Encoding %d chunk(s) in batches of %d …",
+        len(texts),
+        batch_size,
+    )
+
+    # encode() returns a numpy array of shape (N, 384)
+    vectors = model.encode(
+        texts,
+        batch_size=batch_size,
+        show_progress_bar=len(texts) > batch_size,
+        convert_to_numpy=True,
+        normalize_embeddings=True,   # unit-normalised → cosine ≡ dot product
+    )
+
+    for list_pos, chunk_idx in enumerate(pending_indices):
+        # Store as plain Python list[float] — JSON-serialisable, pgvector-ready
+        chunks[chunk_idx]["embedding"] = vectors[list_pos].tolist()
+
+    logger.info("[embeddings] Done.")
+    return chunks