example.py

"""
Amazon Scraper — Scrapeless Scraping API (Python example)

Docs:  https://apidocs.scrapeless.com/doc-857373
Token: https://app.scrapeless.com/passport/login?redirect=/quick-start

The Amazon actor supports four scrape types selected via `input.type`:
  product | seller | keywords | rufus

Run:
    export SCRAPELESS_API_TOKEN="your_api_token"
    pip install requests
    python example.py            # defaults to the "product" type
    python example.py keywords   # or pass a type: product | seller | keywords | rufus
"""

import os
import sys
import json
import requests

API_URL = "https://api.scrapeless.com/api/v1/scraper/request"
API_TOKEN = os.environ.get("SCRAPELESS_API_TOKEN", "YOUR_API_TOKEN")

# Ready-to-use input payloads for each scrape type.
SAMPLE_INPUTS = {
    "product": {
        "type": "product",
        "url": "https://www.amazon.com/dp/B0BQXHK363",
        "zip_code": "",
    },
    "seller": {
        "type": "seller",
        "url": "https://www.amazon.com/sp?seller=A2XZ7JICGUQ1CX",
        "zip_code": "",
    },
    "keywords": {
        "type": "keywords",
        "keywords": "Iphone+14+Pro+512GB",
        "page": "1",
        "domain": "com",
        "zip_code": "",
    },
    "rufus": {
        "type": "rufus",
        "keywords": "macbook",
        "domain": "www.amazon.es",
        "page": "1",
    },
}


def scrape(scrape_type: str):
    if scrape_type not in SAMPLE_INPUTS:
        raise SystemExit(f"Unknown type '{scrape_type}'. Choose one of: {', '.join(SAMPLE_INPUTS)}")

    payload = {"actor": "scraper.amazon", "input": SAMPLE_INPUTS[scrape_type]}
    headers = {"Content-Type": "application/json", "x-api-token": API_TOKEN}

    response = requests.post(API_URL, headers=headers, json=payload, timeout=180)

    # The Amazon actor distinguishes scenarios by HTTP status code.
    if response.status_code == 200:
        # Synchronous success: the body is the scraped data (shape depends on type).
        data = response.json()
        print(f"[200] Success — '{scrape_type}' data received.")
        summarize(scrape_type, data)
        return data

    if response.status_code == 201:
        # Task accepted but still running. Retrieve it later by task id
        # (async retrieval / webhook — see the official documentation).
        body = response.json()
        print(f"[201] Task in progress — message: {body.get('message')}, taskId: {body.get('taskId')}")
        print("      Fetch the result later using the task id (see docs).")
        return body

    if response.status_code == 400:
        # Scraping failed — inspect the error code and message.
        body = response.json()
        print(f"[400] Bad request — code: {body.get('code')}, message: {body.get('message')}")
        return body

    # Any other status: surface it for debugging.
    print(f"[{response.status_code}] Unexpected response:\n{response.text}")
    response.raise_for_status()


def summarize(scrape_type: str, data):
    """Print a short, type-specific summary of a 200 response."""
    if scrape_type == "product":
        print(f"  ASIN:  {data.get('asin')}")
        print(f"  Brand: {data.get('brand')}")
        print(f"  Price: {data.get('final_price')}  ({data.get('availability')})")
    elif scrape_type == "keywords":
        organic = (data.get("result") or {}).get("organic") or []
        print(f"  keyword: {data.get('keyword')} page: {data.get('page')} — {len(organic)} organic results")
        for item in organic[:3]:
            print(f"    - {item.get('title')} :: {item.get('price')}")
    elif scrape_type == "seller":
        print(f"  seller keys: {list(data)[:10]}")
    elif scrape_type == "rufus":
        print(f"  rufus keys: {list(data)[:10]}")
    # Full payload for reference:
    print("\n  Raw response (truncated to 1500 chars):")
    print("  " + json.dumps(data, ensure_ascii=False)[:1500])


if __name__ == "__main__":
    scrape(sys.argv[1] if len(sys.argv) > 1 else "product")