Web Scraping Pipeline for RAG: Clean Data for LLMs

import httpx

ALTERLAB_API_KEY = "YOUR_API_KEY"
ALTERLAB_BASE_URL = "https://api.alterlab.io/v1"


def fetch_page(url: str, render_js: bool = False) -> str:
    """Fetch fully-rendered HTML from any URL."""
    response = httpx.post(
        f"{ALTERLAB_BASE_URL}/scrape",
        headers={"X-API-Key": ALTERLAB_API_KEY, "Content-Type": "application/json"},
        json={
            "url": url,
            "render_js": render_js,
            "wait_for": "networkidle" if render_js else None,
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["html"]

curl -X POST https://api.alterlab.io/v1/scrape \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com/docs/getting-started",
    "render_js": true,
    "wait_for": "networkidle"
  }'

import json
import trafilatura
from trafilatura.settings import use_config

# Disable per-document timeout—let your own retry logic own the clock
config = use_config()
config.set("DEFAULT", "EXTRACTION_TIMEOUT", "0")


def extract_content(html: str, url: str) -> dict:
    """
    Extract main content from HTML.
    Returns dict with keys: text, title, author, date, description.
    Raises ValueError if no content can be extracted.
    """
    result = trafilatura.extract(
        html,
        url=url,
        include_comments=False,
        include_tables=True,
        no_fallback=False,
        config=config,
        output_format="json",
        with_metadata=True,
    )

    if result is None:
        raise ValueError(f"Extraction returned no content for {url}")

    return json.loads(result)

import re
import unicodedata


def normalize_text(text: str) -> str:
    # Canonical Unicode form: convert smart quotes, em-dashes, ligatures
    text = unicodedata.normalize("NFKC", text)

    # Replace invisible/non-breaking whitespace variants
    text = re.sub(r"[\u00a0\u200b\u200c\u200d\ufeff]", " ", text)

    # Collapse horizontal whitespace, preserve single newlines
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)

    # Drop lines shorter than 4 chars (nav artifacts: "›", "|", "»")
    lines = [ln for ln in text.split("\n") if len(ln.strip()) > 3 or ln.strip() == ""]

    return "\n".join(lines).strip()

from __future__ import annotations
import re
from dataclasses import dataclass, field


@dataclass
class Chunk:
    text: str
    url: str
    chunk_index: int
    total_chunks: int
    metadata: dict = field(default_factory=dict)


def split_sentences(text: str) -> list[str]:
    """Sentence-boundary split on terminal punctuation followed by uppercase."""
    return re.split(r"(?<=[.!?])\s+(?=[A-Z\"])", text)


def chunk_document(
    text: str,
    url: str,
    max_tokens: int = 400,
    overlap_sentences: int = 2,
    chars_per_token: float = 4.0,
) -> list[Chunk]:
    """
    Split text into token-bounded chunks with sentence-level overlap.

    Args:
        max_tokens: Approximate token ceiling per chunk.
        overlap_sentences: Sentences carried over to the next chunk.
        chars_per_token: Heuristic for English prose (4.0 is reliable).
    """
    max_chars = int(max_tokens * chars_per_token)
    sentences = split_sentences(text)

    raw_chunks: list[str] = []
    current: list[str] = []
    current_len = 0

    for sentence in sentences:
        slen = len(sentence)
        if current_len + slen > max_chars and current:
            raw_chunks.append(" ".join(current))
            current = current[-overlap_sentences:]
            current_len = sum(len(s) for s in current)
        current.append(sentence)
        current_len += slen

    if current:
        raw_chunks.append(" ".join(current))

    total = len(raw_chunks)
    return [
        Chunk(text=t, url=url, chunk_index=i, total_chunks=total)
        for i, t in enumerate(raw_chunks)
    ]

import asyncio
from openai import AsyncOpenAI
import pinecone

openai_client = AsyncOpenAI()


def init_index(api_key: str, index_name: str) -> pinecone.Index:
    pc = pinecone.Pinecone(api_key=api_key)
    return pc.Index(index_name)


