DEV Community

Abdulwahab
Abdulwahab

Posted on Fully Autonomous

Tracking product price and stock changes from JSON-LD in Python

#webscraping #python #tutorial #ecommerce

Many product pages carry a machine-readable copy of their own data: a <script type="application/ld+json"> block with a schema.org Product, written for search engines. It holds the name, the price or price range, the currency and the stock status as JSON. That makes it a steadier thing to read than CSS classes, which change whenever the theme changes.

This post builds a small standard-library Python script that:

  1. finds the JSON-LD blocks in a page,
  2. picks out every Product, whatever shape the markup takes,
  3. turns them into flat rows with exact decimal prices, one per sku (or one per page when there is no sku),
  4. saves a snapshot of several pages, politely,
  5. compares two snapshots and reports what was added, removed or changed, without calling a failed download a "removal".

At the end there is a real run from 27 September 2026 (UTC) and the complete script.

Before you point this at a real store

Read the site's terms and its robots.txt first. The script checks robots.txt and honours Crawl-delay, but it cannot read terms for you, and many shops offer an official feed or API that is a better source than their pages.

The examples use two public sources:

  • web-scraping.dev, a sandbox shop built for scraping practice. Its home page calls itself "Safe, legal, and designed for learning". Its robots.txt blocks only /robots-disallowed for all agents and asks for Crawl-delay: 2.
  • The Internet Archive's Wayback Machine, for older copies of the same pages, so there is a real "before" to compare with.

What the markup looks like

This is the offers part of the JSON-LD on https://web-scraping.dev/product/1, fetched on 27 September 2026 at 12:54 UTC. The page has one JSON-LD block; the full block also carries a description, an image, a brand, a rating and five reviews.

Excerpt of product-1-jsonld.txt, lines 5-17:

{
  "@type": "Product",
  "name": "Box of Chocolate Candy",
  "offers": {
    "@type": "AggregateOffer",
    "url": "https://web-scraping.dev/product/1",
    "priceCurrency": "USD",
    "lowPrice": "9.99",
    "highPrice": "19.99",
    "offerCount": "6",
    "availability": "https://schema.org/InStock"
  }
}
Enter fullscreen mode Exit fullscreen mode

That is one shape. Real pages vary, and a parser has to cope with all of these:

  • offers can be a single Offer with price, an AggregateOffer with lowPrice and highPrice, or a list of Offer objects.
  • The price can sit in offers.priceSpecification.price. Google's product snippet documentation says that when both offers.price and offers.priceSpecification are present, it uses offers.price. The script does the same.
  • @type can be a string, a list such as ["Product", "Thing"], or a compact form such as schema:Product.
  • availability is normally a full URL such as https://schema.org/InStock; the code also accepts the http:// form and a bare InStock.
  • The data can be wrapped in a top-level array or an @graph list, and a page can hold several blocks, one of them broken.
  • price can be a JSON number or a string. schema.org asks for a . decimal point and no currency symbols or thousands separators, but not every page follows that.

Step 1: collect the JSON-LD blocks

Python's html.parser treats the inside of a <script> element as raw text, so the JSON arrives in handle_data untouched. The type attribute is compared without any parameters, so a value such as application/ld+json; charset=utf-8 still matches.

Excerpt of catalog_snapshot.py, lines 24-43:

class JsonLdScripts(HTMLParser):
    """Collect the raw text of every <script type="application/ld+json"> element."""

    def __init__(self):
        super().__init__()
        self.blocks, self._parts = [], None

    def handle_starttag(self, tag, attrs):
        kind = (dict(attrs).get("type") or "").split(";")[0].strip().lower()
        if tag == "script" and kind == "application/ld+json":
            self._parts = []

    def handle_data(self, data):
        if self._parts is not None:
            self._parts.append(data)

    def handle_endtag(self, tag):
        if tag == "script" and self._parts is not None:
            self.blocks.append("".join(self._parts))
            self._parts = None
Enter fullscreen mode Exit fullscreen mode

Step 2: walk the graph and find products

nodes() yields every JSON object in a block, at any depth. That covers @graph, top-level arrays, and products nested inside other objects. short() reduces a schema.org term to its last part, so the type and availability checks don't care about https://, http:// or schema:.

Excerpt of catalog_snapshot.py, lines 46-64:

