crawl_url_list.py

"""
Crawl a mixed list of URLs concurrently and store the results - Chocodata Web Scraper API

The real use case for an any-URL endpoint: you have a list of URLs, they are not all
the same kind of thing, and you want the content of each without writing a fetcher
per site. Pages come back as html, API endpoints come back as parsed JSON, and you
did not have to know which was which in advance.

Results go to a local SQLite dataset so you can re-run it and diff. Export to CSV:

    sqlite3 -header -csv crawl.db "select url, content_type, chars, title from pages;" > crawl.csv

Usage:
    pip install requests
    export CHOCODATA_API_KEY="your_key"      # free: 1,000 requests, one-time
    python web_scraper_api_codes/crawl_url_list.py
    python web_scraper_api_codes/crawl_url_list.py urls.txt      # one URL per line

Docs: https://chocodata.com/docs
"""
import json
import os
import re
import sqlite3
import sys
import time
from concurrent.futures import ThreadPoolExecutor

import requests

API = "https://api.chocodata.com/api/v1/universal"
KEY = os.environ.get("CHOCODATA_API_KEY")
DB = "crawl.db"

# Keep the pool at or under your plan's concurrency (Free 10, Vibe 30, Pro 50).
# The 120 requests/60s rate limit binds before the concurrency cap does.
WORKERS = 5

# A deliberately mixed list: two plain pages, a JSON API, and one target that a
# stock HTTP client cannot fetch at all. You do not have to know which is which.
DEFAULT_URLS = [
    "https://example.com",
    "https://api.github.com/repos/python/cpython",
    "https://en.wikipedia.org/wiki/Web_scraping",
    "https://quotes.toscrape.com/",
    "https://www.ebay.com/sch/i.html?_nkw=laptop",
]

if not KEY:
    sys.exit("Set CHOCODATA_API_KEY first. Free key (1,000 requests, one-time): https://chocodata.com")


def _reason(r) -> str:
    """Documented status -> actionable message. Never a traceback, never a silent skip."""
    return {
        400: "invalid_params: check the url is absolute and well formed",
        401: "INVALID_API_KEY: key missing or not recognised, get one at https://chocodata.com",
        402: "INSUFFICIENT_CREDITS: balance exhausted, top up at https://chocodata.com/pricing",
        404: "item_not_found: the URL returned 404 (not retryable, not charged)",
        429: "RATE_LIMITED: back off, you are over 120 req/60s or your concurrency cap",
        502: "target_unreachable: the target refused every attempt (retryable, not charged)",
    }.get(r.status_code, f"unexpected HTTP {r.status_code}: {r.text[:80]}")


def fetch_one(url: str) -> dict:
    t0 = time.time()
    try:
        r = requests.get(API, params={"api_key": KEY, "url": url}, timeout=90)
    except requests.exceptions.RequestException as e:
        return {"url": url, "ok": False, "error": f"{type(e).__name__}", "secs": time.time() - t0}
    if r.status_code != 200:
        # 401 and 402 are fatal for the whole run: stop rather than hammer the API.
        return {"url": url, "ok": False, "error": _reason(r), "secs": time.time() - t0,
                "fatal": r.status_code in (401, 402)}
    d = r.json()
    body = d.get("html") or d.get("text") or ""
    payload = json.dumps(d["json"]) if d["content_type"] == "json" else body
    title = re.search(r"<title[^>]*>(.*?)</title>", body, re.S)
    return {"url": url, "ok": True, "content_type": d["content_type"], "chars": len(payload),
            "title": re.sub(r"\s+", " ", title.group(1)).strip()[:120] if title else None,
            "secs": time.time() - t0}


def main(urls) -> int:
    con = sqlite3.connect(DB)
    con.execute("""create table if not exists pages (
        url text primary key, content_type text, chars int, title text,
        secs real, fetched_at text)""")

    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        results = list(pool.map(fetch_one, urls))

    ok = [r for r in results if r["ok"]]
    bad = [r for r in results if not r["ok"]]

    for r in ok:
        con.execute("insert or replace into pages values (?,?,?,?,?,?)",
                    (r["url"], r["content_type"], r["chars"], r["title"], round(r["secs"], 2),
                     time.strftime("%Y-%m-%d %H:%M:%S")))
    con.commit()

    print(f"{'content_type':13} {'chars':>9} {'secs':>6}  url")
    print("-" * 78)
    for r in sorted(ok, key=lambda x: -x["chars"]):
        print(f"{r['content_type']:13} {r['chars']:>9,} {r['secs']:>6.1f}  {r['url'][:44]}")
    for r in bad:
        print(f"{'FAILED':13} {'-':>9} {r['secs']:>6.1f}  {r['url'][:44]}\n              {r['error']}")

    print()
    print(f"{len(ok)}/{len(urls)} fetched | {sum(r['chars'] for r in ok):,} chars | "
          f"{len(ok) * 5} credits used | stored in {DB}")
    if any(r.get("fatal") for r in bad):
        print("Stopped early: an auth or billing error affects every request, not just this one.")
        return 2
    return 1 if bad else 0


if __name__ == "__main__":
    if len(sys.argv) > 1:
        with open(sys.argv[1], encoding="utf-8") as f:
            targets = [l.strip() for l in f if l.strip() and not l.startswith("#")]
    else:
        targets = DEFAULT_URLS
    sys.exit(main(targets))