Skip to content

Instantly share code, notes, and snippets.

@quantumproxies
Created August 21, 2026 12:42
Show Gist options
  • Select an option

  • Save quantumproxies/2ee7db1870eda556090e8024dfbf1675 to your computer and use it in GitHub Desktop.

Select an option

Save quantumproxies/2ee7db1870eda556090e8024dfbf1675 to your computer and use it in GitHub Desktop.
Crawl a site to Markdown files with polite polling — QuanticData Crawl API in Python https://quanticdata.io/blog/how-to-web-crawl-python/
"""Crawl a site to one Markdown file per page, polling with backoff.
pip install requests
export QD_API_KEY=qd_live_...
python3 qd_crawl.py https://quanticdata.io 40 /blog/
$0.0003 a page, and the unfetched share of the budget is refunded when the job settles.
Cheaper alternative for a site you can filter first: POST /v1/map (flat $0.0005, every
URL the sitemaps know) then POST /v1/batch on exactly the URLs you want.
https://quanticdata.io/crawl-map/ · https://quanticdata.io/blog/how-to-web-crawl-python/
"""
import os
import pathlib
import re
import sys
import time
import requests
seed = sys.argv[1] if len(sys.argv) > 1 else "https://quanticdata.io"
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 25
include = sys.argv[3] if len(sys.argv) > 3 else None
BASE = "https://api.quanticdata.io/v1"
H = {"Authorization": f"Bearer {os.environ['QD_API_KEY']}"}
body = {"url": seed, "limit": limit, "depth": 3, "format": "markdown", "contentMode": "article"}
if include:
body["include"] = [include]
job = requests.post(f"{BASE}/crawl", json=body, headers=H, timeout=120).json()
if job.get("type") == "error":
sys.exit(job["message"])
job_id = job["payload"]["id"]
print(f"job {job_id}", file=sys.stderr)
delay = 2.0
while True:
time.sleep(delay)
delay = min(delay * 1.4, 15.0) # hammering a status endpoint achieves nothing
status = requests.get(f"{BASE}/crawl/{job_id}", headers=H, timeout=60).json()["payload"]
print(f" {status['status']}: {status['pagesCrawled']} crawled, "
f"{status['pagesQueued']} queued", file=sys.stderr)
if status["status"] in ("completed", "failed", "cancelled"):
break
out = pathlib.Path("pages")
out.mkdir(exist_ok=True)
written = 0
for page in status.get("pages") or []:
if page.get("error") or not page.get("content"):
continue
name = re.sub(r"[^a-zA-Z0-9._-]+", "-", re.sub(r"^https?://", "", page["url"])).strip("-")
(out / f"{name[:120]}.md").write_text(
f"---\nurl: {page['url']}\ntitle: {page.get('title')}\n---\n\n{page['content']}",
encoding="utf-8")
written += 1
print(f"{written} pages -> pages/", file=sys.stderr)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment