Keep your index in sync with your website

import hashlib, json, re
from clickruntime import on_trigger, credentials, notify, http, state

HOST = "https://docs-search-abc123.svc.us-east-1-aws.pinecone.io"  # index docs-search
SITEMAP = "https://example.com/sitemap.xml"

@on_trigger
def run(event):
    headers = {"Api-Key": credentials.secret("PINECONE_API_KEY"), "X-Pinecone-API-Version": "2025-04"}
    hashes = state.get("hashes") or {}
    urls = re.findall(r"<loc>(.*?)</loc>", http.get(SITEMAP).text)
    added = updated = 0
    for url in urls:
        text = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", http.get(url).text)).strip()
        digest = hashlib.sha256(text.encode()).hexdigest()
        if hashes.get(url) == digest:
            continue
        added, updated = (added + 1, updated) if url not in hashes else (added, updated + 1)
        chunks = [text[i:i + 2000] for i in range(0, len(text), 2000)][:50]
        body = "\n".join(json.dumps({"_id": f"{url}#{n}", "chunk_text": c, "url": url}) for n, c in enumerate(chunks))
        http.post(f"{HOST}/records/namespaces/docs/upsert", data=body,
                  headers={**headers, "Content-Type": "application/x-ndjson"}).raise_for_status()
        hashes[url] = digest
    removed = [u for u in hashes if u not in urls]
    if removed:
        http.post(f"{HOST}/vectors/delete", headers=headers,
                  json={"namespace": "docs", "filter": {"url": {"$in": removed}}}).raise_for_status()
        for u in removed:
            hashes.pop(u)
    state.set("hashes", hashes)
    notify.email_me("Pinecone docs-search sync", f"Added {added}, updated {updated}, removed {len(removed)} pages.")
    return {"added": added, "updated": updated, "removed": len(removed)}