async def embed_texts(texts: list[str]) -> list[list[float]]:
    """Batch embed up to 2048 texts in a single API call."""
    response = await openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
        encoding_format="float",
    )
    return [item.embedding for item in response.data]


async def index_chunks(
    chunks: list["Chunk"],
    index: pinecone.Index,
    batch_size: int = 100,
) -> None:
    """Embed and upsert chunks into Pinecone with source metadata preserved."""
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i : i + batch_size]
        vectors = await embed_texts([c.text for c in batch])

        upserts = [
            {
                "id": f"{c.url}::{c.chunk_index}",
                "values": vectors[j],
                "metadata": {
                    "url": c.url,
                    "chunk_index": c.chunk_index,
                    "total_chunks": c.total_chunks,
                    "text": c.text,  # store inline—avoids a separate fetch at query time
                },
            }
            for j, c in enumerate(batch)
        ]

        index.upsert(vectors=upserts)

import asyncio
import httpx
from fetch import fetch_page
from extract import extract_content
from normalize import normalize_text
from chunker import Chunk, chunk_document
from embed import init_index, index_chunks

PINECONE_API_KEY = "YOUR_PINECONE_KEY"
PINECONE_INDEX = "rag-knowledge-base"


async def ingest_url(url: str, render_js: bool = False) -> dict:
    """
    End-to-end pipeline: URL → indexed, retrievable chunks.
    Returns a summary dict for logging and monitoring.
    """
    # Stage 1: Fetch
    html = fetch_page(url, render_js=render_js)

    # Stage 2: Extract
    extracted = extract_content(html, url)
    raw_text = extracted.get("text", "")
    title = extracted.get("title", "untitled")

    if not raw_text:
        return {"url": url, "status": "no_content", "chunks": 0}

    # Stage 3: Normalize
    clean_text = normalize_text(raw_text)

    # Stage 4: Chunk
    chunks = chunk_document(
        text=clean_text,
        url=url,
        max_tokens=400,
        overlap_sentences=2,
    )

    # Filter degenerate chunks before embedding
    chunks = [c for c in chunks if len(c.text.split()) >= 15]

    # Stage 5: Embed + index
    index = init_index(PINECONE_API_KEY, PINECONE_INDEX)
    await index_chunks(chunks, index)

    return {
        "url": url,
        "title": title,
        "status": "indexed",
        "chunks": len(chunks),
        "approx_tokens": sum(len(c.text) // 4 for c in chunks),
    }


async def ingest_batch(urls: list[str], concurrency: int = 5) -> list[dict]:
    """Ingest multiple URLs with bounded concurrency."""
    semaphore = asyncio.Semaphore(concurrency)

    async def bounded(url: str) -> dict:
        async with semaphore:
            try:
                return await ingest_url(url)
            except Exception as e:
                return {"url": url, "status": "error", "error": str(e)}

    return await asyncio.gather(*[bounded(u) for u in urls])


if __name__ == "__main__":
    urls = [
        "https://docs.python.org/3/library/asyncio-task.html",
        "https://platform.openai.com/docs/guides/embeddings",
        "https://www.pinecone.io/docs/upsert-data/",
    ]
    results = asyncio.run(ingest_batch(urls, concurrency=3))
    for r in results:
        print(r)

import hashlib


_seen_hashes: set[str] = set()


def is_duplicate(text: str) -> bool:
    """Return True if this exact content has already been indexed this run."""
    digest = hashlib.sha256(text.encode()).hexdigest()[:16]
    if digest in _seen_hashes:
        return True
    _seen_hashes.add(digest)
    return False

import asyncio
from typing import TypeVar, Callable, Awaitable

T = TypeVar("T")


async def with_retry(fn: Callable[[], Awaitable[T]], attempts: int = 3) -> T:
    for i in range(attempts):
        try:
            return await fn()
        except Exception:
            if i == attempts - 1:
                raise
            await asyncio.sleep(min(2 ** i, 30))
    raise RuntimeError("unreachable")