fetch_json.py

"""
Fetch a JSON API through the scraper - Chocodata Web Scraper API

parse=auto detects a JSON body and returns it parsed under `json`, without you
having to say so. Useful when you are crawling a mixed list of URLs and do not
know in advance which are pages and which are API endpoints.

    pip install requests
    export CHOCODATA_API_KEY="your_key"      # free: 1,000 requests, one-time
    python web_scraper_api_codes/fetch_json.py

Docs: https://chocodata.com/docs
"""
import json
import os
import sys

import requests

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

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


def _check(r) -> None:
    """Map the API's documented errors onto actionable messages instead of a traceback."""
    if r.status_code == 400:
        issues = r.json().get("issues", [])
        detail = "; ".join(f"{'.'.join(str(p) for p in i.get('path', []))}: {i.get('message')}" for i in issues)
        sys.exit(f"400 invalid_params: {detail or r.text[:160]}. Fix the query string.")
    if r.status_code == 401:
        sys.exit("401 INVALID_API_KEY: key missing or not recognised. Get one: https://chocodata.com")
    if r.status_code == 402:
        sys.exit("402 INSUFFICIENT_CREDITS: balance exhausted. Top up or upgrade: https://chocodata.com/pricing")
    if r.status_code == 404:
        sys.exit("404 item_not_found: the target returned 404 for this URL. "
                 "Check it exists. Retrying will not help, and you were not charged.")
    if r.status_code == 429:
        sys.exit("429 RATE_LIMITED: over 120 requests/60s or your plan's concurrency. Back off and retry.")
    if r.status_code == 502:
        sys.exit("502 target_unreachable: the target refused every attempt. Retryable, and you were not charged.")
    r.raise_for_status()


def fetch(url: str, parse: str = "auto") -> dict:
    r = requests.get(API, params={"api_key": KEY, "url": url, "parse": parse}, timeout=90)
    _check(r)
    return r.json()


if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else "https://api.github.com/repos/python/cpython/languages"
    data = fetch(target)

    # parse=auto returns content_type json only when the body really parsed as JSON.
    # If it did not, you get html instead: always branch on content_type rather than
    # assuming, because parse=json falls back to html rather than erroring.
    if data["content_type"] != "json":
        print(f"content_type is {data['content_type']}, not json: this URL did not return a JSON body.")
        sys.exit(0)

    payload = data["json"]
    print(f"url          : {data['url']}")
    print(f"content_type : {data['content_type']}   <- auto-detected, we never said parse=json")
    print(f"json chars   : {len(json.dumps(payload)):,}")
    print()
    # It came back as a real dict, not a string you have to json.loads yourself.
    if isinstance(payload, dict):
        for k, v in list(payload.items())[:10]:
            print(f"  {str(k):20} {v:>12,}" if isinstance(v, int) else f"  {str(k):20} {v}")
        print()
        print(f"OK: parsed {len(payload)} top-level fields from {data['url']}")
    else:
        print(json.dumps(payload, indent=2)[:800])
        print()
        print(f"OK: parsed a JSON {type(payload).__name__} from {data['url']}")