Build a price monitor

{
  "parsed": {
    "product": {
      "asin": "B0DRFGNWV1",
      "title": "Redragon K521 Upgrade Rainbow LED Gaming Keyboard, 104 Keys Wired Mechanical Feeling Keyboard ..."
    },
    "pricing": {
      "currency": "$",
      "current_price": 14,
      "deal_badge": null,
      "discount_percent": null,
      "list_price": 15.99
    }
  }
}

import json
import os
import re
import sqlite3
from datetime import datetime, timezone

import requests

ZENROWS_API_KEY = "YOUR_ZENROWS_API_KEY"
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")  # optional

# Retailers Zenrows has prepared (Amazon, Walmart, eBay, Best Buy...) need no selectors.
# For any other store, give a css_extractor with the name and price selectors.
STORE_SELECTORS = {"name": "h1.product_title", "price": "p.price", "sku": ".sku"}
PRODUCTS = [
    {"url": "https://www.amazon.com/dp/B0DRFGNWV1"},
    {"url": "https://www.scrapingcourse.com/ecommerce/product/adrienne-trek-jacket/", "css_extractor": STORE_SELECTORS},
    {"url": "https://www.scrapingcourse.com/ecommerce/product/aero-daily-fitness-tee/", "css_extractor": STORE_SELECTORS},
]

db = sqlite3.connect("prices.db")
db.execute("CREATE TABLE IF NOT EXISTS prices (url TEXT, name TEXT, price REAL, checked_at TEXT)")

def zenrows_params(item: dict) -> dict:
    """Extract for prepared retailers, css_extractor for any other store."""
    if "css_extractor" in item:
        return {"mode": "auto", "css_extractor": json.dumps(item["css_extractor"])}
    return {"mode": "auto", "proxy_country": "us", "extract": "auto"}

def read_product(body: dict) -> dict:
    """Normalize either response to a name and a numeric price."""
    if "parsed" in body:  # extract=auto
        parsed = body["parsed"]
        product = parsed.get("product") or {}
        pricing = parsed.get("pricing") or {}
        return {
            "name": product.get("name") or product.get("title"),
            "price": pricing.get("current_price", product.get("price")),
        }
    match = re.search(r"[\d.,]+", body.get("price") or "")  # css_extractor: "$57.00"
    return {"name": body.get("name"), "price": float(match.group().replace(",", "")) if match else None}

def get_product(item: dict) -> dict:
    response = requests.get(
        "https://api.zenrows.com/v1/",
        params={"apikey": ZENROWS_API_KEY, "url": item["url"], **zenrows_params(item)},
        timeout=180,
    )
    response.raise_for_status()
    return read_product(response.json())

def alert(message: str) -> None:
    if SLACK_WEBHOOK_URL:
        requests.post(SLACK_WEBHOOK_URL, json={"text": message}, timeout=30)
    else:
        print(message)

def record(url: str, product: dict) -> None:
    """Store today's price, then compare it with the last one stored for this URL."""
    if product["price"] is None:
        print(f"No price found today: {url}")
        return
    previous = db.execute(
        "SELECT price FROM prices WHERE url = ? ORDER BY checked_at DESC LIMIT 1", (url,)
    ).fetchone()
    price = float(product["price"])
    db.execute(
        "INSERT INTO prices VALUES (?, ?, ?, ?)",
        (url, product["name"], price, datetime.now(timezone.utc).isoformat()),
    )
    db.commit()

    if previous and price < previous[0]:
        drop = (previous[0] - price) / previous[0] * 100
        alert(f"Price drop: {product['name']} {previous[0]:.2f} -> {price:.2f} (-{drop:.0f}%)\n{url}")
    else:
        print(f"{product['name'][:40]}: {price:.2f}")

if __name__ == "__main__":
    for item in PRODUCTS:
        record(item["url"], get_product(item))

import json
import sys

from zenrows import ZenRowsBatchClient
from zenrows.batch import Calendar, Daily

from monitor import PRODUCTS, read_product, record, zenrows_params

client = ZenRowsBatchClient(api_key="YOUR_ZENROWS_API_KEY")
TASKS = [{"url": item["url"], "zenrows_params": zenrows_params(item)} for item in PRODUCTS]

def schedule() -> None:
    """Create the daily job once. Zenrows runs it every day at 08:00."""
    job = client.submit_scheduled(
        Calendar(times_of_day=["08:00"], cadence=Daily(), timezone="America/New_York"),
        TASKS,
    )
    print("Scheduled job:", job.job_id)

def check(job_id: str) -> None:
    """Read the job's latest run and record every price."""
    run = client.job(job_id).run
    for task in run.results(status="successful"):
        body = json.loads(run.download_task_to_memory(task))
        record(str(task.url), read_product(body))

if __name__ == "__main__":
    if sys.argv[1] == "schedule":
        schedule()
    else:
        check(sys.argv[2])