Scrape Retail Price Data Without Getting Blocked

from dataclasses import dataclass, field
from typing import List

@dataclass
class ScrapeTarget:
    retailer: str
    url: str
    sku: str
    render_js: bool = False  # per-target JS rendering flag
    country: str = "us"

TARGETS: List[ScrapeTarget] = [
    ScrapeTarget("amazon",  "https://www.amazon.com/dp/B0CHWQR1VW",       "B0CHWQR1VW", render_js=True),
    ScrapeTarget("walmart", "https://www.walmart.com/ip/123456789",        "123456789",  render_js=False),
    ScrapeTarget("target",  "https://www.target.com/p/A-12345678",         "A-12345678", render_js=True),
    ScrapeTarget("bestbuy", "https://www.bestbuy.com/site/12345678.p",     "12345678",   render_js=True),
]

import asyncio
import httpx
from models import ScrapeTarget

API_KEY  = "YOUR_API_KEY"
ENDPOINT = "https://api.alterlab.io/v1/scrape"

# Limit concurrent requests per retailer to avoid velocity triggers
CONCURRENCY: dict[str, int] = {
    "amazon":  2,
    "walmart": 3,
    "target":  2,
    "bestbuy": 3,
}

async def fetch_page(client: httpx.AsyncClient, target: ScrapeTarget) -> dict:
    payload = {
        "url":        target.url,
        "render_js":  target.render_js,
        "country":    target.country,
        "session":    target.retailer,  # reuse browser session per retailer
    }
    resp = await client.post(
        ENDPOINT,
        headers={"X-API-Key": API_KEY},
        json=payload,
        timeout=45.0,
    )
    resp.raise_for_status()
    return {"target": target, "html": resp.json()["content"]}

async def scrape_all(targets: list[ScrapeTarget]) -> list[dict]:
    sems = {r: asyncio.Semaphore(CONCURRENCY.get(r, 2)) for r in {t.retailer for t in targets}}

    async def bounded(t: ScrapeTarget):
        async with sems[t.retailer]:
            return await fetch_page(client, t)

    async with httpx.AsyncClient() as client:
        return await asyncio.gather(*[bounded(t) for t in targets], return_exceptions=True)

from selectolax.parser import HTMLParser
import re

# CSS selectors for price extraction per retailer
PRICE_SELECTORS: dict[str, str] = {
    "amazon":  "#corePriceDisplay_desktop_feature_div .a-price-whole",
    "walmart": '[itemprop="price"]',
    "target":  '[data-test="product-price"]',
    "bestbuy": ".priceView-customer-price span",
}

OOS_SELECTORS: list[str] = [
    '[data-automation="out-of-stock"]',
    '[data-test="soldOutMessage"]',
    ".fulfillment-add-to-cart-button--disabled",
    "#outOfStock",
]

def parse_price(retailer: str, html: str) -> float | None:
    tree     = HTMLParser(html)
    selector = PRICE_SELECTORS.get(retailer)
    if not selector:
        return None

    node = tree.css_first(selector)
    if not node:
        return None

    raw     = node.text(strip=True)
    cleaned = re.sub(r"[^\d.]", "", raw)

    try:
        return float(cleaned)
    except ValueError:
        return None

def parse_availability(html: str) -> bool:
    tree = HTMLParser(html)
    return not any(tree.css_first(s) for s in OOS_SELECTORS)

import httpx

API_KEY  = "YOUR_API_KEY"
ENDPOINT = "https://api.alterlab.io/v1/scrape"

async def fetch_walmart_price(item_id: str) -> dict:
    """
    Walmart's terra-firma item API returns structured pricing JSON.
    No JS rendering required — faster and selector-drift-proof.
    """
    target_url = f"https://www.walmart.com/terra-firma/item/{item_id}?rgs=PROD"

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            ENDPOINT,
            headers={"X-API-Key": API_KEY},
            json={"url": target_url, "render_js": False, "extract_json": True},
        )

    data = resp.json()["json_content"]
    offer = data["payload"]["offers"]["primaryOffer"]

    return {
        "price":        offer["offerPrice"],
        "availability": offer["availabilityStatus"],
        "currency":     "USD",
    }

import psycopg2

DDL = """
CREATE TABLE IF NOT EXISTS price_snapshots (
    id         BIGSERIAL PRIMARY KEY,
    sku        TEXT        NOT NULL,
    retailer   TEXT        NOT NULL,
    price      NUMERIC(10,2),
    available  BOOLEAN,
    scraped_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_price_sku
    ON price_snapshots (sku, retailer, scraped_at DESC);
"""

def write_snapshot(conn, sku: str, retailer: str, price: float | None, available: bool) -> None:
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO price_snapshots (sku, retailer, price, available) VALUES (%s, %s, %s, %s)",
            (sku, retailer, price, available),
        )
    conn.commit()

def get_cheapest(conn, sku: str) -> list[dict]:
    """Return cheapest available offers for a SKU from the last 24 hours."""
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT retailer, price, scraped_at
            FROM   price_snapshots
            WHERE  sku = %s
              AND  available = TRUE
              AND  scraped_at > NOW() - INTERVAL '24 hours'
            ORDER  BY price ASC
            LIMIT  5
            """,
            (sku,),
        )
        rows = cur.fetchall()
    return [{"retailer": r[0], "price": float(r[1]), "scraped_at": r[2].isoformat()} for r in rows]

# Start worker and beat scheduler
celery -A price_pipeline worker --loglevel=info --concurrency=10 &
celery -A price_pipeline beat   --loglevel=info &

from collections import defaultdict
import structlog

log = structlog.get_logger()

class DriftMonitor:
    def __init__(self, alert_threshold: float = 0.05):
        self.threshold = alert_threshold
        self._counts: dict = defaultdict(lambda: {"total": 0, "null": 0})

    def record(self, retailer: str, price: float | None) -> None:
        self._counts[retailer]["total"] += 1
        if price is None:
            self._counts[retailer]["null"] += 1
        self._check_drift(retailer)

    def _check_drift(self, retailer: str) -> None:
        counts = self._counts[retailer]
        if counts["total"] < 20:
            return  # insufficient sample

        null_rate = counts["null"] / counts["total"]
        if null_rate > self.threshold:
            log.warning(
                "selector_drift_detected",
                retailer=retailer,
                null_rate=round(null_rate, 3),
                total_requests=counts["total"],
            )