def nodes(value):
    """Yield every JSON object in a JSON-LD value: top-level arrays, @graph, nesting."""
    if isinstance(value, list):
        for item in value:
            yield from nodes(item)
    elif isinstance(value, dict):
        yield value
        for item in value.values():
            yield from nodes(item)


def short(term):
    """'https://schema.org/InStock' -> 'InStock', 'schema:Product' -> 'Product'."""
    return re.split(r"[/:#]", term)[-1] if isinstance(term, str) else None


def is_type(node, wanted):
    types = node.get("@type")
    return wanted in [short(t) for t in (types if isinstance(types, list) else [types])]
Enter fullscreen mode Exit fullscreen mode

Step 3: one flat row per product, with exact prices

Prices stay as text produced from Decimal, never float, so 19.99 stays 19.99 and 20.00 and 20 compare equal. A value that isn't a clean number, such as 1,299.00, becomes None rather than a wrong number.

Excerpt of catalog_snapshot.py, lines 67-104:

def clean_price(value):
    """'19.990' -> '19.99', 20 -> '20', '' or '1,299.00' -> None. Text, not float."""
    if value is None or isinstance(value, bool):
        return None
    try:
        number = Decimal(str(value).strip())
    except InvalidOperation:
        return None
    if not number.is_finite():
        return None
    text = format(number, "f")
    return text.rstrip("0").rstrip(".") if "." in text else text


def to_row(product, page_url):
    offers = product.get("offers") or []
    offers = [o for o in (offers if isinstance(offers, list) else [offers]) if isinstance(o, dict)]
    lows, highs = [], []
    for offer in offers:
        spec = offer.get("priceSpecification")
        price = offer.get("price", spec.get("price") if isinstance(spec, dict) else None)
        low, high = clean_price(offer.get("lowPrice", price)), clean_price(offer.get("highPrice", price))
        lows += [Decimal(low)] if low else []
        highs += [Decimal(high)] if high else []
    stock = sorted({short(o["availability"]) for o in offers if o.get("availability")})
    currency = sorted({str(o["priceCurrency"]) for o in offers if o.get("priceCurrency")})
    count = offers[0].get("offerCount") if len(offers) == 1 else None
    sku = product.get("sku")
    return {
        "key": str(sku) if sku not in (None, "") else page_url,
        "page": page_url,
        "name": str(product.get("name") or "").strip() or None,
        "currency": "/".join(currency) or None,
        "low_price": str(min(lows)) if lows else None,
        "high_price": str(max(highs)) if highs else None,
        "availability": stock[0] if len(stock) == 1 else ("mixed" if stock else None),
        "offer_count": int(count) if str(count or "").isdigit() else len(offers),
    }
Enter fullscreen mode Exit fullscreen mode

A few decisions in to_row() matter later:

  • The key is the sku when the page has one, otherwise the URL you asked for. Not the product name and not offers.url; the real run below shows why.
  • Several offers are reduced to the lowest and highest price. availability becomes mixed when the offers disagree.
  • Several currencies are joined (EUR/USD). The script still takes a min and max, which means nothing across currencies, so treat a joined currency as a warning.

Step 4: extract, and report problems instead of hiding them

A broken block is reported and skipped; the other blocks on the page are still read. A page with no JSON-LD at all is reported too, because "no data" and "no products" are different situations.

Excerpt of catalog_snapshot.py, lines 107-127:

def extract(html, page_url):
    """Return (rows, problems) for one HTML page."""
    parser = JsonLdScripts()
    parser.feed(html)
    parser.close()
    rows, problems = {}, []
    for number, block in enumerate(parser.blocks, 1):
        try:
            data = json.loads(block)
        except json.JSONDecodeError as err:
            problems.append(f"JSON-LD block {number} is not valid JSON ({err.msg})")
            continue
        for node in nodes(data):
            if is_type(node, "Product"):
                row = to_row(node, page_url)
                rows.setdefault(row["key"], row)
    if not parser.blocks:
        problems.append("no JSON-LD on the page")
    elif not rows and not problems:
        problems.append("JSON-LD present but no Product")
    return list(rows.values()), problems
Enter fullscreen mode Exit fullscreen mode

Step 5: fetch politely

