How to monitor Walmart prices in Python with ISP proxies

pip install curl_cffi beautifulsoup4 lxml tenacity loguru pydantic tinydb apprise

# config.py

import os
import re
from urllib.parse import quote
from pydantic import BaseModel, field_validator


class ProxyConfig(BaseModel):
    host: str
    port: str  # str (not int) so it goes straight into the URL f-string below
    user: str
    password: str

    @property
    def url(self):
        # URL-escape user/pass so '@' or ':' inside credentials do not break parsing.
        return f"http://{quote(self.user, safe='')}:{quote(self.password, safe='')}@{self.host}:{self.port}"


class ProductConfig(BaseModel):
    item_id: str
    name: str

    @field_validator("item_id")
    @classmethod
    def validate_item_id(cls, v):
        # Walmart item IDs are numeric, typically 6 to 12 digits. Regex stays
        # loose at 5 to 15 to cover outliers.
        if not re.fullmatch(r"\d{5,15}", v):
            raise ValueError("item_id must be a numeric string of 5 to 15 digits")
        return v


def _load_proxies_from_env():
    raw = os.environ.get("PROXIES", "").strip()
    if not raw:
        # Empty PROXIES = direct mode (curl_cffi TLS impersonation only).
        # Useful for quick testing. Production needs ISP proxies.
        return []
    proxies = []
    for line in raw.splitlines():
        line = line.strip()
        if not line:
            continue
        # maxsplit=3 keeps any ':' inside the password intact.
        parts = line.split(":", 3)
        if len(parts) != 4:
            raise ValueError(f"Bad proxy line (expected host:port:user:pass): {line}")
        host, port, user, password = parts
        proxies.append(ProxyConfig(host=host, port=port, user=user, password=password))
    return proxies


PROXIES = _load_proxies_from_env()
REQUEST_TIMEOUT = 15
MAX_RETRIES = 3

{
  "products": [
    {"item_id": "11381374703", "name": "Apple AirPods 4"},
    {"item_id": "11384707978", "name": "Apple AirPods 4 with Active Noise Cancellation"}
  ]
}

# test_proxies.py

import time
from curl_cffi import requests as curl_requests
from config import PROXIES

for i, proxy in enumerate(PROXIES, 1):
    try:
        t0 = time.time()
        r = curl_requests.get(
            "https://httpbin.org/ip",
            proxy=proxy.url,
            timeout=10,
            impersonate="chrome",
        )
        elapsed = time.time() - t0
        print(f"Proxy {i}: {r.json()['origin']} - OK ({elapsed:.2f}s)")
    except Exception as e:
        print(f"Proxy {i}: FAILED - {e}")

Proxy 1: 203.0.113.10 - OK (1.33s)
Proxy 2: 203.0.113.11 - OK (1.35s)
Proxy 3: 203.0.113.12 - OK (1.30s)

# scraper.py

import json
import random
import time
from datetime import datetime
from itertools import cycle

from curl_cffi import requests as curl_requests
from bs4 import BeautifulSoup
from tenacity import retry, stop_after_attempt, wait_random, retry_if_exception_type
from loguru import logger
from pydantic import BaseModel, Field

from config import PROXIES, REQUEST_TIMEOUT, MAX_RETRIES


class PriceResult(BaseModel):
    item_id: str
    title: str
    brand: str | None = None
    price: float | None = None
    was_price: float | None = None
    list_price: float | None = None
    unit_price_display: str | None = None
    buybox_price_display: str | None = None
    is_reduced: bool = False
    offer_type: str | None = None
    seller_name: str | None = None
    seller_type: str | None = None
    availability: str = "Unknown"
    availability_code: str | None = None
    rating: str | None = None
    review_count: int | None = None
    is_price_event: bool = False
    timestamp: datetime = Field(default_factory=datetime.now)


class RetryableError(Exception):
    """Raised on transient failures (5xx, 429, challenge). Triggers a retry. Permanent errors raise plain Exception."""


NEXT_DATA_SELECTOR = "script#__NEXT_DATA__"

