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

"""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:])

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()