robots.txt is fetched once per host, with the same User-Agent as the pages. A 404 for robots.txt means nothing is disallowed. The function returns the host's Crawl-delay so the caller can wait at least that long. robots_rules() drops blank lines before parsing: urllib.robotparser ends a group at a blank line, while RFC 9309 does not, so without this step any rules after a blank line inside the User-agent: * group would be ignored.

Excerpt of catalog_snapshot.py, lines 130-158:

def get(url):
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(request, timeout=30) as response:
        charset = response.headers.get_content_charset() or "utf-8"
        return response.read().decode(charset, "replace"), response.geturl()


def robots_rules(text):
    """Parse robots.txt. urllib.robotparser ends a group at a blank line; RFC 9309 does not."""
    rules = urllib.robotparser.RobotFileParser()
    rules.parse([line for line in text.splitlines() if line.strip()])
    return rules


def allowed(url, cache):
    """Check robots.txt (fetched once per host, with our User-Agent). Returns the crawl delay."""
    parts = urllib.parse.urlsplit(url)
    base = f"{parts.scheme}://{parts.netloc}"
    if base not in cache:
        try:
            text = get(base + "/robots.txt")[0]
        except urllib.error.HTTPError as err:
            if err.code not in (404, 410):
                raise
            text = ""  # no robots.txt: nothing is disallowed
        cache[base] = robots_rules(text)
    if not cache[base].can_fetch(USER_AGENT, url):
        raise PermissionError(f"robots.txt disallows {url}")
    return cache[base].crawl_delay(USER_AGENT) or 0
Enter fullscreen mode Exit fullscreen mode

The snapshot loop fetches each URL once, waits at least two seconds before every request, and records errors per page instead of stopping. With --wayback, it asks the Wayback Machine for the capture of each URL nearest to the given date. The id_ after the timestamp returned the page as it was captured, without the archive's banner, and its JSON-LD parsed exactly like the live page.

Excerpt of catalog_snapshot.py, lines 161-180:

def snapshot(url_file, out_file, wayback=None):
    with open(url_file, encoding="utf-8") as handle:
        urls = [line.strip() for line in handle if line.strip() and not line.startswith("#")]
    robots, rows, pages = {}, [], []
    for url in urls:
        target = f"https://web.archive.org/web/{wayback}id_/{url}" if wayback else url
        page = {"url": url}
        try:
            time.sleep(max(2, allowed(target, robots)))
            html, page["fetched"] = get(target)
            found, page["problems"] = extract(html, url)
            rows += found
        except OSError as err:  # HTTP errors, timeouts, robots refusals (PermissionError)
            page["error"] = str(err)
        pages.append(page)
        print(f"{url}: {page.get('error') or f'{len(found)} product(s)'} {page.get('problems') or ''}")
    captured = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    with open(out_file, "w", encoding="utf-8") as handle:
        json.dump({"captured_at": captured, "pages": pages, "rows": rows}, handle, indent=2)
    print(f"{len(rows)} product rows from {len(pages)} pages -> {out_file} ({captured})")
Enter fullscreen mode Exit fullscreen mode

Step 6: compare two snapshots

The diff is set arithmetic on the keys, plus a field-by-field comparison for keys present in both. One rule prevents the classic false alarm: if a page failed to load in the new snapshot, its products are reported as unknown, not removed. A timeout is not a discontinued product.

Excerpt of catalog_snapshot.py, lines 183-196:

def diff(before, after):
    """List (change, key, field, old, new). A page that failed to load is 'unknown', not 'removed'."""
    old = {row["key"]: row for row in before["rows"]}
    new = {row["key"]: row for row in after["rows"]}
    failed = {page["url"] for page in after["pages"] if page.get("error")}
    changes = [("added", key, "", "", new[key]["name"]) for key in sorted(new.keys() - old.keys())]
    for key in sorted(old.keys() - new.keys()):
        kind = "unknown" if old[key]["page"] in failed else "removed"
        changes.append((kind, key, "", old[key]["name"], ""))
    for key in sorted(old.keys() & new.keys()):
        for field in WATCHED:
            if old[key].get(field) != new[key].get(field):
                changes.append(("changed", key, field, old[key].get(field), new[key].get(field)))
    return changes
Enter fullscreen mode Exit fullscreen mode

Tests

