Minimum Advertised Price Monitoring: How to Build an Automated MAP Tracker in Python

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install httpx parsel apscheduler aiofiles playwright

playwright install chromium

mkdir map_monitor
cd map_monitor
touch config.py scraper.py checker.py alerts.py storage.py scheduler.py main.py

map_monitor/
├── config.py          # Retailer URLs, MAP values, selectors
├── scraper.py         # Price collection logic
├── checker.py         # MAP comparison and violation detection
├── alerts.py          # Email and Slack notification logic
├── storage.py         # Violation persistence
├── scheduler.py       # Scheduling and run orchestration
└── main.py            # Entry point

"""config.py -- single source of truth for products, retailers, and runtime settings."""

PRODUCTS = [
    {
        "name": "Box of Chocolate Candy",
        "map_price": 12.99,   # matches .product-price-full on the detail page
        "retailers": [
            {
                "name": "WebScrapingDev",
                # Server-rendered product page: prices are in the static HTML,
                # no JavaScript execution required for the demo site.
                # In production: swap for your retailer's product page URL.
                "url": "https://web-scraping.dev/product/1",
                # .price > span returns the current advertised/sale price ($9.99).
                # In production: use the price selector for your retailer's page.
                "price_selector": ".price > span",
            },
        ],
    },
    {
        "name": "Hiking Boots for Outdoor Adventures",
        "map_price": 89.99,
        "retailers": [
            {
                "name": "WebScrapingDev",
                "url": "https://web-scraping.dev/product/7",
                "price_selector": ".price > span",
            },
        ],
    },
    {
        "name": "Running Shoes for Men",
        "map_price": 49.99,
        "retailers": [
            {
                "name": "WebScrapingDev",
                "url": "https://web-scraping.dev/product/21",
                "price_selector": ".price > span",
            },
            {
                # XPath selector example for a retailer with a different page structure.
                # parsel handles both CSS and XPath, so selector format can vary per retailer.
                "name": "RetailerB",
                "url": "https://www.retailer-b.com/running-shoes",
                "price_selector": "//div[@class='product-price']/span/text()",
            },
        ],
    },
]

REQUEST_DELAY = 1.5
MAX_CONCURRENT = 5
ALERT_EMAIL = "compliance@yourbrand.com"
SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"

if __name__ == "__main__":
    print(f"Loaded {len(PRODUCTS)} product(s):")
    for p in PRODUCTS:
        print(f"  {p['name']} | MAP: ${p['map_price']} | Retailers: {len(p['retailers'])}")

# From the directory containing map_monitor/
python -m map_monitor.scraper
python -m map_monitor.checker
python -m map_monitor.main

"""scraper.py -- MAP price collection from server-rendered product pages."""

import asyncio
import re
from typing import Optional

import httpx
import parsel

from .config import PRODUCTS, REQUEST_DELAY

HEADERS = {
    "User-Agent": "MAPMonitor/1.0 (compliance-bot; contact@yourbrand.com)",
    "Accept-Language": "en-US,en;q=0.9",
}


def parse_price(response_text: str, selector: str) -> Optional[float]:
    """Extract and normalize a price from page HTML using CSS or XPath selector."""
    tree = parsel.Selector(text=response_text)

    # Detect selector type: XPath starts with // or (//
    if selector.startswith("//") or selector.startswith("(//"):
        raw = tree.xpath(selector).get(default="")
    else:
        raw = tree.css(selector).get(default="")

    if not raw:
        return None

    # Match comma-thousands numbers as single tokens; take last match.
    # "Was $12.99 Now $9.99" correctly returns 9.99 rather than 12.99.
    matches = re.findall(r"\d[\d,]*(?:\.\d+)?", raw.strip())
    if not matches:
        return None

    try:
        return float(matches[-1].replace(",", ""))
    except ValueError:
        return None


