price_monitor.py

"""
Amazon price monitor - a runnable use case on the Chocodata Amazon Scraper API.

Polls an Amazon search, stores every observation in a local SQLite dataset, and
prints any price change since the previous run. Competitor price tracking is the
most common reason people scrape Amazon, so it is here end to end rather than as
a snippet.

    pip install requests
    export CHOCODATA_API_KEY="your_key"     # free key (1,000 requests, one-time): https://chocodata.com
    python amazon_scraper_api_codes/price_monitor.py "gaming laptop"
    # ... run it again later to see the diffs

Export the dataset to CSV with one command:
    sqlite3 -header -csv amazon_prices.db "SELECT * FROM observations;" > prices.csv

Cost: 1 request (5 credits) per run per query.
Docs: https://chocodata.com/docs
"""
import os
import sqlite3
import sys
import time

import requests

API = "https://api.chocodata.com/api/v1/amazon/search"
KEY = os.environ.get("CHOCODATA_API_KEY")
DB = "amazon_prices.db"

if not KEY:
    sys.exit("Set CHOCODATA_API_KEY first. Free key: https://chocodata.com")


def _check(r) -> None:
    """Map the API's errors onto actionable messages instead of a traceback."""
    if r.status_code == 400:
        sys.exit("400 invalid_params: 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 fetch(query: str) -> list:
    """One API call -> the current ranked cards for this query."""
    r = requests.get(API, params={"api_key": KEY, "query": query}, timeout=90)
    _check(r)
    return r.json().get("products", [])


def setup(conn: sqlite3.Connection) -> None:
    conn.execute(
        """CREATE TABLE IF NOT EXISTS observations (
               asin TEXT, query TEXT, title TEXT, price REAL, currency TEXT,
               rating REAL, reviews_count INTEGER, is_sponsored INTEGER, ts INTEGER,
               PRIMARY KEY (asin, ts)
           )"""
    )


def previous_price(conn: sqlite3.Connection, asin: str):
    row = conn.execute(
        "SELECT price FROM observations WHERE asin = ? ORDER BY ts DESC LIMIT 1", (asin,)
    ).fetchone()
    return row[0] if row else None


def main(query: str) -> None:
    conn = sqlite3.connect(DB)
    setup(conn)
    now = int(time.time())
    products = fetch(query)

    changes = 0
    for item in products:
        asin, price = item.get("asin"), item.get("price")
        # Amazon renders some cards without a scalar price (variant / "see options" rows).
        if not asin or not isinstance(price, (int, float)):
            continue

        before = previous_price(conn, asin)
        if before is not None and before != price:
            delta = price - before
            arrow = "UP  " if delta > 0 else "DOWN"
            pct = (delta / before) * 100 if before else 0
            print(f"{arrow} {(item.get('title') or '')[:56]:56} {before:>9.2f} -> {price:>9.2f} ({pct:+.1f}%)")
            changes += 1

        conn.execute(
            "INSERT OR REPLACE INTO observations VALUES (?,?,?,?,?,?,?,?,?)",
            (asin, query, item.get("title"), price, item.get("currency"),
             item.get("rating"), item.get("reviews_count"),
             int(bool(item.get("is_sponsored"))), now),
        )

    conn.commit()
    tracked = conn.execute("SELECT COUNT(DISTINCT asin) FROM observations").fetchone()[0]
    conn.close()

    priced = sum(1 for p in products if isinstance(p.get("price"), (int, float)))
    print(f"\n{len(products)} cards this run ({priced} priced) | {changes} price change(s) "
          f"| {tracked} products tracked in {DB}")
    if changes == 0:
        print("No changes yet. Run it again in an hour, or schedule it (cron / GitHub Actions).")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "gaming laptop")