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

{ "@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" } } 

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...