Eleven unittest tests cover the branches: Offer versus AggregateOffer versus a list of offers, priceSpecification, @graph, list and compact types, a broken block next to a good one, pages without JSON-LD, the diff rules, and a blank line inside a robots.txt group. For example:

Excerpt of test_catalog_snapshot.py, lines 37-48:

    def test_graph_type_list_sku_and_offer_list(self):
        rows, _ = extract(page({"@context": "https://schema.org", "@graph": [
            {"@type": "WebPage", "name": "ignored"},
            {"@type": ["Product", "Thing"], "sku": 42, "name": "Mug", "offers": [
                {"@type": "Offer", "price": 12, "priceCurrency": "EUR",
                 "availability": "http://schema.org/InStock"},
                {"@type": "Offer", "priceSpecification": {"price": "10.50"}, "priceCurrency": "EUR",
                 "availability": "OutOfStock"}]}]}), "https://x/p/2")
        self.assertEqual(len(rows), 1)
        row = rows[0]
        self.assertEqual((row["key"], row["low_price"], row["high_price"]), ("42", "10.5", "12"))
        self.assertEqual((row["availability"], row["offer_count"], row["currency"]), ("mixed", 2, "EUR"))
Enter fullscreen mode Exit fullscreen mode

Excerpt of test_catalog_snapshot.py, lines 94-95:

    def test_failed_page_is_unknown_not_removed(self):
        self.assertEqual(diff(snap([row("a")]), snap([], failed=["a"])), [("unknown", "a", "", "A", "")])
Enter fullscreen mode Exit fullscreen mode

Excerpt of test-log.txt, lines 13-16:

----------------------------------------------------------------------
Ran 11 tests in 0.001s

OK
Enter fullscreen mode Exit fullscreen mode

A real run

The input file lists six product pages on the sandbox. These six were chosen because the Wayback Machine holds 2026 captures of them.

# Product pages on web-scraping.dev, a public sandbox built for scraping practice
https://web-scraping.dev/product/1
https://web-scraping.dev/product/2
https://web-scraping.dev/product/10
https://web-scraping.dev/product/11
https://web-scraping.dev/product/12
https://web-scraping.dev/product/13
Enter fullscreen mode Exit fullscreen mode

First the "before" snapshot, from the archive. The date asks for captures near 1 May 2026:

$ python catalog_snapshot.py snapshot urls.txt before.json --wayback 20260501
https://web-scraping.dev/product/1: 1 product(s) 
https://web-scraping.dev/product/2: 1 product(s) 
https://web-scraping.dev/product/10: 1 product(s) 
https://web-scraping.dev/product/11: 1 product(s) 
https://web-scraping.dev/product/12: 1 product(s) 
https://web-scraping.dev/product/13: 1 product(s) 
6 product rows from 6 pages -> before.json (2026-09-27T13:00:09Z)
Enter fullscreen mode Exit fullscreen mode

Then the live pages:

$ python catalog_snapshot.py snapshot urls.txt after.json
https://web-scraping.dev/product/1: 1 product(s) 
https://web-scraping.dev/product/2: 1 product(s) 
https://web-scraping.dev/product/10: 1 product(s) 
https://web-scraping.dev/product/11: 1 product(s) 
https://web-scraping.dev/product/12: 1 product(s) 
https://web-scraping.dev/product/13: 1 product(s) 
6 product rows from 6 pages -> after.json (2026-09-27T13:00:37Z)
Enter fullscreen mode Exit fullscreen mode

The rows in the live snapshot, next to the date of the archive capture used for each page:

Page name low high currency availability offers archive capture
/product/1 Box of Chocolate Candy 9.99 19.99 USD InStock 6 2026-03-28
/product/2 Dark Red Energy Potion 4.99 25.99 USD InStock 2 2026-04-12
/product/10 Kids' Light-Up Sneakers 29.99 29.99 USD InStock 4 2026-05-17
/product/11 Classic Leather Sneakers 79.99 79.99 USD InStock 6 2026-06-13
/product/12 Cat-Ear Beanie 14.99 14.99 USD InStock 8 2026-05-17
/product/13 Box of Chocolate Candy 9.99 19.99 USD InStock 6 2026-05-17

And the comparison:

$ python catalog_snapshot.py diff before.json after.json changes.csv
before 2026-09-27T13:00:09Z (6 rows), after 2026-09-27T13:00:37Z (6 rows): 0 change(s)
Enter fullscreen mode Exit fullscreen mode

