search.py

"""
Amazon Search - Chocodata Amazon Scraper API

Runnable example. It calls the LIVE API and prints the real JSON response.

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

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

import requests

API = "https://api.chocodata.com/api/v1/amazon/search"
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 errors onto actionable messages instead of a traceback."""
    if r.status_code == 400:
        issues = r.json().get("issues", [])
        detail = "; ".join(f"{'.'.join(map(str, i.get('path', [])))}: {i.get('message')}" for i in issues)
        sys.exit(f"400 invalid_params: {detail or 'check your 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 == 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: Amazon refused this attempt. Retryable, and you were not charged. Try again in ~10s.")
    r.raise_for_status()


def search(query: str, domain: str = "com", sort_by: str = "best_match", start_page: int = 1) -> dict:
    """Search Amazon and return the ranked result cards as structured JSON."""
    params = {
        "api_key": KEY,
        "query": query,
        "domain": domain,
        "sort_by": sort_by,
        "start_page": start_page,
    }
    r = requests.get(API, params=params, timeout=90)
    _check(r)
    return r.json()


if __name__ == "__main__":
    query = sys.argv[1] if len(sys.argv) > 1 else "laptop"
    data = search(query)
    products = data["products"]

    if not products:
        sys.exit(f"Amazon returned a results page with no cards for {query!r}. "
                 "Check the spelling, or try a broader term.")

    print(json.dumps(products[0], indent=2, ensure_ascii=False))
    print()

    organic = [p for p in products if p["organic_position"] is not None]
    sponsored = [p for p in products if p["is_sponsored"]]
    print(f"{len(products)} cards for {query!r} | {len(organic)} organic, {len(sponsored)} sponsored")
    for p in organic[:5]:
        price = f"${p['price']}" if p["price"] is not None else "no price"
        print(f"  #{p['organic_position']:<3} {p['asin']}  {price:>10}  {(p['title'] or '')[:52]}")