Monitorare i prezzi Amazon con Python

export EVERYDATA_API_KEY="ds_your_key"

import os

import requests

API = "https://api.everydata.io"
KEY = os.environ["EVERYDATA_API_KEY"]

resp = requests.get(
    f"{API}/amz/amazon-lookup-product-by-asin",
    params={"asin": "B07ZPKN6YR", "domainCode": "com"},
    headers={"x-api-key": KEY},
    timeout=60,
)
resp.raise_for_status()
product = resp.json()

print(product["productTitle"])
print(product["price"], product["currency"], "-", product["warehouseAvailability"])

"""Check the price of a few Amazon products and append each check to prices.csv."""
import csv
import os
import re
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

API = "https://api.everydata.io"
KEY = os.environ["EVERYDATA_API_KEY"]
CSV_FILE = Path("prices.csv")
DOMAIN = "com"  # amazon.com – use "de", "co.uk", "fr" … for other marketplaces

# ASIN -> the price at which you want to hear about it
WATCHLIST = {
    "B07ZPKN6YR": 30.00,
}


def api_get(path: str, params: dict) -> dict:
    """GET an everydata.io endpoint. Waits when rate limited (429) and retries a failed page load (5xx) once."""
    retried = False
    for _ in range(4):
        resp = requests.get(f"{API}{path}", params=params, headers={"x-api-key": KEY}, timeout=60)
        if resp.status_code == 429:
            # Too many requests at once. The answer says how long to wait: "... Try again in 12 seconds."
            wait = re.search(r"in (\d+) seconds", resp.text)
            time.sleep(int(wait.group(1)) + 1 if wait else 60)
            continue
        if resp.status_code >= 500 and not retried:
            retried = True  # 5xx answers are not counted against your quota
            time.sleep(10)
            continue
        resp.raise_for_status()  # 4xx (wrong key, quota used up, bad parameter) or a second 5xx
        return resp.json()
    raise RuntimeError("still rate limited after several waits – try again later")


def last_price(asin: str) -> float | None:
    """The most recent price we saved for this ASIN, if any."""
    if not CSV_FILE.exists():
        return None
    with CSV_FILE.open(newline="") as f:
        rows = [r for r in csv.DictReader(f) if r["asin"] == asin and r["price"]]
    return float(rows[-1]["price"]) if rows else None


def main() -> None:
    previous = {asin: last_price(asin) for asin in WATCHLIST}
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    new_file = not CSV_FILE.exists()

    with CSV_FILE.open("a", newline="") as f:
        writer = csv.writer(f)
        if new_file:
            writer.writerow(["checked_at", "asin", "title", "price", "retail_price", "currency", "availability"])

        for asin, target in WATCHLIST.items():
            try:
                p = api_get("/amz/amazon-lookup-product-by-asin", {"asin": asin, "domainCode": DOMAIN})
            except (requests.RequestException, RuntimeError) as exc:
                print(f"{asin}: could not check this time ({exc}) – the next run tries again")
                continue  # one failed product does not stop the others
            price = p.get("price") or None  # 0 means the page showed no price
            writer.writerow([
                now, asin, p.get("productTitle"), price or "", p.get("retailPrice") or "",
                p.get("currency") or "", p.get("warehouseAvailability") or "",
            ])

            if price is None:
                print(f"{asin}: no price on the page ({p.get('warehouseAvailability')})")
            elif price <= target:
                print(f"{asin}: {price} {p.get('currency')} – at or below your target of {target}")
            elif previous[asin] is not None and price < previous[asin]:
                print(f"{asin}: dropped from {previous[asin]} to {price} {p.get('currency')}")


if __name__ == "__main__":
    main()

# Every day at 07:00 – edit with: crontab -e
0 7 * * * cd /home/you/price-tracker && EVERYDATA_API_KEY=ds_your_key /usr/bin/python3 track.py >> track.log 2>&1

resp = requests.get(
    f"{API}/amz/amazon-price-history",
    params={"asin": "B07ZPKN6YR", "domainCode": "com", "days": 90},
    headers={"x-api-key": KEY},
    timeout=60,
)
data = resp.json()
print(data["count"], "observations")
for point in data["history"]:
    print(point["observedAt"], point["price"], point["inStock"])