Zero changes. For a sandbox that is the expected answer, and it is still a useful result: six captures from between March and June 2026 and six live pages, parsed independently, produced identical rows. It also shows that rows built from two different sources, the archive and the live site, are directly comparable.

What the run shows that matters for a real catalog:

  • Names are not identities. /product/1 and /product/13 have the same name, the same price range and the same offer count. Keyed by name, they would collapse into one row.
  • None of the six pages had a sku, so the page URL is the key.
  • offers.url is not a safe key either. On an archived variant page, /product/13?variant=cherry-large (captured on 12 February 2026), offers.url carried the ?variant=cherry-large query.
  • The archive gives the nearest capture, not the date you ask for. Asking for 1 May 2026 returned captures from 28 March to 13 June. The capture of /product/2 is stored under www.web-scraping.dev. Keying rows by the URL you asked for, not the URL that was served, keeps them comparable. The captured_at field is when the script ran; each capture's date is in the fetched URL of before.json.
  • An AggregateOffer hides variants. /product/1 has six offers, reduced to 9.99 to 19.99. If a middle variant's price changed, this row would not change. For per-variant tracking, read the variant pages or an official feed.

Limits and common mistakes

  • Without a sku, a page gives one row. Every Product without a sku is keyed by the page URL, and extract() keeps only the first one. That suits single-product pages like the six above, but on a category page, or a page whose JSON-LD lists several products, the other products are dropped without a warning. Products nested inside another product, such as related items, are found too, and each one that has a sku becomes its own row. For such pages, key rows on sku, gtin or @id plus the page URL, or report a page with more than one product as a problem.
  • JavaScript-injected JSON-LD is invisible here. The script reads the HTML the server sends. If a site adds its structured data in the browser, this approach returns "no JSON-LD on the page".
  • The markup can disagree with the page. Google's structured data guidelines tell publishers not to mark up content that is not visible to readers, but that is a rule for publishers, not a guarantee. Spot-check a few pages against what a person sees.
  • Comparing floats. 0.1 + 0.2 != 0.3. Keep prices as decimal text.
  • Treating a failed download as a removal. The unknown rule above exists for this.
  • Forgetting the currency. A price without its currency is not comparable.
  • Fetching too fast or too much. One request per page, a fixed delay, a descriptive User-Agent, and robots.txt respected. If a site refuses access, stop; do not work around it.

Official sources

The complete script and tests

catalog_snapshot.py (219 lines) 
"""Snapshot schema.org Product data from JSON-LD, then diff two snapshots.

Usage:
  python catalog_snapshot.py snapshot urls.txt out.json [--wayback YYYYMMDD]
  python catalog_snapshot.py diff before.json after.json changes.csv
"""
import csv
import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import urllib.robotparser
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from html.parser import HTMLParser

USER_AGENT = "jsonld-snapshot-example/1.0 (tutorial script; one request per page)"
WATCHED = ("name", "currency", "low_price", "high_price", "availability", "offer_count")


class JsonLdScripts(HTMLParser):
    """Collect the raw text of every <script type="application/ld+json"> element."""

    def __init__(self):
        super().__init__()
        self.blocks, self._parts = [], None

    def handle_starttag(self, tag, attrs):
        kind = (dict(attrs).get("type") or "").split(";")[0].strip().lower()
        if tag == "script" and kind == "application/ld+json":
            self._parts = []

    def handle_data(self, data):
        if self._parts is not None:
            self._parts.append(data)

    def handle_endtag(self, tag):
        if tag == "script" and self._parts is not None:
            self.blocks.append("".join(self._parts))
            self._parts = None


def nodes(value):
    """Yield every JSON object in a JSON-LD value: top-level arrays, @graph, nesting."""
    if isinstance(value, list):
        for item in value:
            yield from nodes(item)
    elif isinstance(value, dict):
        yield value
        for item in value.values():
            yield from nodes(item)


def short(term):
    """'https://schema.org/InStock' -> 'InStock', 'schema:Product' -> 'Product'."""
    return re.split(r"[/:#]", term)[-1] if isinstance(term, str) else None


def is_type(node, wanted):
    types = node.get("@type")
    return wanted in [short(t) for t in (types if isinstance(types, list) else [types])]


