Amazon price watcher in 40 lines — ASINs to a CSV history with movers, QuanticData collectors https://quanticdata.io/blog/how-to-price-watch-on-amazon/

qd_price_watch.py

"""Append today's Amazon prices to a CSV and print what moved since last run.

    pip install requests
    export QD_API_KEY=qd_live_...
    python3 qd_price_watch.py B09XS7JWHH B09HM94VDS us

The CSV is the whole database. Run it from cron; `price_value` is the numeric field
(`price` keeps Amazon's formatting, which differs per marketplace).
$0.003 per product. https://quanticdata.io/collectors/amazon-product-api/
Guide: https://quanticdata.io/blog/how-to-price-watch-on-amazon/
"""
import csv
import os
import pathlib
import sys
import time
from datetime import date

import requests

asins = [a for a in sys.argv[1:] if len(a) == 10]
country = next((a for a in sys.argv[1:] if len(a) == 2), "us")
if not asins:
    sys.exit("usage: python3 qd_price_watch.py <ASIN> [ASIN…] [country]")

BASE = "https://api.quanticdata.io/v1"
H = {"Authorization": f"Bearer {os.environ['QD_API_KEY']}"}
HISTORY = pathlib.Path("prices.csv")

run = requests.post(f"{BASE}/scraper/collectors/amazon_product/run", headers=H, timeout=300,
                    json={"asins": asins, "country": country, "max_results": len(asins)}).json()
if run.get("type") == "error":
    sys.exit(run["message"])
run = run["payload"]
while run.get("status") in ("queued", "running"):
    time.sleep(3)
    run = requests.get(f"{BASE}/scraper/collectors/runs/{run['run_id']}",
                       headers=H, timeout=60).json()["payload"]

previous = {}
if HISTORY.exists():
    for row in csv.DictReader(HISTORY.open(encoding="utf-8")):
        previous[row["asin"]] = (row["date"], float(row["price_value"] or 0))
else:
    with HISTORY.open("w", newline="", encoding="utf-8") as fh:
        csv.writer(fh).writerow(["date", "asin", "price_value", "availability", "title"])

today = date.today().isoformat()
with HISTORY.open("a", newline="", encoding="utf-8") as fh:
    w = csv.writer(fh)
    for p in run.get("results") or []:
        price = p.get("price_value")
        if price is None:
            continue
        w.writerow([today, p["asin"], price, p.get("availability"), (p.get("title") or "")[:80]])
        was = previous.get(p["asin"])
        if was and was[1] and was[1] != price:
            delta = 100 * (price - was[1]) / was[1]
            print(f"{'▲' if price > was[1] else '▼'} {p['asin']}  {was[1]} → {price}  "
                  f"({delta:+.1f}% since {was[0]})  {(p.get('title') or '')[:50]}")
        else:
            print(f"  {p['asin']}  {price}  {(p.get('title') or '')[:50]}")