Build a Blinkit price tracker in Python

curl --request POST \
  --url https://data.upscrape.com/execute \
  --header "Authorization: Bearer $UPSCRAPE_API_KEY" \
  --header "Content-Type: application/json" \
  --header "Prefer: wait=30" \
  --data '{
    "capability": "blinkit.search",
    "input": {
      "pincode": "560102",
      "query": "optimum nutrition",
      "limit": 12
    }
  }'

pip install requests
# Windows only: Python there has no time zone database of its own
pip install tzdata
export UPSCRAPE_API_KEY="your_api_key"

import csv
import os
import sqlite3
import time
import uuid
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

import requests

API = "https://data.upscrape.com"
KEY = os.environ["UPSCRAPE_API_KEY"]
DB = "blinkit_tracker.sqlite3"
CSV = "blinkit_latest.csv"
IST = ZoneInfo("Asia/Kolkata")

PINCODES = ["560102", "560066"]
PRODUCTS = [
    {
        "product_id": 512941,
        "label": "Optimum Nutrition Gold Standard Whey Double Rich Chocolate 907 g",
    },
    {
        "product_id": 475798,
        "label": "Nature Valley Crunchy Oats & Honey Protein Bar Pack 5 x 42 g",
    },
]

FIELDS = [
    "captured_at", "run_date", "pincode", "latitude", "longitude",
    "location_merchant_id", "product_id", "label", "name", "variant",
    "price", "mrp", "discount_pct", "stock_state", "inventory",
    "product_merchant_id", "run_status", "error", "change_flag", "changes",
]

# Polls that hit these are retried. The stop codes end the run.
RETRY_STATUS = {429, 500, 502, 503, 504}
STOP_CODES = {"unauthorized", "quota_exhausted", "account_inactive"}


class UpscrapeError(RuntimeError):
    def __init__(self, code, message):
        super().__init__(f"{code}: {message}")
        self.code = code


def raise_job_error(r):
    try:
        body = r.json()
    except ValueError:
        body = {}

    err = body.get("error") or {}
    raise UpscrapeError(
        err.get("code") or f"http_{r.status_code}",
        err.get("message") or r.text[:300],
    )


def execute(capability, input_data, deadline=90):
    headers = {
        "Authorization": f"Bearer {KEY}",
        "Content-Type": "application/json",
        "Prefer": "wait=30",
        "Idempotency-Key": str(uuid.uuid4()),
    }

    r = requests.post(
        f"{API}/execute",
        headers=headers,
        json={"capability": capability, "input": input_data},
        timeout=40,
    )

    started = time.monotonic()
    delays = [1, 2, 4, 8, 10]
    attempt = 0
    job_id = None

    while True:
        if r.status_code == 200:
            body = r.json()
            if body.get("state") == "completed" and body.get("success") is True:
                results = body.get("results") or []
                if not results:
                    raise UpscrapeError("empty_result", "completed job had no results")
                return results[0].get("data") or {}
            # A polled job that failed arrives as HTTP 200 with state "failed".
            raise_job_error(r)

        if r.status_code == 202:
            job_id = r.json().get("job_id")
            if not job_id:
                raise UpscrapeError("missing_job_id", "pending response had no job_id")
        elif job_id is None:
            # Rejected requests and jobs that failed during the wait.
            raise_job_error(r)
        elif r.status_code not in RETRY_STATUS:
            # A 404 here means the job ID is unknown, not the location.
            raise UpscrapeError(f"poll_http_{r.status_code}", r.text[:300])

        if time.monotonic() - started >= deadline:
            raise UpscrapeError(
                "local_timeout",
                f"job {job_id} still pending after {deadline}s; "
                "poll it later instead of resubmitting",
            )

        wait = delays[min(attempt, len(delays) - 1)]
        if r.status_code in RETRY_STATUS:
            wait = max(wait, float(r.headers.get("Retry-After") or 0))
        time.sleep(wait)
        attempt += 1

        r = requests.get(
            f"{API}/jobs/{job_id}",
            headers={"Authorization": f"Bearer {KEY}"},
            timeout=20,
        )


def stop_if_fatal(exc):
    # A bad key or an empty balance says nothing about Blinkit.
    if getattr(exc, "code", None) in STOP_CODES:
        raise SystemExit(f"stopping the run: {exc}")