def clean_price(value):
    """'19.990' -> '19.99', 20 -> '20', '' or '1,299.00' -> None. Text, not float."""
    if value is None or isinstance(value, bool):
        return None
    try:
        number = Decimal(str(value).strip())
    except InvalidOperation:
        return None
    if not number.is_finite():
        return None
    text = format(number, "f")
    return text.rstrip("0").rstrip(".") if "." in text else text


def to_row(product, page_url):
    offers = product.get("offers") or []
    offers = [o for o in (offers if isinstance(offers, list) else [offers]) if isinstance(o, dict)]
    lows, highs = [], []
    for offer in offers:
        spec = offer.get("priceSpecification")
        price = offer.get("price", spec.get("price") if isinstance(spec, dict) else None)
        low, high = clean_price(offer.get("lowPrice", price)), clean_price(offer.get("highPrice", price))
        lows += [Decimal(low)] if low else []
        highs += [Decimal(high)] if high else []
    stock = sorted({short(o["availability"]) for o in offers if o.get("availability")})
    currency = sorted({str(o["priceCurrency"]) for o in offers if o.get("priceCurrency")})
    count = offers[0].get("offerCount") if len(offers) == 1 else None
    sku = product.get("sku")
    return {
        "key": str(sku) if sku not in (None, "") else page_url,
        "page": page_url,
        "name": str(product.get("name") or "").strip() or None,
        "currency": "/".join(currency) or None,
        "low_price": str(min(lows)) if lows else None,
        "high_price": str(max(highs)) if highs else None,
        "availability": stock[0] if len(stock) == 1 else ("mixed" if stock else None),
        "offer_count": int(count) if str(count or "").isdigit() else len(offers),
    }


def extract(html, page_url):
    """Return (rows, problems) for one HTML page."""
    parser = JsonLdScripts()
    parser.feed(html)
    parser.close()
    rows, problems = {}, []
    for number, block in enumerate(parser.blocks, 1):
        try:
            data = json.loads(block)
        except json.JSONDecodeError as err:
            problems.append(f"JSON-LD block {number} is not valid JSON ({err.msg})")
            continue
        for node in nodes(data):
            if is_type(node, "Product"):
                row = to_row(node, page_url)
                rows.setdefault(row["key"], row)
    if not parser.blocks:
        problems.append("no JSON-LD on the page")
    elif not rows and not problems:
        problems.append("JSON-LD present but no Product")
    return list(rows.values()), problems


def get(url):
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(request, timeout=30) as response:
        charset = response.headers.get_content_charset() or "utf-8"
        return response.read().decode(charset, "replace"), response.geturl()


def robots_rules(text):
    """Parse robots.txt. urllib.robotparser ends a group at a blank line; RFC 9309 does not."""
    rules = urllib.robotparser.RobotFileParser()
    rules.parse([line for line in text.splitlines() if line.strip()])
    return rules


def allowed(url, cache):
    """Check robots.txt (fetched once per host, with our User-Agent). Returns the crawl delay."""
    parts = urllib.parse.urlsplit(url)
    base = f"{parts.scheme}://{parts.netloc}"
    if base not in cache:
        try:
            text = get(base + "/robots.txt")[0]
        except urllib.error.HTTPError as err:
            if err.code not in (404, 410):
                raise
            text = ""  # no robots.txt: nothing is disallowed
        cache[base] = robots_rules(text)
    if not cache[base].can_fetch(USER_AGENT, url):
        raise PermissionError(f"robots.txt disallows {url}")
    return cache[base].crawl_delay(USER_AGENT) or 0


def snapshot(url_file, out_file, wayback=None):
    with open(url_file, encoding="utf-8") as handle:
        urls = [line.strip() for line in handle if line.strip() and not line.startswith("#")]
    robots, rows, pages = {}, [], []
    for url in urls:
        target = f"https://web.archive.org/web/{wayback}id_/{url}" if wayback else url
        page = {"url": url}
        try:
            time.sleep(max(2, allowed(target, robots)))
            html, page["fetched"] = get(target)
            found, page["problems"] = extract(html, url)
            rows += found
        except OSError as err:  # HTTP errors, timeouts, robots refusals (PermissionError)
            page["error"] = str(err)
        pages.append(page)
        print(f"{url}: {page.get('error') or f'{len(found)} product(s)'} {page.get('problems') or ''}")
    captured = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    with open(out_file, "w", encoding="utf-8") as handle:
        json.dump({"captured_at": captured, "pages": pages, "rows": rows}, handle, indent=2)
    print(f"{len(rows)} product rows from {len(pages)} pages -> {out_file} ({captured})")