async def fetch_price(
    client: httpx.AsyncClient,
    retailer: dict,
    product_name: str,
    delay: float = REQUEST_DELAY,
) -> dict:
    """
    Fetch a single retailer page and extract the advertised price.

    Targets web-scraping.dev/product/{id} for the demo, which serves prices
    in static HTML at .price > span. In production: point url at your
    retailer's product page and update price_selector accordingly.
    """
    await asyncio.sleep(delay)

    try:
        response = await client.get(
            retailer["url"],
            headers=HEADERS,
            timeout=15,
            follow_redirects=True,
        )
        response.raise_for_status()
        price = parse_price(response.text, retailer["price_selector"])

        if price is None:
            print(
                f"[WARN] parse_price returned None for {retailer['name']} -- "
                "selector may be broken or page structure changed."
            )

        return {
            "product": product_name,
            "retailer": retailer["name"],
            "url": retailer["url"],
            "price": price,
            "error": None,
        }

    except Exception as e:
        print(f"[ERROR] Failed to fetch {retailer['name']}: {e}")
        return {
            "product": product_name,
            "retailer": retailer["name"],
            "url": retailer["url"],
            "price": None,
            "error": str(e),
        }


async def scrape_all_retailers(product: dict) -> list[dict]:
    """Scrape all retailer pages for a product concurrently."""
    semaphore = asyncio.Semaphore(5)

    async def bounded_fetch(client, retailer):
        async with semaphore:
            return await fetch_price(client, retailer, product["name"])

    async with httpx.AsyncClient() as client:
        tasks = [
            bounded_fetch(client, retailer)
            for retailer in product["retailers"]
        ]
        return await asyncio.gather(*tasks)


async def scrape_all_products() -> list[dict]:
    """Scrape all products across all retailers."""
    all_results = []
    for product in PRODUCTS:
        results = await scrape_all_retailers(product)
        all_results.extend(results)
    return all_results


if __name__ == "__main__":
    async def main():
        print("Testing static scraper against web-scraping.dev...\n")
        results = await scrape_all_products()
        for r in results:
            status = f"${r['price']:.2f}" if r["price"] else f"ERROR: {r.get('error', 'None')}"
            print(f"  {r['product']} @ {r['retailer']}: {status}")

    asyncio.run(main())

import httpx
from parsel import Selector

response = httpx.get("https://www.scrapingcourse.com/javascript-rendering")
tree = Selector(text=response.text)

prices = tree.css(".product-price").getall()
print(prices)
# Output: []

# The page body contains only a fallback message:
fallback = tree.css("body").get()
print("Enable JavaScript" in fallback)
# Output: True

$env:DECODO_PROXY_USERNAME="user-YOURZONE"
$env:DECODO_PROXY_PASSWORD="your_password"
python playwright_scraper.py
```

For a longer-lived setup, store them in a `.env` file in the project root:
```
DECODO_PROXY_USERNAME=user-YOURZONE
DECODO_PROXY_PASSWORD=your_password

from dotenv import load_dotenv
load_dotenv()

"""playwright_scraper.py -- headless browser MAP tracker for JS-rendered pages."""

import asyncio
import csv
import os
import smtplib
from datetime import datetime
from email.mime.text import MIMEText

import httpx
from parsel import Selector
from playwright.async_api import async_playwright

TARGET_URL = "https://www.scrapingcourse.com/javascript-rendering"

# MAP thresholds: product name -> MAP floor
# Chaz Kangeroo Hoodie lists at $52 -- below the $59.99 MAP floor, triggering
# a violation on the first run. The other 2 products are at or above their floors.
MAP_PRICES = {
    "Chaz Kangeroo Hoodie": 59.99,
    "Teton Pullover Hoodie": 70.00,
    "Bruno Compete Hoodie": 63.00,
}

ALERT_EMAIL = "compliance@yourbrand.com"
SMTP_HOST = "smtp.gmail.com"
SMTP_PORT = 587
SMTP_USER = "alerts@yourbrand.com"
SMTP_PASS = "your_app_password"


def demonstrate_httpx_failure():
    """
    Show that httpx returns no prices from a JS-rendered page.
    Run this before the Playwright scraper to observe the difference.
    """
    print("Fetching with httpx (no JS execution)...")
    response = httpx.get(TARGET_URL)
    tree = Selector(text=response.text)
    prices = tree.css(".product-price").getall()
    print(f"  Prices found by httpx: {len(prices)}")
    print(f"  JS fallback message present: {'Enable JavaScript' in response.text}")
    print()