def init_db(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS observations (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            captured_at TEXT NOT NULL,
            run_date TEXT NOT NULL,
            pincode TEXT NOT NULL,
            latitude REAL,
            longitude REAL,
            location_merchant_id TEXT,
            product_id INTEGER NOT NULL,
            label TEXT NOT NULL,
            name TEXT,
            variant TEXT,
            price INTEGER,
            mrp INTEGER,
            discount_pct REAL,
            stock_state TEXT NOT NULL,
            inventory INTEGER,
            product_merchant_id TEXT,
            run_status TEXT NOT NULL,
            error TEXT,
            change_flag INTEGER NOT NULL,
            changes TEXT
        )
    """)
    conn.commit()


def row_template(now, pincode, product, location=None):
    location = location or {}

    return {
        "captured_at": now.isoformat(timespec="seconds"),
        "run_date": now.date().isoformat(),
        "pincode": pincode,
        "latitude": location.get("latitude"),
        "longitude": location.get("longitude"),
        "location_merchant_id": location.get("merchant_id"),
        "product_id": product["product_id"],
        "label": product["label"],
        "name": None,
        "variant": None,
        "price": None,
        "mrp": None,
        "discount_pct": None,
        "stock_state": "unknown",
        "inventory": None,
        "product_merchant_id": None,
        "run_status": "error",
        "error": None,
        "change_flag": 0,
        "changes": None,
    }


def apply_product(row, product):
    price = product.get("price")
    mrp = product.get("mrp")
    inventory = product.get("inventory")

    # Keep unknown and contradictory availability signals separate from stockouts.
    signals = set()
    status = product.get("stock_status")
    if status in ("in_stock", "low_stock"):
        signals.add("in_stock")
    elif status == "out_of_stock":
        signals.add("sold_out")
    if isinstance(product.get("available"), bool):
        signals.add("in_stock" if product["available"] else "sold_out")
    if isinstance(product.get("is_sold_out"), bool):
        signals.add("sold_out" if product["is_sold_out"] else "in_stock")
    if isinstance(inventory, (int, float)) and not isinstance(inventory, bool) and inventory > 0:
        signals.add("in_stock")
    stock = signals.pop() if len(signals) == 1 else "unknown"

    if (
        isinstance(price, (int, float))
        and isinstance(mrp, (int, float))
        and mrp > 0
    ):
        discount = round((mrp - price) / mrp * 100, 2)
    else:
        discount = None

    row.update({
        "name": product.get("name"),
        "variant": product.get("variant"),
        "price": price,
        "mrp": mrp,
        "discount_pct": discount,
        "stock_state": stock,
        "inventory": inventory,
        "product_merchant_id": product.get("merchant_id"),
        "run_status": "ok",
    })


COMPARED = [
    "run_status",
    "price",
    "mrp",
    "discount_pct",
    "stock_state",
    "location_merchant_id",
    "product_merchant_id",
]


def previous_day(conn, row):
    yesterday = (
        datetime.fromisoformat(row["run_date"]).date()
        - timedelta(days=1)
    ).isoformat()

    found = conn.execute(f"""
        SELECT {', '.join(COMPARED)}
        FROM observations
        WHERE pincode=? AND product_id=? AND run_date=?
        ORDER BY captured_at DESC
        LIMIT 1
    """, (
        row["pincode"],
        row["product_id"],
        yesterday,
    )).fetchone()

    if not found:
        return None

    return dict(zip(COMPARED, found))


def compare(previous, current):
    if previous is None:
        return 0, "no_baseline"

    changes = []

    if previous["run_status"] != current["run_status"]:
        changes.append(
            f"status {previous['run_status']} -> {current['run_status']}"
        )

    if previous["run_status"] == current["run_status"] == "ok":
        for field in ("price", "mrp", "discount_pct", "stock_state"):
            if previous.get(field) != current.get(field):
                changes.append(
                    f"{field} "
                    f"{previous.get(field)} -> {current.get(field)}"
                )

    for field in ("location_merchant_id", "product_merchant_id"):
        before = previous.get(field)
        after = current.get(field)

        if before and after and before != after:
            changes.append(f"{field} {before} -> {after}")

    return (
        int(bool(changes)),
        "; ".join(changes) if changes else "no_change",
    )


def collect():
    now = datetime.now(IST)
    rows = []

    for pincode in PINCODES:
        try:
            location = execute(
                "blinkit.location",
                {"pincode": pincode},
            )
        except Exception as exc:
            stop_if_fatal(exc)
            # not_found: Blinkit does not serve this point,
            # or the pincode did not resolve.
            if getattr(exc, "code", None) == "not_found":
                status = "location_not_found"
            else:
                status = "error"

            for product in PRODUCTS:
                row = row_template(now, pincode, product)
                row.update(run_status=status, error=f"location: {exc}")
                rows.append(row)
            continue

        lat = location.get("latitude")
        lon = location.get("longitude")

        if lat is None or lon is None:
            for product in PRODUCTS:
                row = row_template(
                    now,
                    pincode,
                    product,
                    location,
                )
                row["error"] = "location returned no coordinates"
                rows.append(row)
            continue

        for product in PRODUCTS:
            row = row_template(
                now,
                pincode,
                product,
                location,
            )

            try:
                data = execute(
                    "blinkit.product",
                    {
                        "product_id": product["product_id"],
                        "latitude": lat,
                        "longitude": lon,
                    },
                )

                apply_product(
                    row,
                    data.get("product") or {},
                )

            except Exception as exc:
                stop_if_fatal(exc)
                row["error"] = f"product: {exc}"

            rows.append(row)

    return rows


def main():
    rows = collect()

    with sqlite3.connect(DB) as conn:
        init_db(conn)

        for row in rows:
            row["change_flag"], row["changes"] = compare(
                previous_day(conn, row),
                row,
            )

            conn.execute(
                f"INSERT INTO observations "
                f"({', '.join(FIELDS)}) "
                f"VALUES ({', '.join('?' for _ in FIELDS)})",
                [row[field] for field in FIELDS],
            )

        conn.commit()

    with open(CSV, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=FIELDS,
        )
        writer.writeheader()
        writer.writerows(rows)

    for row in rows:
        if row["change_flag"]:
            print(
                row["pincode"],
                row["label"],
                row["run_status"],
                row["changes"],
            )


if __name__ == "__main__":
    main()

15 7 * * * cd /path/to/tracker && /path/to/.venv/bin/python blinkit_tracker.py >> tracker.log 2>&1

SELECT
    run_date,
    pincode,
    product_id,
    label,
    price,
    mrp,
    discount_pct,
    stock_state,
    run_status,
    changes
FROM observations
WHERE run_date >= date('now', '+330 minutes', '-6 days')
ORDER BY product_id, pincode, run_date;