#!/usr/bin/env python3 """Convert JSON API pages to CSV on your side. Only Pick6 has a native format=csv. Independent service, not affiliated with or endorsed by DraftKings. No wagering. Run: python3 export-csv.py --resource dfs/players --sport NBA --max-pages 3 --out rows.csv python3 export-csv.py --resource stats/player-games --sport NBA --season 2027 --max-pages 2 --out logs.csv Resources: events, odds, history, dfs/players, dfs/slates, dfs/contests, stats/player-games. Technical access is not a redistribution license: keep exports to your own use unless you hold rights. """ import argparse import hashlib import json import os import sys import time from datetime import datetime, timezone from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen import csv API_KEY = os.environ.get("DRAFTKINGS_API_KEY") BASE = os.environ.get("DRAFTKINGS_API_BASE", "https://draftkingsapi.com").rstrip("/") CACHE_DIR = Path(os.environ.get("DRAFTKINGS_CACHE_DIR", ".draftkings-cache")) # Collection targets five minutes. Polling faster spends quota without newer data. CACHE_TTL_SECONDS = 300 class ApiError(Exception): def __init__(self, status, message): super().__init__(f"HTTP {status}: {message}") self.status = status class Client: """Bounded client: a request budget, a timeout, a file cache and limited retries.""" def __init__(self, max_requests, use_cache=True): if not API_KEY: sys.exit("Set DRAFTKINGS_API_KEY in your environment. Create a key at https://draftkingsapi.com/dashboard") self.max_requests = max_requests self.requests = 0 self.use_cache = use_cache def get(self, path, **params): query = {k: v for k, v in params.items() if v is not None} url = f"{BASE}{path}?{urlencode(query)}" if query else f"{BASE}{path}" cached = self._read_cache(url) if cached is not None: return cached body = self._fetch(url) self._write_cache(url, body) return body def _fetch(self, url): for attempt in range(3): if self.requests >= self.max_requests: raise ApiError(0, f"Local budget of {self.max_requests} requests reached. Raise it deliberately.") self.requests += 1 request = Request(url, headers={"Authorization": f"Bearer {API_KEY}", "Accept": "application/json"}) try: with urlopen(request, timeout=20) as response: return json.load(response) except HTTPError as error: message = _error_message(error) can_retry = self.requests < self.max_requests if error.code == 429 and attempt == 0 and can_retry: # Minute limits reset at the next UTC minute. A second 429 usually means the monthly allowance. time.sleep(61 - datetime.now(timezone.utc).second) continue if error.code >= 500 and attempt < 2 and can_retry: time.sleep(2 ** attempt) continue raise ApiError(error.code, message) from None except (URLError, TimeoutError) as error: if attempt < 2 and self.requests < self.max_requests: time.sleep(2 ** attempt) continue raise ApiError(0, f"Network failure: {error}") from None raise ApiError(429, "Account request limit reached. Check the allowance in your dashboard.") def _cache_path(self, url): return CACHE_DIR / (hashlib.sha256(url.encode()).hexdigest() + ".json") def _read_cache(self, url): path = self._cache_path(url) if not self.use_cache or not path.exists() or time.time() - path.stat().st_mtime > CACHE_TTL_SECONDS: return None return json.loads(path.read_text()) def _write_cache(self, url, body): if self.use_cache: CACHE_DIR.mkdir(exist_ok=True) self._cache_path(url).write_text(json.dumps(body)) def _error_message(error): try: return json.load(error)["error"]["message"] except Exception: return error.reason def flatten(row, prefix=""): flat = {} for key, value in row.items(): name = f"{prefix}{key}" if isinstance(value, dict) and prefix == "": flat.update(flatten(value, f"{name}.")) elif isinstance(value, (dict, list)): flat[name] = json.dumps(value, separators=(",", ":")) else: flat[name] = value return flat def safe(value): # Spreadsheets execute cells that start with these characters, so quote them. if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t", "\r"): return "'" + value return value def main(): parser = argparse.ArgumentParser() parser.add_argument("--resource", required=True, choices=["events", "odds", "history", "dfs/players", "dfs/slates", "dfs/contests", "stats/player-games"]) parser.add_argument("--sport") parser.add_argument("--season", type=int) parser.add_argument("--max-pages", type=int, default=3) parser.add_argument("--out", default="rows.csv") args = parser.parse_args() client = Client(max_requests=args.max_pages) path, cursor_paged = f"/api/v1/{args.resource}", args.resource.startswith("stats/") if cursor_paged and not (args.sport and args.season): sys.exit("Statistics require --sport and --season.") rows, offset, after = [], 0, None for _ in range(args.max_pages): if cursor_paged: page = client.get(path, sport=args.sport, season=args.season, limit=200, after=after) else: page = client.get(path, sport=args.sport, limit=100, offset=offset) rows.extend(flatten(row) for row in page["data"]) if not page["pagination"]["has_more"]: break after = page["pagination"].get("next") offset += 100 if not rows: sys.exit("No rows returned. Nothing written.") columns = sorted({key for row in rows for key in row}) with open(args.out, "w", newline="", encoding="utf-8") as file: writer = csv.DictWriter(file, fieldnames=columns) writer.writeheader() writer.writerows({key: safe(value) for key, value in row.items()} for row in rows) print(f"Wrote {len(rows)} rows and {len(columns)} columns to {args.out} using {client.requests} requests.") if __name__ == "__main__": try: main() except ApiError as error: sys.exit(str(error))