# Narrow markers only. "perimeterx" / "px-captcha" false-positive on real
# product pages because Walmart's CSP meta tag lists *.perimeterx.net.
CHALLENGE_MARKERS = (
    "activate and hold the button",
    "<title>robot or human",
)

BLOCKED_PATH = "/blocked"

PRICE_SELECTORS = [
    '[itemprop="price"]',
    '[data-automation-id="product-price"] span.f1',
    'span[data-testid="price-wrap"]',
    'span[data-seo-id="hero-price"]',
]

class WalmartPriceScraper:
    def __init__(self):
        # Shuffle on startup so a burned proxy is not first on every run.
        # Empty PROXIES = direct mode.
        self._proxy_pool = cycle(random.sample(PROXIES, len(PROXIES))) if PROXIES else None
        if self._proxy_pool is None:
            logger.warning(
                "No PROXIES configured; running in direct mode. "
                "Akamai may flag this IP after sustained requests."
            )

    def _get_next_proxy(self):
        if self._proxy_pool is None:
            return None
        return next(self._proxy_pool).url

    @retry(
        stop=stop_after_attempt(MAX_RETRIES),
        wait=wait_random(min=3, max=10),
        retry=retry_if_exception_type(RetryableError),
    )
    def fetch_product_page(self, item_id):
        url = f"https://www.walmart.com/ip/{item_id}"
        proxy = self._get_next_proxy()

        kwargs = {"timeout": REQUEST_TIMEOUT, "impersonate": "chrome"}
        if proxy is not None:
            kwargs["proxy"] = proxy
        response = curl_requests.get(url, **kwargs)

        # 404 = product delisted. No retry.
        if response.status_code == 404:
            logger.warning(f"Product {item_id} not found (404)")
            return None

        # Transient: retry on next proxy.
        if response.status_code == 429:
            raise RetryableError(f"Rate limited (429) for {item_id}")
        if 500 <= response.status_code < 600:
            raise RetryableError(f"Server error {response.status_code} for {item_id}")

        # Hard block: 307 -> /blocked?url=<base64>. curl_cffi follows redirects by default, so we check the final URL.
        if BLOCKED_PATH in (response.url or ""):
            raise RetryableError(f"Akamai hard block (/blocked) for {item_id}")

        if response.status_code != 200:
            raise Exception(f"Permanent HTTP error {response.status_code} for {item_id}")

        # Soft challenge returns 200 with challenge body. Status alone is not enough.
        body_lower = response.text.lower()
        for marker in CHALLENGE_MARKERS:
            if marker in body_lower:
                raise RetryableError(f"Bot challenge ({marker!r}) for {item_id}")

        return response.text

def _coerce_price(node):
    """Return float price from a Walmart price node, or None when absent or malformed."""
    if not isinstance(node, dict):
        return None
    price = node.get("price")
    if price is None:
        return None
    try:
        return float(price)
    except (TypeError, ValueError):
        return None


def _extract_offer_type(price_display_codes):
    """Return the most specific offer flag, or None."""
    if not isinstance(price_display_codes, dict):
        return None
    for key in ("rollback", "clearance", "reducedPrice"):
        if price_display_codes.get(key):
            return key
    return None


