#!/usr/bin/env python3 """ BotScope export: every AI answer with its full text, brand mentions and citations. Needs Python 3.8+ and nothing else. Create an API key in BotScope under Settings -> Developers, then in a terminal (Terminal on a Mac): export BOTSCOPE_API_KEY="bsk_..." python3 botscope_export.py On Windows (PowerShell): $env:BOTSCOPE_API_KEY = "bsk_..." py botscope_export.py That downloads every project into ./botscope-export. Options: --project "Northwind" one project (by name or id; repeat for several) --since 2026-10-01 only answers collected on or after this date --out ./northwind-export where to write the files --include-reports also export one-off Quick Scan reports Re-run it whenever you like: answers already downloaded are skipped, so a re-run only fetches what's new and takes seconds to minutes. Files written: answers.csv one row per answer: project, date, AI assistant, prompt, full answer, brands mentioned, number of citations citations.csv one row per cited source: answer, position, title, URL, domain answers.jsonl the complete raw answers, one per line. Keep this file: it is how re-runs know what's already downloaded. The API allows 120 requests a minute per key, and each answer is one request, so the first full download runs at about 6,600 answers an hour. The script paces itself and waits automatically if the limit is reached. """ import argparse import csv import json import os import sys import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone API_BASE = os.environ.get("BOTSCOPE_API_BASE", "https://api.botscope.ai/api/v1") PAGE_SIZE = 200 # the API's maximum MIN_INTERVAL = 60 / 110 # stay just under 120 requests/minute PROVIDER_NAMES = { "openai": "ChatGPT", "anthropic": "Claude", "google": "Gemini", "perplexity": "Perplexity", "copilot": "Copilot", "google_ai_overview": "Google AI Overview", } _last_request = 0.0 def collected_via(answer): """'web' (the public web app, as a logged-out user sees it), 'API' or 'Google search'.""" if answer.get("provider") == "google_ai_overview": return "Google search" source = answer.get("source") or "" return "web" if source.endswith("web") else "API" if source == "api" else source def fail(message): sys.exit(f"\nError: {message}") def api_get(path, params=None): """GET one API path, paced under the rate limit, retrying 429s and server errors.""" global _last_request url = API_BASE + path if params: url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) req = urllib.request.Request(url, headers={ "Authorization": f"Bearer {API_KEY}", "Accept": "application/json", "User-Agent": "botscope-export/1.0", }) for attempt in range(8): wait = MIN_INTERVAL - (time.monotonic() - _last_request) if wait > 0: time.sleep(wait) _last_request = time.monotonic() try: with urllib.request.urlopen(req, timeout=60) as resp: return json.load(resp) except urllib.error.HTTPError as e: if e.code == 429: retry_after = int(e.headers.get("Retry-After") or 30) print(f" rate limit reached, waiting {retry_after}s...", flush=True) time.sleep(retry_after) continue if e.code in (401, 403): fail("the API key was refused. Check BOTSCOPE_API_KEY, and that the " "workspace is on an active paid plan (API access needs one).") if e.code == 404: return None if e.code < 500: fail(f"{e.code} from {path}: {e.read().decode(errors='replace')[:300]}") except (urllib.error.URLError, TimeoutError, ConnectionError): pass time.sleep(min(2 ** attempt, 60)) # network blip or 5xx: back off and retry fail(f"giving up on {path} after repeated errors. Re-run to resume.") def parse_ts(value): return datetime.fromisoformat(value.replace("Z", "+00:00")) def list_answers(project_id, since): """Yield the answer index for a project, newest first (no text yet).""" seen = set() before = None while True: page = api_get(f"/projects/{project_id}/responses", {"limit": PAGE_SIZE, "before": before}) rows = (page or {}).get("data", []) new = [r for r in rows if r["id"] not in seen] for r in new: if since and parse_ts(r["createdAt"]) < since: return seen.add(r["id"]) yield r next_before = (page or {}).get("pagination", {}).get("nextBefore") if not next_before or not new: return # The cursor is "older than this timestamp". Step 1ms back toward the # newest so answers saved in the same millisecond as the page's last one # aren't skipped; the overlap is de-duplicated above. before = (parse_ts(next_before) + timedelta(milliseconds=1)).isoformat().replace("+00:00", "Z") def load_store(path): answers = {} if os.path.exists(path): with open(path, encoding="utf-8") as f: for line in f: line = line.strip() if line: a = json.loads(line) answers[a["id"]] = a return answers def write_csvs(out_dir, answers): rows = sorted(answers.values(), key=lambda a: (a.get("projectName", ""), a.get("createdAt", ""))) # utf-8-sig so Excel opens the files with accents and symbols intact. with open(os.path.join(out_dir, "answers.csv"), "w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow(["answer_id", "project", "scan_id", "collected_at_utc", "ai_assistant", "model", "layer", "collected_via", "prompt", "answer", "own_brand_mentioned", "brands_mentioned", "citation_count"]) for a in rows: brands = [] for d in a.get("brandDetections") or []: if d.get("brandName") and d["brandName"] not in brands: brands.append(d["brandName"]) own = any(d.get("isOwnBrand") for d in a.get("brandDetections") or []) w.writerow([a["id"], a.get("projectName", ""), a.get("scanId") or "", a.get("createdAt", ""), PROVIDER_NAMES.get(a.get("provider"), a.get("provider")), a.get("model") or "", a.get("layer") or "", collected_via(a), a.get("prompt") or "", a.get("response") or "", "yes" if own else "no", "; ".join(brands), len(a.get("citations") or [])]) with open(os.path.join(out_dir, "citations.csv"), "w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow(["answer_id", "project", "collected_at_utc", "ai_assistant", "prompt", "position", "title", "url", "domain", "source", "kind"]) for a in rows: for c in a.get("citations") or []: w.writerow([a["id"], a.get("projectName", ""), a.get("createdAt", ""), PROVIDER_NAMES.get(a.get("provider"), a.get("provider")), a.get("prompt") or "", c.get("position", ""), c.get("title", ""), c.get("url", ""), c.get("domain", ""), c.get("source", ""), c.get("kind", "")]) def main(): global API_KEY parser = argparse.ArgumentParser(description="Export BotScope answers and citations to CSV.") parser.add_argument("--project", action="append", help="project name or id (repeatable)") parser.add_argument("--since", help="only answers collected on or after this date (YYYY-MM-DD)") parser.add_argument("--out", default="botscope-export", help="output folder (default ./botscope-export)") parser.add_argument("--include-reports", action="store_true", help="also export one-off Quick Scan reports") args = parser.parse_args() API_KEY = os.environ.get("BOTSCOPE_API_KEY", "").strip() if not API_KEY: fail('set your API key first: export BOTSCOPE_API_KEY="bsk_..."') since = None if args.since: try: since = datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=timezone.utc) except ValueError: fail("--since must look like 2026-10-01") projects = api_get("/projects", {"type": "all" if args.include_reports else "project"})["data"] if args.project: wanted = {p.lower() for p in args.project} chosen = [p for p in projects if p["id"] in wanted or p["name"].lower() in wanted] if len(chosen) < len(wanted): names = "\n ".join(p["name"] for p in projects) fail(f"couldn't match every --project. Projects on this key:\n {names}") projects = chosen os.makedirs(args.out, exist_ok=True) store_path = os.path.join(args.out, "answers.jsonl") answers = load_store(store_path) print(f"{len(answers)} answers already downloaded in {args.out}/") fetched = 0 with open(store_path, "a", encoding="utf-8") as store: for project in projects: todo = [r for r in list_answers(project["id"], since) if r["id"] not in answers] if not todo: print(f"{project['name']}: up to date") continue print(f"{project['name']}: {len(todo)} new answers (about {max(1, round(len(todo) / 110))} min)") for i, row in enumerate(todo, 1): detail = api_get(f"/projects/{project['id']}/responses/{row['id']}") if detail is None: continue # deleted since it was listed answer = detail["data"] answer["projectId"] = project["id"] answer["projectName"] = project["name"] store.write(json.dumps(answer, ensure_ascii=False) + "\n") store.flush() # an interrupted run keeps everything fetched so far answers[answer["id"]] = answer fetched += 1 if i % 50 == 0 or i == len(todo): print(f" {i}/{len(todo)}", flush=True) write_csvs(args.out, answers) cites = sum(len(a.get("citations") or []) for a in answers.values()) print(f"\nDone: {fetched} new, {len(answers)} answers and {cites} citations in {args.out}/") if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit("\nStopped. Everything downloaded so far is saved; re-run to carry on.")