Website and PDF RAG

Implement this recipe in my project:
https://docs.context.dev/use-cases/build-rag-from-websites.md

Read the recipe and linked API guides, inspect this project's stack, and build the flow using its existing conventions.

Build retrieval from website pages and uploaded documents. Preserve source identity, page-range uncertainty, and access controls. Retain the last good index after failed refreshes, and verify citations against retrieved evidence.

Reuse existing Context.dev configuration and keep secret API keys on the server. If Context.dev is not set up yet, follow https://docs.context.dev/agent-quickstart.md first. Add focused tests, run the relevant checks, and document setup and how to try the result.

export function canonicalSourceUrl(raw: string) {
  const url = new URL(raw);
  if (!["https:", "http:"].includes(url.protocol)) throw new Error("Expected a web URL");
  url.hash = "";
  for (const name of [...url.searchParams.keys()]) {
    if (name.startsWith("utm_") || name === "gclid" || name === "fbclid") {
      url.searchParams.delete(name);
    }
  }
  return url.href;
}

const uniquePages = new Map<string, (typeof pages)[number]>();
for (const page of pages) {
  const key = canonicalSourceUrl(page.metadata.url);
  const existing = uniquePages.get(key);
  if (existing && existing.markdown !== page.markdown) {
    throw new Error("Review conflicting content for one canonical URL");
  }
  if (!existing) uniquePages.set(key, page);
}

import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen

MAX_BYTES = 50 * 1024 * 1024

def parse_source(path, title, open_url, start=1, end=5, ocr=False):
    if start < 1 or end < start:
        raise ValueError("Use an inclusive page range starting at 1")
    path = Path(path)
    if path.stat().st_size > MAX_BYTES:
        raise ValueError("Split the PDF before uploading: maximum 50 MiB")
    body = path.read_bytes()
    if len(body) > MAX_BYTES:
        raise ValueError("File changed or exceeds 50 MiB")

    file_hash = hashlib.sha256(body).hexdigest()
    source_id = hashlib.sha256(
        f"{file_hash}:{start}:{end}:{ocr}".encode()
    ).hexdigest()
    source = {
        "id": source_id,
        "title": title,
        "open_url": open_url,
        "file_hash": file_hash,
        "requested_pages": {"start": start, "end": end},
        "retrieved_at": datetime.now(timezone.utc).isoformat(),
        "ocr_requested": ocr,
        "status": "pending",
        "markdown": "",
    }
    query = urlencode({
        "extension": "pdf", "pdf[start]": start, "pdf[end]": end,
        "ocr": str(ocr).lower(),
    })
    request = Request(
        f"https://api.context.dev/v1/parse?{query}",
        data=body,
        method="POST",
        headers={
            "Authorization": f"Bearer {os.environ['CONTEXT_DEV_API_KEY']}",
            "Content-Type": "application/pdf",
        },
    )
    try:
        with urlopen(request, timeout=60) as response:
            result = json.load(response)
        text = result.get("markdown", "")
        source["status"] = "ready" if result.get("success") and text.strip() else "empty"
        source["markdown"] = text if source["status"] == "ready" else ""
        source["detected_type"] = result.get("type")
    except HTTPError as error:
        try:
            code = json.loads(error.read()).get("error_code")
        except (ValueError, AttributeError):
            code = None
        finally:
            error.close()
        source["status"] = "ocr_required" if code == "PDF_IMAGES_ONLY" else "failed"
        source["http_status"] = error.code
        source["error_code"] = code
    except (URLError, TimeoutError, ValueError):
        source["status"] = "failed"
    return source

import hashlib

def source_chunks(source, max_chars=3500):
    if max_chars < 1:
        raise ValueError("max_chars must be positive")
    if source["status"] != "ready":
        return []
    texts = []
    pending = ""
    for paragraph in source["markdown"].split("\n\n"):
        if len(pending) + len(paragraph) + 2 > max_chars and pending:
            texts.append(pending)
            pending = ""
        while len(paragraph) > max_chars:
            texts.append(paragraph[:max_chars])
            paragraph = paragraph[max_chars:]
        if paragraph.strip():
            pending = f"{pending}\n\n{paragraph}".strip()
    if pending:
        texts.append(pending)
    return [{
        "id": hashlib.sha256(f"{source['id']}:{index}:{text}".encode()).hexdigest(),
        "source_id": source["id"],
        "title": source["title"],
        "open_url": source["open_url"],
        "requested_pages": source["requested_pages"],
        "retrieved_at": source["retrieved_at"],
        "text": text,
    } for index, text in enumerate(texts)]

def resolve_evidence(claims, retrieved_chunks):
    available = {chunk["id"]: chunk for chunk in retrieved_chunks}
    resolved = []
    for claim in claims:
        if not claim.get("evidence"):
            raise ValueError("Claim has no retrieved evidence")
        citations = []
        for item in claim["evidence"]:
            chunk = available.get(item["chunk_id"])
            quote = item.get("quote", "").strip()
            if chunk is None or not quote or quote not in chunk["text"]:
                raise ValueError("Citation does not resolve to retrieved text")
            citations.append({
                "source_id": chunk["source_id"], "title": chunk["title"],
                "url": chunk["open_url"], "excerpt": quote,
                "requested_pages": chunk["requested_pages"],
            })
        resolved.append({"text": claim["text"], "citations": citations})
    return resolved