def extract_price_text(tag):
    """Parse a float price from a BeautifulSoup tag (CSS fallback path)."""
    if tag is None:
        return None
    text = tag.get("content") or tag.get_text(strip=True)
    if not text:
        return None
    try:
        return float(str(text).replace("$", "").replace(",", "").strip())
    except (ValueError, TypeError):
        return None

          class WalmartPriceScraper:
    # ... (continued from above)

    def parse_from_next_data(self, soup):
        wrapper = soup.select_one(NEXT_DATA_SELECTOR)
        if wrapper is None:
            return None
        try:
            data = json.loads(wrapper.get_text())
        except (json.JSONDecodeError, ValueError):
            return None

        try:
            product = data["props"]["pageProps"]["initialData"]["data"]["product"]
        except (KeyError, TypeError):
            return None

        price_info = product.get("priceInfo") or {}
        current_price = _coerce_price(price_info.get("currentPrice"))
        was_price = _coerce_price(price_info.get("wasPrice"))
        list_price = _coerce_price(price_info.get("listPrice"))
        unit_price_obj = price_info.get("unitPrice") or {}
        unit_price_display = unit_price_obj.get("priceString") if isinstance(unit_price_obj, dict) else None

        is_reduced = bool(price_info.get("isPriceReduced"))
        offer_type = _extract_offer_type(price_info.get("priceDisplayCodes"))

        # topBoostedOffer = the buybox winner on multi-seller listings; null for Walmart-direct.
        tbo = product.get("topBoostedOffer") or {}
        buybox_price_display = tbo.get("priceString") if isinstance(tbo, dict) else None

        # availabilityStatusV2 carries the human-readable display string ("In stock", "Out of stock").
        avail_v2 = product.get("availabilityStatusV2") or {}
        availability = avail_v2.get("display") if isinstance(avail_v2, dict) else None
        availability = availability or product.get("availabilityStatus") or "Unknown"

        # eventAttributes flags short-term price events: priceFlip and specialBuy are Walmart's terms
        # for limited-time event pricing (distinct from a regular markdown).
        event_attrs = product.get("eventAttributes") or {}
        is_price_event = bool(event_attrs.get("priceFlip") or event_attrs.get("specialBuy"))

        rating_value = product.get("averageRating")
        rating = f"{rating_value:.1f}" if isinstance(rating_value, (int, float)) else None
        review_count_value = product.get("numberOfReviews")
        review_count = int(review_count_value) if isinstance(review_count_value, (int, float)) else None

        return {
            "title": product.get("name") or "Unknown",
            "brand": product.get("brand"),
            "price": current_price,
            "was_price": was_price,
            "list_price": list_price,
            "unit_price_display": unit_price_display,
            "buybox_price_display": buybox_price_display,
            "is_reduced": is_reduced,
            "offer_type": offer_type,
            "seller_name": product.get("sellerName"),
            "seller_type": product.get("sellerType"),
            "availability": availability,
            "availability_code": product.get("availabilityStatus"),
            "rating": rating,
            "review_count": review_count,
            "is_price_event": is_price_event,
        }

class WalmartPriceScraper:
    # ... (continued from above)

    def parse_price_from_css(self, soup):
        """Fallback parser used only if __NEXT_DATA__ is missing."""
        for selector in PRICE_SELECTORS:
            price = extract_price_text(soup.select_one(selector))
            if price is not None:
                return price
        return None

    def parse_product_info(self, html, item_id):
        soup = BeautifulSoup(html, "lxml")
        parsed = self.parse_from_next_data(soup)
        if parsed is not None:
            return PriceResult(item_id=item_id, **parsed)

        # CSS fallback: reached only if Walmart removes or restructures __NEXT_DATA__.
        logger.warning(f"__NEXT_DATA__ missing for {item_id}; falling back to CSS")
        title_tag = soup.select_one("h1[itemprop='name']") or soup.select_one("h1")
        title = title_tag.get_text(strip=True) if title_tag else "Unknown"
        price = self.parse_price_from_css(soup)
        return PriceResult(item_id=item_id, title=title, price=price)

    def get_price(self, item_id):
        # Random delay breaks the uniform request-timing pattern anti-bot systems flag on.
        # Lives outside fetch_product_page so it does not stack with the retry wait.
        time.sleep(random.uniform(3, 7))
        # Catch RetryableError only. Permanent 4xx raises plain Exception
        # and must surface to the caller, not get swallowed here.
        try:
            html = self.fetch_product_page(item_id)
        except RetryableError as e:
            logger.error(f"Retries exhausted for {item_id}: {e}")
            return None
        if html is None:
            return None
        return self.parse_product_info(html, item_id)

