#!/usr/bin/env python3 """ Crypto Card Geography Index: collection runner. Runs a fixed prompt set across ChatGPT, Claude, Perplexity and Google AI Overviews, from each of nine markets, N times each. The prompt text is byte-identical in every market. Only the location changes: web_search_country_iso_code on the LLM endpoints, location_code on the SERP endpoint. That makes geography the single variable. Credentials come from ../.env. Usage: python3 runner.py --runs 3 --out raw.jsonl --dump-raw raw_full/ python3 runner.py --runs 1 --markets NG,IN,US --limit 2 --out pilot.jsonl """ import argparse, base64, json, os, re, sys, threading, time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone import requests BASE = "https://api.dataforseo.com" TRANSIENT = {40101, 40102, 40103, 50000} PROMPT_LIMIT, KEYWORD_LIMIT = 500, 700 ENGINES = { "chatgpt": {"path": "/v3/ai_optimization/chat_gpt/llm_responses/live", "model": "gpt-5-mini"}, "claude": {"path": "/v3/ai_optimization/claude/llm_responses/live", "model": "claude-haiku-4-5-20251001"}, "perplexity": {"path": "/v3/ai_optimization/perplexity/llm_responses/live","model": "sonar"}, } AIO_PATH = "/v3/serp/google/organic/live/advanced" # Measured in edition one. The cap counts real returned cost, not these. UNIT = {"chatgpt": 0.0105, "claude": 0.0238, "perplexity": 0.0062, "google_aio": 0.0040} MD_LINK = re.compile(r"\[([^\]]*)\]\(([^)]*)\)") def strip_md_links(md): return MD_LINK.sub(lambda m: m.group(1), md) def load_env(path="../.env"): for p in (path, ".env"): if os.path.exists(p): for line in open(p): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) return def auth(): lo = os.environ.get("DFS_LOGIN") or os.environ.get("DATAFORSEO_LOGIN") pw = os.environ.get("DFS_PASSWORD") or os.environ.get("DATAFORSEO_PASSWORD") if not lo or not pw: sys.exit("No credentials in .env") t = base64.b64encode(f"{lo}:{pw}".encode()).decode() return {"Authorization": f"Basic {t}", "Content-Type": "application/json"} def post(path, payload, headers, retries=3): for attempt in range(retries): try: r = requests.post(BASE + path, headers=headers, data=json.dumps(payload), timeout=200) if r.status_code == 200: body = r.json() tasks = body.get("tasks") or [] # DataForSEO returns HTTP 200 with a task-level transient code. if tasks and tasks[0].get("status_code") in TRANSIENT and attempt < retries - 1: time.sleep(5 * (attempt + 1)); continue return body if r.status_code in (429, 500, 502, 503): time.sleep(4 * (attempt + 1)); continue return {"_http_error": r.status_code, "_body": r.text[:400]} except requests.RequestException as e: if attempt == retries - 1: return {"_exception": f"{type(e).__name__}: {e}"} time.sleep(4 * (attempt + 1)) return {"_error": "exhausted retries"} def call(engine, prompt, iso, market, headers): if engine == "google_aio": return post(AIO_PATH, [{"keyword": prompt["text"][:KEYWORD_LIMIT], "location_code": market["aio_location_code"], "language_code": "en", "load_async_ai_overview": True}], headers) cfg = ENGINES[engine] return post(cfg["path"], [{"user_prompt": prompt["text"][:PROMPT_LIMIT], "model_name": cfg["model"], "web_search": True, "web_search_country_iso_code": iso}], headers) def extract(engine, raw): out = {"text": "", "sources": [], "fan_out": [], "cost": 0.0, "error": None, "web_search_ran": None, "input_tokens": None, "output_tokens": None} if not isinstance(raw, dict) or "tasks" not in raw: out["error"] = json.dumps(raw)[:300]; return out out["cost"] = raw.get("cost", 0.0) or 0.0 tasks = raw.get("tasks") or [] if not tasks: out["error"] = "no tasks"; return out task = tasks[0] if task.get("status_code") != 20000: out["error"] = f"{task.get('status_code')}: {task.get('status_message')}"; return out results = task.get("result") or [] if not results: out["error"] = "no result"; return out if engine == "google_aio": parts, srcs = [], [] for block in results[0].get("items") or []: if block.get("type") != "ai_overview": continue for ref in block.get("references") or []: if ref.get("url"): srcs.append({"url": ref["url"], "title": ref.get("title", "")}) for el in block.get("items") or []: for ref in el.get("references") or []: if ref.get("url"): srcs.append({"url": ref["url"], "title": ref.get("title", "")}) # Element text is plain prose. The block markdown embeds citation # URLs, which would score vendor domains as product mentions. t, ti = (el.get("text") or "").strip(), (el.get("title") or "").strip() if ti: parts.append(ti if ti.endswith((".", ":")) else ti + ".") if t: parts.append(t) if not parts and block.get("markdown"): parts.append(strip_md_links(block["markdown"]).strip()) out["text"] = "\n".join(parts).strip() seen, uniq = set(), [] for s in srcs: if s["url"] not in seen: seen.add(s["url"]); uniq.append(s) out["sources"] = uniq if not out["text"]: out["error"] = "no ai_overview present" return out res = results[0] out["fan_out"] = res.get("fan_out_queries") or [] out["web_search_ran"] = bool(res.get("web_search")) out["input_tokens"], out["output_tokens"] = res.get("input_tokens"), res.get("output_tokens") seen = set() for item in res.get("items") or []: if item.get("type") != "message": continue for sec in item.get("sections") or []: if sec.get("text"): out["text"] += sec["text"] + "\n" # Gemini-style repetition also occurs here; keep sources unique. for ann in sec.get("annotations") or []: u = ann.get("url") if u and u not in seen: seen.add(u); out["sources"].append({"url": u, "title": ann.get("title", "")}) out["text"] = out["text"].strip() if not out["text"]: out["error"] = "empty text" return out def load_done(path): done = set() if os.path.exists(path): for line in open(path): try: r = json.loads(line) if not r.get("error"): done.add((r["prompt_id"], r["market"], r["engine"], r["run"])) except Exception: continue return done def main(): ap = argparse.ArgumentParser() ap.add_argument("--prompts", default="prompts.json") ap.add_argument("--markets-file", default="markets.json") ap.add_argument("--markets", default="") ap.add_argument("--runs", type=int, default=3) ap.add_argument("--limit", type=int, default=0) ap.add_argument("--out", default="raw.jsonl") ap.add_argument("--dump-raw", default="") ap.add_argument("--workers", type=int, default=6) ap.add_argument("--cap", type=float, default=20.0) ap.add_argument("--yes", action="store_true") ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() load_env() prompts = json.load(open(args.prompts)) if args.limit: prompts = prompts[:args.limit] markets = json.load(open(args.markets_file)) if args.markets: keep = {m.strip() for m in args.markets.split(",")} markets = {k: v for k, v in markets.items() if k in keep} jobs, done = [], load_done(args.out) for iso, mk in markets.items(): engines = ["chatgpt", "perplexity", "google_aio"] + (["claude"] if mk.get("claude") else []) for p in prompts: for e in engines: for run in range(1, args.runs + 1): if (p["id"], iso, e, run) not in done: jobs.append((p, iso, mk, e, run)) est = sum(UNIT.get(e, 0.02) for _, _, _, e, _ in jobs) print(f"prompts:{len(prompts)} markets:{len(markets)} runs:{args.runs}") print(f"done:{len(done)} remaining:{len(jobs)} est ${est:.2f} cap ${args.cap:.2f}") if args.dry_run or not jobs: return if not args.yes and input("proceed? [y/N] ").strip().lower() != "y": return if args.dump_raw: os.makedirs(args.dump_raw, exist_ok=True) headers = auth() lock = threading.Lock() state = {"spent": 0.0, "n": 0, "aborted": False} fh = open(args.out, "a") def work(job): p, iso, mk, e, run = job if state["aborted"]: return None t0 = time.time() raw = call(e, p, iso, mk, headers) if args.dump_raw: fn = os.path.join(args.dump_raw, f"{p['id']}__{iso}__{e}__run{run}.json") json.dump(raw, open(fn, "w"), indent=2) parsed = extract(e, raw) return {"prompt_id": p["id"], "scenario": p["scenario"], "stage": p["stage"], "topic": p.get("topic", ""), "prompt": p["text"], "market": iso, "market_name": mk["name"], "adoption_rank": mk.get("adoption_rank"), "engine": e, "model": "google_ai_overviews" if e == "google_aio" else ENGINES[e]["model"], "run": run, "ts": datetime.now(timezone.utc).isoformat(), "elapsed_s": round(time.time() - t0, 2), "text": parsed["text"], "sources": parsed["sources"], "fan_out": parsed["fan_out"], "web_search_ran": parsed["web_search_ran"], "input_tokens": parsed["input_tokens"], "output_tokens": parsed["output_tokens"], "cost": parsed["cost"], "error": parsed["error"]} with ThreadPoolExecutor(max_workers=min(args.workers, 6)) as pool: futs = {pool.submit(work, j): j for j in jobs} for f in as_completed(futs): rec = f.result() if rec is None: continue with lock: fh.write(json.dumps(rec) + "\n"); fh.flush() state["spent"] += rec["cost"]; state["n"] += 1 flag = "ERR " if rec["error"] else "ok " print(f"[{state['n']}/{len(jobs)}] {flag}{rec['market']} {rec['engine']:<11} " f"{rec['prompt_id']:<7} run{rec['run']} {rec['elapsed_s']:>6.1f}s ${state['spent']:.4f}" + (f" <- {rec['error'][:50]}" if rec["error"] else "")) if state["spent"] > args.cap and not state["aborted"]: state["aborted"] = True print(f"\n!!! COST CAP HIT ${state['spent']:.4f} > ${args.cap:.2f}. Aborting. !!!\n") for g in futs: g.cancel() fh.close() print(f"\ndone. {state['n']} calls, ${state['spent']:.4f}, -> {args.out}") if state["aborted"]: sys.exit("ABORTED ON COST CAP") if __name__ == "__main__": main()