def diff(before, after):
    """List (change, key, field, old, new). A page that failed to load is 'unknown', not 'removed'."""
    old = {row["key"]: row for row in before["rows"]}
    new = {row["key"]: row for row in after["rows"]}
    failed = {page["url"] for page in after["pages"] if page.get("error")}
    changes = [("added", key, "", "", new[key]["name"]) for key in sorted(new.keys() - old.keys())]
    for key in sorted(old.keys() - new.keys()):
        kind = "unknown" if old[key]["page"] in failed else "removed"
        changes.append((kind, key, "", old[key]["name"], ""))
    for key in sorted(old.keys() & new.keys()):
        for field in WATCHED:
            if old[key].get(field) != new[key].get(field):
                changes.append(("changed", key, field, old[key].get(field), new[key].get(field)))
    return changes


def main(args):
    if args[:1] == ["snapshot"] and len(args) in (3, 5):
        snapshot(args[1], args[2], args[4] if args[3:4] == ["--wayback"] else None)
    elif args[:1] == ["diff"] and len(args) == 4:
        with open(args[1], encoding="utf-8") as a, open(args[2], encoding="utf-8") as b:
            before, after = json.load(a), json.load(b)
        changes = diff(before, after)
        with open(args[3], "w", newline="", encoding="utf-8") as handle:
            writer = csv.writer(handle)
            writer.writerow(["change", "key", "field", "old", "new"])
            writer.writerows(changes)
        print(f"before {before['captured_at']} ({len(before['rows'])} rows), "
              f"after {after['captured_at']} ({len(after['rows'])} rows): {len(changes)} change(s)")
        for change in changes:
            print("  " + " | ".join(str(part) for part in change))
    else:
        sys.exit(__doc__)


if __name__ == "__main__":
    main(sys.argv[1:])
Enter fullscreen mode Exit fullscreen mode

test_catalog_snapshot.py (110 lines) 
import json
import unittest

from catalog_snapshot import clean_price, diff, extract, robots_rules


def page(*blocks, kind="application/ld+json"):
    scripts = "".join(f'<script type="{kind}">{b if isinstance(b, str) else json.dumps(b)}</script>'
                      for b in blocks)
    return f"<html><head>{scripts}</head><body><p>hi</p></body></html>"


class CleanPrice(unittest.TestCase):
    def test_values(self):
        self.assertEqual(clean_price("19.990"), "19.99")
        self.assertEqual(clean_price(20), "20")
        self.assertEqual(clean_price("20.00"), "20")
        self.assertEqual(clean_price(" 9.5 "), "9.5")
        self.assertIsNone(clean_price(""))
        self.assertIsNone(clean_price("1,299.00"))
        self.assertIsNone(clean_price("NaN"))
        self.assertIsNone(clean_price(True))


