How to Ingest CSV into Qdrant with EnConvert

import json
import os
import time
import uuid

import requests
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    FieldCondition,
    Filter,
    MatchValue,
    PayloadSchemaType,
    PointStruct,
    VectorParams,
)
from sentence_transformers import SentenceTransformer

HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}
COLLECTION = "csv_chunks"
CSV_PATH = "/path/to/your/product-catalogue.csv"

# Step 1: Upload the CSV (the multipart field is "files", plural)
with open(CSV_PATH, "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files=[("files", (os.path.basename(CSV_PATH), f, "text/csv"))],
        timeout=60,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job completes (give up after 15 minutes)
deadline = time.monotonic() + 15 * 60
while True:
    poll = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=30,
    )
    poll.raise_for_status()
    job = poll.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(f"Ingest {job['status']}: {job.get('error_message')}")
    if time.monotonic() > deadline:
        raise TimeoutError(f"Ingest job {job_id} is still {job['status']}")
    time.sleep(2)

# Step 3: Download the JSONL output. output_url is a signed URL (valid 15
# minutes), fetched with a plain GET and no API key.
jsonl = requests.get(job["output_url"], timeout=60)
jsonl.raise_for_status()

# One JSON object per line. A CSV is a single table with no headings, so Ingest
# keeps it whole as one chunk: headings_path is [] and section is "". For an
# uploaded file, metadata.source_url is the filename.
chunks = [
    json.loads(line)
    for line in jsonl.content.decode("utf-8").split("\n")
    if line.strip()
]

# Step 4: The whole table is one chunk, but all-MiniLM-L6-v2 embeds only the
# first 256 word pieces of its input and ignores the rest. So cut the table into
# groups of rows that fit that limit, and repeat the header line on every group
# so each one keeps its column names. The "|---|---|" separator line is left out
# of the text that gets embedded, because BERT turns every "-" into its own word
# piece and the separator would use up most of the budget. The payload keeps the
# full group as a Markdown table. A single row longer than the budget would
# still be truncated by the model.
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
tokenizer = embedding_model.tokenizer
token_budget = embedding_model.max_seq_length - 2  # room for [CLS] and [SEP]

payloads = []
embed_texts = []
for chunk in chunks:
    lines = chunk["content"].split("\n")
    if not lines[0].startswith("|"):  # skip non-table chunks, e.g. a truncation note
        continue
    header, separator = lines[0], lines[1]
    rows = [line for line in lines[2:] if line.startswith("|")]

    # Fill each group with rows until the next row would pass the token budget.
    header_tokens = len(tokenizer.tokenize(header))
    groups = []  # each group is a list of (row number, row line)
    current, used = [], header_tokens
    for number, row in enumerate(rows, start=1):
        row_tokens = len(tokenizer.tokenize(row))
        if current and used + row_tokens > token_budget:
            groups.append(current)
            current, used = [], header_tokens
        current.append((number, row))
        used += row_tokens
    if current:
        groups.append(current)

    for group in groups:
        row_lines = [row for _, row in group]
        text = "\n".join([header, separator, *row_lines])
        embed_texts.append("\n".join([header, *row_lines]))
        payloads.append(
            {
                **chunk["metadata"],
                "content": text,
                "word_count": len(text.split()),
                "first_row": group[0][0],
            }
        )
if not payloads:
    raise RuntimeError("Ingest returned no table chunks")
SOURCE_NAME = payloads[0]["source_url"]  # the filename as Ingest labelled it
print(f"Prepared {len(payloads)} row groups for embedding")

# Step 5: Embed each row group
vectors = embedding_model.encode(embed_texts, normalize_embeddings=True)

# Step 6: Connect to Qdrant and create the collection if needed
qdrant = QdrantClient(
    url=os.environ["QDRANT_URL"],
    api_key=os.environ["QDRANT_API_KEY"],
    timeout=60,
)

if not qdrant.collection_exists(COLLECTION):
    qdrant.create_collection(
        collection_name=COLLECTION,
        vectors_config=VectorParams(size=vectors.shape[1], distance=Distance.COSINE),
    )
    # Qdrant Cloud runs in strict mode by default, which rejects filters on
    # unindexed payload keys, and indexes should exist before the data is
    # uploaded.
    qdrant.create_payload_index(
        collection_name=COLLECTION,
        field_name="source_url",
        field_schema=PayloadSchemaType.KEYWORD,
    )

# Step 7: Upsert each row group as a point. The payload is the group text plus
# the chunk metadata (source_url, title, headings_path, section, word_count,
# chunk_index) and first_row. Qdrant payloads accept lists, so headings_path is
# stored as is. Points from an earlier run of this file are deleted first, so a
# shorter CSV on a re-run leaves no stale trailing groups. The id is derived from
# file, chunk and row, so it stays stable between runs and never collides across
# files.
only_this_file = Filter(
    must=[FieldCondition(key="source_url", match=MatchValue(value=SOURCE_NAME))]
)
qdrant.delete(collection_name=COLLECTION, points_selector=only_this_file)

points = [
    PointStruct(
        id=str(
            uuid.uuid5(
                uuid.NAMESPACE_URL,
                f"{payload['source_url']}:{payload['chunk_index']}:{payload['first_row']}",
            )
        ),
        vector=vector.tolist(),
        payload=payload,
    )
    for payload, vector in zip(payloads, vectors)
]

for start in range(0, len(points), 100):
    qdrant.upsert(
        collection_name=COLLECTION, points=points[start : start + 100], wait=True
    )
print(f"Upserted {len(points)} points into Qdrant")

# Step 8: Search with a payload filter applied alongside vector ranking. A CSV
# has no headings, so section is empty on every point and cannot narrow
# anything. source_url is the payload field that tells files apart, which
# matters once the collection holds several.
query_vector = embedding_model.encode(
    ["products suitable for outdoor installation"], normalize_embeddings=True
)[0].tolist()

hits = qdrant.query_points(
    collection_name=COLLECTION,
    query=query_vector,
    query_filter=only_this_file,
    limit=3,
    with_payload=True,
)

for point in hits.points:
    print(f"{point.score:.3f} {point.payload['source_url']} rows from {point.payload['first_row']}")
    print(point.payload["content"])