Document ingestion pipelines for RAG

import hashlib
import sqlite3
from pathlib import Path
from datetime import datetime

# A small docs/ folder to ingest, created here so the script is fully
# self-contained -- in a real pipeline these files already exist on disk.
Path("docs").mkdir(exist_ok=True)
Path("docs/refund-policy.txt").write_text(
    "Refunds are processed within 7 business days of the return being received.\n"
    "Store credit is issued immediately if the customer chooses that option instead of a bank refund.\n"
)
Path("docs/shipping-policy.txt").write_text(
    "Standard shipping takes 5 to 8 business days within the country for most locations.\n"
    "Express shipping is available at checkout and typically arrives within 2 business days.\n"
    "International shipping times vary by destination and customs processing at the border.\n"
)

conn = sqlite3.connect(":memory:")
conn.execute("""
CREATE TABLE chunks (
    chunk_id TEXT PRIMARY KEY, source TEXT, chunk_index INTEGER,
    text TEXT, ingested_at TEXT
)
""")
# The ledger: one row per file already ingested, keyed by a hash of its
# contents. This is what makes re-running the pipeline safe.
conn.execute("""
CREATE TABLE ingested_files (path TEXT PRIMARY KEY, content_hash TEXT, ingested_at TEXT)
""")
conn.commit()


def chunk_text(text, max_chars=120):
    paras = [p.strip() for p in text.split("\n") if p.strip()]
    chunks, current = [], ""
    for p in paras:
        if len(current) + len(p) > max_chars and current:
            chunks.append(current)
            current = p
        else:
            current = (current + " " + p).strip()
    if current:
        chunks.append(current)
    return chunks


def ingest_folder(folder):
    added, skipped = 0, 0
    for path in sorted(Path(folder).glob("*.txt")):
        content = path.read_text()
        content_hash = hashlib.sha256(content.encode()).hexdigest()

        existing = conn.execute(
            "SELECT content_hash FROM ingested_files WHERE path = ?", (str(path),)
        ).fetchone()
        if existing and existing[0] == content_hash:
            skipped += 1
            continue  # already ingested, and the file has not changed

        for i, chunk in enumerate(chunk_text(content)):
            chunk_id = hashlib.sha256(f"{path}:{i}".encode()).hexdigest()[:12]
            conn.execute(
                "INSERT OR REPLACE INTO chunks VALUES (?,?,?,?,?)",
                (chunk_id, path.name, i, chunk, datetime.now().isoformat()),
            )
        conn.execute(
            "INSERT OR REPLACE INTO ingested_files VALUES (?,?,?)",
            (str(path), content_hash, datetime.now().isoformat()),
        )
        added += 1
    conn.commit()
    return added, skipped


added, skipped = ingest_folder("docs")
print(f"first run  -> files newly ingested: {added}, files skipped: {skipped}")
n_chunks = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"chunks stored: {n_chunks}")

# Run it again with nothing changed on disk -- an idempotent pipeline
# should do no duplicate work.
added2, skipped2 = ingest_folder("docs")
print(f"\nsecond run -> files newly ingested: {added2}, files skipped: {skipped2}")
n_chunks2 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"chunks stored: {n_chunks2}  (unchanged)")

sample = conn.execute("SELECT source, chunk_index, text FROM chunks LIMIT 2").fetchall()
print("\nsample chunks:")
for s in sample:
    print(" ", s)