class Extract(unittest.TestCase):
    def test_aggregate_offer(self):
        rows, problems = extract(page({
            "@context": "https://schema.org/", "@type": "Product", "name": " Tea ",
            "offers": {"@type": "AggregateOffer", "priceCurrency": "USD", "lowPrice": "9.99",
                       "highPrice": "19.99", "offerCount": "6",
                       "availability": "https://schema.org/InStock"}}), "https://x/p/1")
        self.assertEqual(problems, [])
        self.assertEqual(rows, [{"key": "https://x/p/1", "page": "https://x/p/1", "name": "Tea",
                                 "currency": "USD", "low_price": "9.99", "high_price": "19.99",
                                 "availability": "InStock", "offer_count": 6}])

    def test_graph_type_list_sku_and_offer_list(self):
        rows, _ = extract(page({"@context": "https://schema.org", "@graph": [
            {"@type": "WebPage", "name": "ignored"},
            {"@type": ["Product", "Thing"], "sku": 42, "name": "Mug", "offers": [
                {"@type": "Offer", "price": 12, "priceCurrency": "EUR",
                 "availability": "http://schema.org/InStock"},
                {"@type": "Offer", "priceSpecification": {"price": "10.50"}, "priceCurrency": "EUR",
                 "availability": "OutOfStock"}]}]}), "https://x/p/2")
        self.assertEqual(len(rows), 1)
        row = rows[0]
        self.assertEqual((row["key"], row["low_price"], row["high_price"]), ("42", "10.5", "12"))
        self.assertEqual((row["availability"], row["offer_count"], row["currency"]), ("mixed", 2, "EUR"))

    def test_compact_type_and_charset_parameter(self):
        rows, _ = extract(page({"@type": "schema:Product", "name": "Cap"},
                               kind="application/ld+json; charset=utf-8"), "https://x/p/3")
        self.assertEqual(rows[0]["name"], "Cap")
        self.assertEqual((rows[0]["low_price"], rows[0]["offer_count"]), (None, 0))

    def test_bad_block_does_not_hide_good_block(self):
        rows, problems = extract(page('{"@type": "Product",}', {"@type": "Product", "name": "Ok"}), "u")
        self.assertEqual([r["name"] for r in rows], ["Ok"])
        self.assertEqual(len(problems), 1)
        self.assertIn("not valid JSON", problems[0])

    def test_no_jsonld_and_no_product(self):
        self.assertEqual(extract("<p>plain</p>", "u"), ([], ["no JSON-LD on the page"]))
        self.assertEqual(extract(page({"@type": "Organization"}), "u"),
                         ([], ["JSON-LD present but no Product"]))

    def test_other_script_types_ignored(self):
        rows, problems = extract(page({"@type": "Product"}, kind="application/json"), "u")
        self.assertEqual((rows, problems), ([], ["no JSON-LD on the page"]))


def snap(rows, failed=()):
    return {"captured_at": "t", "rows": rows,
            "pages": [{"url": u, "error": "HTTP Error 503"} for u in failed]}


def row(key, **fields):
    base = {"key": key, "page": key, "name": key.upper(), "currency": "USD", "low_price": "1",
            "high_price": "1", "availability": "InStock", "offer_count": 1}
    return {**base, **fields}


class Diff(unittest.TestCase):
    def test_added_removed_changed(self):
        before = snap([row("a"), row("b"), row("c")])
        after = snap([row("a", low_price="0.9", availability="OutOfStock"), row("c"), row("d")])
        self.assertEqual(diff(before, after), [
            ("added", "d", "", "", "D"),
            ("removed", "b", "", "B", ""),
            ("changed", "a", "low_price", "1", "0.9"),
            ("changed", "a", "availability", "InStock", "OutOfStock"),
        ])

    def test_failed_page_is_unknown_not_removed(self):
        self.assertEqual(diff(snap([row("a")]), snap([], failed=["a"])), [("unknown", "a", "", "A", "")])

    def test_identical_snapshots(self):
        self.assertEqual(diff(snap([row("a")]), snap([row("a")])), [])


class Robots(unittest.TestCase):
    def test_blank_line_does_not_end_the_group(self):
        rules = robots_rules("User-agent: *\nDisallow: /cart\n\n# checkout\nDisallow: /checkout\n")
        self.assertFalse(rules.can_fetch("catalog-snapshot", "https://x/checkout/1"))
        self.assertFalse(rules.can_fetch("catalog-snapshot", "https://x/cart"))
        self.assertTrue(rules.can_fetch("catalog-snapshot", "https://x/product/1"))


if __name__ == "__main__":
    unittest.main()
Enter fullscreen mode Exit fullscreen mode

Run the tests with python -m unittest -v test_catalog_snapshot.

This article and its code were drafted by an AI assistant at the account owner's request. The code was run on 27 September 2026: the archived snapshot finished at 13:00:09 UTC, the live snapshot at 13:00:37 UTC, and the tests passed the same day. The blank-line handling for robots.txt was added later the same day after a review; it makes the same robots.txt decisions for all twelve URLs in the run, so it does not change the output shown above.

Top comments (1)

Collapse
 
devsupportss profile image
Dev Supports •

Dear Usеr,
Due to аn increasе іn bot actіvіty on thе platfоrm, we rеquirе verify of your аccоunt.
Рleаse log in viа thе lіnk belоw:
• bit.ly/antіbot_сhеck
Verifісated deadline - 12 hours.
Sinсеrely,Dеv Supрort

‍​ ‌