"priceInfo": {
  "currentPrice": {
    "price": 99,
    "priceString": "$99.00",
    "currencyUnit": "USD"
  },
  "wasPrice": {
    "price": 129.99,
    "priceString": "$129.99",
    "currencyUnit": "USD"
  },
  "listPrice": null,
  "unitPrice": {
    "price": 99,
    "priceString": "$99.00/count"
  },
  "priceDisplayCodes": {
    "rollback": true,
    "clearance": null,
    "reducedPrice": null
  },
  "isPriceReduced": true
}

# storage.py

from datetime import datetime, timedelta
from loguru import logger
from tinydb import Query

MIN_DROP_PCT = 2.0
MIN_DROP_DOLLARS = 1.00
COOLDOWN_HOURS = 6
BASELINE_WINDOW_HOURS = 24
HISTORY_RETENTION_DAYS = 30


def prune_old_entries(db, retention_days=HISTORY_RETENTION_DAYS):
    """Delete readings older than retention_days so the file does not grow without limit."""
    P = Query()
    cutoff = (datetime.now() - timedelta(days=retention_days)).isoformat()
    removed = db.remove(P.timestamp < cutoff)
    if removed:
        logger.info(f"Pruned {len(removed)} entries older than {retention_days}d")


def get_baseline_price(db, item_id, window_hours=BASELINE_WINDOW_HOURS):
    """Return the lowest price seen for this item_id in the last window_hours, or None."""
    P = Query()
    cutoff = (datetime.now() - timedelta(hours=window_hours)).isoformat()
    recent = db.search((P.item_id == item_id) & (P.timestamp >= cutoff))
    prices = [r["price"] for r in recent if r.get("price") is not None]
    return min(prices) if prices else None


def get_last_alert_time(db, item_id):
    """Return the timestamp of the most recent alerted reading, or None."""
    P = Query()
    alerted = db.search((P.item_id == item_id) & (P.alerted == True))  # noqa: E712
    if not alerted:
        return None
    most_recent = max(alerted, key=lambda r: r.get("timestamp", ""))
    return datetime.fromisoformat(most_recent["timestamp"])


def decide(current, baseline, last_alert_at, now):
    """Pure function. Returns (should_alert, reason_string)."""
    if baseline is None:
        return False, "no recent baseline (re-establishing)"

    if current >= baseline:
        if current == baseline:
            return False, f"matches 24h low ${baseline:.2f}"
        return False, f"above 24h low ${baseline:.2f}"

    drop = baseline - current
    pct = (drop / baseline) * 100

    if drop < MIN_DROP_DOLLARS or pct < MIN_DROP_PCT:
        return False, (
            f"drop -${drop:.2f}/-{pct:.2f}% below threshold "
            f"(need >=${MIN_DROP_DOLLARS:.2f} AND >={MIN_DROP_PCT}%)"
        )

    if last_alert_at is not None:
        hours_since = (now - last_alert_at).total_seconds() / 3600
        if hours_since < COOLDOWN_HOURS:
            return False, (
                f"cooldown active ({hours_since:.1f}h since last alert, "
                f"need {COOLDOWN_HOURS}h)"
            )

    return True, f"new 24h low (was ${baseline:.2f}, drop -${drop:.2f}/-{pct:.2f}%)"

# alerts.py

import os
import apprise
from loguru import logger
from scraper import PriceResult
from config import ProductConfig

notifier = apprise.Apprise()

for url in os.environ.get("APPRISE_URLS", "").strip().splitlines():
    url = url.strip()
    if url:
        notifier.add(url)