class MAPTracker:

    def __init__(self):
        # Decodo residential proxy (required).
        proxy_user = os.environ.get("DECODO_PROXY_USERNAME") or os.environ.get("DECODO_USERNAME")
        proxy_pass = os.environ.get("DECODO_PROXY_PASSWORD") or os.environ.get("DECODO_PASSWORD")
        if not proxy_user or not proxy_pass:
            raise ValueError(
                "Decodo proxy required. Set DECODO_PROXY_USERNAME and DECODO_PROXY_PASSWORD "
                "(or DECODO_USERNAME and DECODO_PASSWORD)."
            )
        self.proxy_config = {
            "server": "http://gate.decodo.com:7000",
            "username": proxy_user,
            "password": proxy_pass,
        }

    async def scrape_product_grid(self, page) -> list[dict]:
        """
        Navigate to the JS-rendered product listing and extract all products.
        httpx returns an empty product grid for this URL.
        Playwright executes the JS and exposes the full product list.
        """
        # Use domcontentloaded (less strict than networkidle) - proxy can delay networkidle
        await page.goto(TARGET_URL, wait_until="domcontentloaded", timeout=90000)
        await asyncio.sleep(3)  # Allow JS to render product grid

        # Wait for the JS-rendered product grid to appear
        await page.wait_for_selector("#product-grid .product-item", timeout=15000)

        products = await page.evaluate('''() => {
            const items = document.querySelectorAll("#product-grid .product-item");
            return Array.from(items).map(item => {
                const nameEl = item.querySelector(".product-name");
                const priceEl = item.querySelector(".product-price");
                const name = nameEl ? nameEl.textContent.trim() : "";
                const raw = priceEl ? priceEl.textContent.replace(/[^0-9.]/g, "") : "";
                const price = raw ? parseFloat(raw) : null;
                return { name, price };
            });
        }''')

        return products

    def classify_violation(self, listed_price: float, map_floor: float) -> str:
        """Grade the result: clean, warning (within 5% of MAP), or violation."""
        if listed_price >= map_floor:
            return "clean"
        elif listed_price >= map_floor * 0.95:
            return "warning"
        else:
            return "violation"

    def send_alert(self, subject: str, body: str):
        """Send email alert for confirmed violations."""
        try:
            msg = MIMEText(body)
            msg["Subject"] = subject
            msg["From"] = SMTP_USER
            msg["To"] = ALERT_EMAIL
            with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
                server.starttls()
                server.login(SMTP_USER, SMTP_PASS)
                server.send_message(msg)
            print(f"  Alert sent: {subject}")
        except Exception as e:
            print(f"  Alert failed: {e}")

    async def run(self):
        async with async_playwright() as p:
            browser = await p.chromium.launch(
                headless=True,
                channel="chrome",
                proxy=self.proxy_config,
                args=[
                    "--disable-blink-features=AutomationControlled",
                    "--disable-http2",
                ],
            )
            context = await browser.new_context(
                viewport={"width": 1920, "height": 1080},
                user_agent=(
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                    "AppleWebKit/537.36 (KHTML, like Gecko) "
                    "Chrome/131.0.0.0 Safari/537.36"
                ),
            )
            page = await context.new_page()
            await page.add_init_script(
                "Object.defineProperty(navigator, 'webdriver', { get: () => undefined });"
            )

            print("Fetching with Playwright (JS execution enabled)...")
            scraped_products = await self.scrape_product_grid(page)
            print(f"  Products found by Playwright: {len(scraped_products)}\n")

            await context.close()
            await browser.close()

            # Cross-reference scraped prices against MAP thresholds
            results = []
            for product in scraped_products:
                if product["name"] not in MAP_PRICES or product["price"] is None:
                    continue

                map_floor = MAP_PRICES[product["name"]]
                status = self.classify_violation(product["price"], map_floor)
                gap = round(product["price"] - map_floor, 2)

                row = {
                    "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"),
                    "product": product["name"],
                    "retailer": "ScrapingCourse",
                    "listed_price": product["price"],
                    "map_floor": map_floor,
                    "gap": gap,
                    "status": status,
                    "url": TARGET_URL,
                }
                results.append(row)

                symbol = {"clean": "[OK]", "warning": "[!!]", "violation": "[XX]"}[status]
                print(f"Checking: {product['name']}")
                print(f"  {symbol} ${product['price']} vs MAP ${map_floor} [{status.upper()}]")

                if status == "violation":
                    self.send_alert(
                        f"MAP VIOLATION: {product['name']} @ ScrapingCourse",
                        (
                            f"{product['name']} is listed at ${product['price']}, "
                            f"${abs(gap):.2f} below MAP floor of ${map_floor}.\n\n"
                            f"URL: {TARGET_URL}"
                        ),
                    )

            self.save_results(results)
            self.print_summary(results)

    def save_results(self, results: list, filename: str = "map_violations.csv"):
        if not results:
            print("No results to save.")
            return
        with open(filename, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=results[0].keys())
            writer.writeheader()
            writer.writerows(results)
        print(f"\nSaved {len(results)} records to {filename}")

    def print_summary(self, results: list):
        violations = sum(1 for r in results if r["status"] == "violation")
        warnings = sum(1 for r in results if r["status"] == "warning")
        clean = sum(1 for r in results if r["status"] == "clean")
        print(f"\n--- Run Summary ---")
        print(f"Products checked : {len(results)}")
        print(f"Clean            : {clean}")
        print(f"Warnings (< 5%)  : {warnings}")
        print(f"Violations       : {violations}")


