scrape.py

"""Amazon scraper in Python: product prices, ratings and price drops to CSV.

Runs the Amazon Products Scraper (an Apify Actor) for one or more keywords on
an Amazon marketplace and saves one row per product to a CSV file.

    export APIFY_TOKEN="<your-apify-token>"
    python scrape.py "wireless earbuds" --max-results 60
    python scrape.py "air fryer" --marketplace UK --price-min 50 --price-max 100 --sort avg_review
"""

import argparse
import csv
import json
import os
import sys
from decimal import Decimal

from apify_client import ApifyClient

ACTOR = "myth84/amazon-search-scraper"

# Columns written to the CSV. Every row of the dataset has more (see the README).
COLUMNS = [
    "platformProductId",
    "title",
    "currency",
    "price",
    "originalPrice",
    "discountPercent",
    "rating",
    "reviewCount",
    "isChoice",
    "isBestSeller",
    "isSponsored",
    "shippingFree",
    "deliveryEstimate",
    "searchQuery",
    "searchRank",
    "url",
]

# Keys of the attributes column, written as columns of their own.
ATTRIBUTES = ["boughtPastMonth", "coupon"]


def make_parser(description):
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument("keywords", nargs="+", help='searches as typed on Amazon, e.g. "wireless earbuds"')
    parser.add_argument("--marketplace", default="US", choices=["US", "UK", "DE", "FR", "IT", "ES", "MX", "JP", "CA"],
                        help="amazon.com (US), amazon.co.uk (UK), amazon.de (DE)... (default US)")
    parser.add_argument("--sort", default="relevance", choices=["relevance", "price_low", "price_high", "avg_review"],
                        help="Amazon's featured order, price or average customer review")
    parser.add_argument("--price-min", type=float, help="minimum price, in the marketplace currency")
    parser.add_argument("--price-max", type=float, help="maximum price, in the marketplace currency")
    parser.add_argument("--no-sponsored", action="store_true", help="skip sponsored products")
    parser.add_argument("--max-results", type=int, default=50,
                        help="products per keyword (default 50); you pay per product saved")
    parser.add_argument("--max-charge", type=Decimal, help="stop the run once it has cost this many USD")
    return parser


def build_input(args):
    run_input = {
        "searchQueries": args.keywords,
        "marketplace": args.marketplace,
        "sortBy": args.sort,
        "maxResults": args.max_results,
    }
    if args.price_min is not None:
        run_input["priceMin"] = args.price_min
    if args.price_max is not None:
        run_input["priceMax"] = args.price_max
    if args.no_sponsored:
        run_input["includeSponsored"] = False
    return run_input


def run_actor(run_input, max_charge=None):
    """Runs the Actor, waits for it to finish and returns the rows of its dataset."""
    if not os.environ.get("APIFY_TOKEN"):
        sys.exit("Set APIFY_TOKEN first: Apify Console > Settings > API & Integrations.")
    client = ApifyClient(os.environ["APIFY_TOKEN"])
    run = client.actor(ACTOR).call(run_input=run_input, max_total_charge_usd=max_charge)
    if run is None:
        sys.exit("The run could not be started.")
    status = getattr(run.status, "value", run.status)  # SUCCEEDED, FAILED, TIMED-OUT...
    if status != "SUCCEEDED":
        sys.exit(f"The run ended as {status}: open its log in Apify Console.")
    return list(client.dataset(run.default_dataset_id).iterate_items())


def cell(value):
    """Lists and objects as JSON text, so that every CSV row has the same columns."""
    return json.dumps(value, ensure_ascii=False) if isinstance(value, (list, dict)) else value


def write_csv(rows, path, columns=COLUMNS, append=False):
    new_file = not (append and os.path.exists(path))
    with open(path, "a" if append else "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        if new_file:
            writer.writerow(columns + ATTRIBUTES)
        for row in rows:
            attributes = row.get("attributes") or {}
            writer.writerow([cell(row.get(column)) for column in columns]
                            + [cell(attributes.get(key)) for key in ATTRIBUTES])


def main():
    parser = make_parser("Scrape Amazon product prices, ratings and badges to CSV.")
    parser.add_argument("--output", default="amazon_products.csv", help="CSV file (default amazon_products.csv)")
    args = parser.parse_args()

    rows = run_actor(build_input(args), args.max_charge)
    write_csv(rows, args.output)
    print(f"{len(rows)} products saved to {args.output}")

    priced = sorted((row for row in rows if row.get("price") is not None), key=lambda row: row["price"])
    for row in priced[:5]:
        print(f'{row["price"]:>9} {row["currency"]}  {row.get("rating") or "-":>3}/5 '
              f'({row.get("reviewCount") or 0} ratings)  {row["title"][:80]}')


if __name__ == "__main__":
    main()