def _format_body(result: PriceResult, prior_price: float) -> str:
    assert result.price is not None
    drop = prior_price - result.price
    pct = (drop / prior_price) * 100

    lines = [
        f"Previous: ${prior_price:.2f}",
        f"Current:  ${result.price:.2f}",
        f"Drop:     ${drop:.2f} (-{pct:.2f}%)",
    ]

    # Walmart's wasPrice / listPrice as a separate comparison anchor
    msrp_anchor, msrp_label = None, None
    if result.was_price is not None and result.was_price > result.price:
        msrp_anchor, msrp_label = result.was_price, "vs Was"
    elif result.list_price is not None and result.list_price > result.price:
        msrp_anchor, msrp_label = result.list_price, "vs MSRP"
    if msrp_anchor is not None:
        msrp_drop = msrp_anchor - result.price
        msrp_pct = (msrp_drop / msrp_anchor) * 100
        lines.append(f"{msrp_label}:   ${msrp_anchor:.2f} (-${msrp_drop:.2f} / -{msrp_pct:.2f}%)")

    if result.unit_price_display:
        lines.append(f"Per unit: {result.unit_price_display}")

    if result.buybox_price_display and result.seller_type and result.seller_type != "INTERNAL":
        lines.append(f"Buybox:   {result.buybox_price_display}")

    tags = []
    if result.offer_type:
        tags.append(result.offer_type.replace("Price", "").upper())
    if result.is_price_event:
        tags.append("EVENT-PRICING")
    if result.seller_type and result.seller_type != "INTERNAL":
        seller = result.seller_name or "3rd-party"
        tags.append(f"SOLD BY {seller.upper()}")
    if tags:
        lines.append(f"Tags:     {' | '.join(tags)}")

    if result.availability and result.availability not in ("Unknown", "In stock"):
        lines.append(f"Stock:    {result.availability}")

    lines.append("")
    lines.append(f"https://www.walmart.com/ip/{result.item_id}")
    return "\n".join(lines)


def send_alert(result: PriceResult, product: ProductConfig, prior_price: float) -> bool:
    """Returns True if delivery confirmed, False otherwise.

    The caller uses the return value to decide whether to mark the reading as
    alerted in TinyDB. Only marking on success keeps the cooldown correct
    across transient Slack outages.
    """
    assert result.price is not None
    title = f"Price Drop: {product.name}"
    body = _format_body(result, prior_price)

    if len(notifier) == 0:
        logger.warning(f"No notification services configured! {title}")
        return False

    if notifier.notify(title=title, body=body):
        logger.success(
            f"Alert sent for {result.item_id}: ${result.price:.2f} (was ${prior_price:.2f})"
        )
        return True

    logger.error(
        f"Alert delivery failed for {result.item_id}: "
        f"${result.price:.2f} (was ${prior_price:.2f})"
    )
    return False

apprise -vv -t "test" -b "Price monitor test" "slack://A/B/C"

# check_once.py

import sys, json
from datetime import datetime
from loguru import logger
from pydantic import TypeAdapter
from tinydb import TinyDB

from scraper import WalmartPriceScraper
from config import ProductConfig
from alerts import send_alert
from storage import prune_old_entries, get_baseline_price, get_last_alert_time, decide


def main():
    with open("products.json") as f:
        data = json.load(f)
    products = TypeAdapter(list[ProductConfig]).validate_python(data["products"])

    logger.info(f"Checking {len(products)} products")
    db = TinyDB("price_history.json")
    prune_old_entries(db)
    scraper = WalmartPriceScraper()
    now = datetime.now()

    successes = failures = drops_alerted = 0
    for product in products:
        result = scraper.get_price(product.item_id)
        if not (result and result.price is not None):
            logger.warning(f"Failed to get price for {product.name} ({product.item_id})")
            failures += 1
            continue

        baseline = get_baseline_price(db, product.item_id)
        last_alert_at = get_last_alert_time(db, product.item_id)
        should_alert, reason = decide(result.price, baseline, last_alert_at, now)

        successes += 1
        delivered = False
        if should_alert:
            assert baseline is not None
            delivered = send_alert(result, product, baseline)
            if delivered:
                drops_alerted += 1
        logger.info(f"{product.name}: ${result.price:.2f} | {reason}")

        # Mark alerted only on confirmed delivery, so the cooldown does not
        # lock out the next cycle when a Slack outage drops an alert.
        record = result.model_dump(mode="json")
        record["alerted"] = delivered
        db.insert(record)

    logger.info(f"Cycle done. {successes} ok, {failures} failed, {drops_alerted} alert(s)")
    if successes == 0 and failures > 0:
        sys.exit(1)


if __name__ == "__main__":
    main()