if __name__ == "__main__":
    demonstrate_httpx_failure()
    tracker = MAPTracker()
    asyncio.run(tracker.run())

timestamp,product,retailer,listed_price,map_floor,gap,status,url
2026-03-12 09:15,Chaz Kangeroo Hoodie,ScrapingCourse,52.00,59.99,-7.99,violation,https://www.scrapingcourse.com/javascript-rendering
2026-03-12 09:15,Teton Pullover Hoodie,ScrapingCourse,70.00,70.00,0.0,clean,https://www.scrapingcourse.com/javascript-rendering
2026-03-12 09:15,Bruno Compete Hoodie,ScrapingCourse,63.00,63.00,0.0,clean,https://www.scrapingcourse.com/javascript-rendering

"""decodo_scraper.py -- drop-in replacement for scraper.py using Decodo's Web Scraping API."""

import asyncio
import base64
import os

import httpx

from .config import PRODUCTS
from .scraper import parse_price

DECODO_USERNAME = os.environ.get("DECODO_USERNAME", "YOUR_USERNAME")
DECODO_PASSWORD = os.environ.get("DECODO_PASSWORD", "YOUR_PASSWORD")
DECODO_ENDPOINT = "https://scraper-api.decodo.com/v2/scrape"

# Basic auth token: Base64-encoded "username:password"
_auth_token = base64.b64encode(f"{DECODO_USERNAME}:{DECODO_PASSWORD}".encode()).decode()


async def fetch_price_decodo(
    client: httpx.AsyncClient,
    retailer: dict,
    product_name: str,
) -> dict:
    """Fetch price using Decodo's Web Scraping API."""
    payload = {
        "target": "universal",
        "url": retailer["url"],
        "headless": "html",      # Enable JS rendering (Advanced plan required)
        "geo_location": "us",
    }

    try:
        response = await client.post(
            DECODO_ENDPOINT,
            json=payload,
            headers={
                "Authorization": f"Basic {_auth_token}",
                "Content-Type": "application/json",
            },
            timeout=30,
        )
        response.raise_for_status()
        html = response.json()["results"][0].get("content", "")
        price = parse_price(html, retailer["price_selector"])

        if price is None:
            print(f"[WARN] parse_price returned None for {retailer['name']} via Decodo API.")

        return {
            "product": product_name,
            "retailer": retailer["name"],
            "url": retailer["url"],
            "price": price,
            "error": None,
        }

    except Exception as e:
        print(f"[ERROR] Decodo API fetch failed for {retailer['name']}: {e}")
        return {
            "product": product_name,
            "retailer": retailer["name"],
            "url": retailer["url"],
            "price": None,
            "error": str(e),
        }


async def scrape_all_products_decodo() -> list[dict]:
    """Scrape all products across all retailers via Decodo API."""
    all_results = []
    async with httpx.AsyncClient() as client:
        for product in PRODUCTS:
            tasks = [
                fetch_price_decodo(client, retailer, product["name"])
                for retailer in product["retailers"]
            ]
            results = await asyncio.gather(*tasks)
            all_results.extend(results)
    return all_results


if __name__ == "__main__":
    async def main():
        if DECODO_USERNAME == "YOUR_USERNAME":
            print("Set DECODO_USERNAME and DECODO_PASSWORD env vars before running.")
            return
        print("Testing Decodo scraper...")
        results = await scrape_all_products_decodo()
        for r in results:
            status = f"${r['price']:.2f}" if r["price"] else f"ERROR: {r.get('error', 'None')}"
            print(f"  {r['product']} @ {r['retailer']}: {status}")

    asyncio.run(main())