How to Ingest Confluence Pages into Pinecone with EnConvert

import json
import os
import time

import requests
from pinecone import Pinecone
from sentence_transformers import SentenceTransformer

ENCONVERT_API_KEY = os.environ["ENCONVERT_API_KEY"]
PINECONE_API_KEY = os.environ["PINECONE_API_KEY"]
# An existing Pinecone index with dimension 384 and the cosine metric
PINECONE_INDEX_NAME = os.environ["PINECONE_INDEX_NAME"]
# A publicly accessible Confluence page, e.g.
# https://your-org.atlassian.net/wiki/spaces/ENG/pages/123456/deployment-runbook
PAGE_URL = os.environ["PAGE_URL"]
# An H1 to H3 heading on that page; replace it with a heading from your own page
SECTION = "Rollback"

HEADERS = {"X-API-Key": ENCONVERT_API_KEY}

# Step 1: Submit the Confluence page URL to the Ingest endpoint
submit = requests.post(
    "https://api.enconvert.com/v2/ingest",
    headers=HEADERS,
    json={
        "mode": "urls",
        "urls": [PAGE_URL],
        # all-MiniLM-L6-v2 only reads the first 256 tokens, so keep chunks small
        "chunk": {"max_words": 200},
    },
    timeout=60,
)
submit.raise_for_status()
job_id = submit.json()["job_id"]

# Step 2: Poll until the job completes
# A page behind a login is skipped, so the job fails with an error_message
while True:
    poll = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=60,
    )
    poll.raise_for_status()
    job = poll.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(
            f"Ingest job {job_id} {job['status']}: {job.get('error_message')}"
        )
    time.sleep(5)

# Step 3: Download the JSONL output, one chunk per line
# The signed URL is valid for 15 minutes. Plain GET, no API key.
output = requests.get(job["output_url"], timeout=60)
output.raise_for_status()
chunks = [
    json.loads(line)
    for line in output.content.decode("utf-8").splitlines()
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks ready for embedding")

# Step 4: Embed each chunk (384-dimensional, normalized vectors)
model = SentenceTransformer("all-MiniLM-L6-v2")
embeddings = model.encode(
    [chunk["content"] for chunk in chunks], normalize_embeddings=True
)

# Step 5: Connect to Pinecone
pc = Pinecone(api_key=PINECONE_API_KEY)
index = pc.index(name=PINECONE_INDEX_NAME)

# Step 6: Upsert each chunk with metadata mapped from the chunk metadata
vectors = []
for chunk, embedding in zip(chunks, embeddings):
    metadata = chunk["metadata"]
    vectors.append({
        "id": chunk["id"],
        "values": embedding.tolist(),
        "metadata": {
            "text": chunk["content"],
            "title": metadata["title"],
            "section": metadata["section"],
            # Pinecone accepts a list of strings as a metadata value
            "headings_path": metadata["headings_path"],
            "source_url": metadata["source_url"],
            "chunk_index": metadata["chunk_index"],
            "word_count": metadata["word_count"],
        },
    })

# Upsert in batches of 100
for start in range(0, len(vectors), 100):
    index.upsert(vectors=vectors[start:start + 100])

print(f"Upserted {len(vectors)} chunks into Pinecone")

# Pinecone is eventually consistent, so wait briefly before querying
time.sleep(10)

# Step 7: Query narrowed to one page and one section by a metadata filter
query_vector = model.encode(
    "how do we roll back a failed deployment", normalize_embeddings=True
).tolist()

results = index.query(
    vector=query_vector,
    top_k=3,
    include_metadata=True,
    filter={
        "$and": [
            {"source_url": {"$eq": PAGE_URL}},
            {"section": {"$eq": SECTION}},
        ]
    },
)

for match in results.matches:
    print(" > ".join(match.metadata["headings_path"]))
    print(match.metadata["text"][:200])