Export every answer to a warehouse

Read every answer, mention, and citation 1 week at a time, and save them as 3 CSV files that you can load into a data warehouse.

The API gives the raw data as 3 lists. Each list becomes 1 table:

Endpoint	Each row is
`/v1/answers`	1 AI answer, with its text
`/v1/mentions`	1 brand that an answer ranks or names
`/v1/citations`	1 source that an answer cites

Each mention and citation row has the answer_id of its answer, which is the id of the answer row. Use it to join the 3 tables. This script reads the last 4 weeks, 1 week at a time, and writes 3 CSV files. It sends include_archived=true, so the copy also has the answers of archived prompts:

import csv
import os
from datetime import datetime, timedelta, timezone

import requests

API = "https://api.lumirank.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['LUMIRANK_API_KEY']}"}


def walk(path, params):
    offset = 0
    while True:
        resp = requests.get(f"{API}{path}", headers=HEADERS,
                            params={**params, "limit": 100, "offset": offset}, timeout=60)
        resp.raise_for_status()
        body = resp.json()
        yield from body["data"]
        if not body["meta"]["has_more"]:
            return
        offset += 100


today = datetime.now(timezone.utc).date()
weeks = [(today - timedelta(days=d + 6), today - timedelta(days=d)) for d in range(0, 28, 7)]
for table in ("answers", "mentions", "citations"):
    with open(f"{table}.csv", "w", newline="") as out:
        writer = None
        for start, end in weeks:
            params = {"start_date": start.isoformat(), "end_date": end.isoformat(), "include_archived": "true"}
            for row in walk(f"/{table}", params):
                if writer is None:
                    writer = csv.DictWriter(out, fieldnames=list(row))
                    writer.writeheader()
                writer.writerow(row)
    print("wrote", table)

Notes:

The lists have completed answers only. Without include_archived=true, they leave out the answers of archived prompts, as the Results page in the app does.
The dates filter on answered_at, so each answer falls in exactly 1 week. No answer appears twice, and none is missed.
For a job that runs every day, fetch only yesterday's data, and use answer_id as the key when you load it. See Data freshness.
Each request returns up to 100 rows, and a brand can make 200 requests a minute, so you can read up to 20,000 rows a minute. See Rate limits.