diff_runs.py

#!/usr/bin/env python3
"""
diff_runs.py
-------------
Compares two output files from this project (JSON, as written by
output_writer.save) and reports what changed between them, keyed on `sku` —
the identifier the README already tells people to diff on for price
monitoring and assortment tracking, but that nothing in this repo actually
computed.

    python3 diff_runs.py --old craigslist_products.2026-09-14.json \\
                          --new craigslist_products.2026-09-15.json

Typical use is a scheduled re-run of one of the four scraper engines, kept
under a dated filename, diffed against the previous one:

    python3 playwright_scraper.py --url "$URL" --out "girls_$(date +%F)"
    python3 diff_runs.py --old "girls_$(ls -t girls_*.json | sed -n 2p)" \\
                          --new "girls_$(date +%F).json" --out diff.json

Four buckets, each keyed on sku:

  added          — sku present in --new, absent from --old
  removed        — sku present in --old, absent from --new (delisted, or just
                   off this particular page/category run)
  changed        — sku present in both, with a different price,
                   original_price, discount_pct, currency or in_stock
  source_changed — sku present in both with a different price, but also a
                   different price_source: one run got the DOM-corrected
                   figure and the other the raw JSON-LD one, so the two are
                   not comparable on price. Reported separately because this
                   says something about our own two snapshots, not about the
                   site — and --fail-on-change deliberately ignores it.

A product this project's parser could not recover a sku for (None) cannot be
matched across runs at all, so it is counted and reported separately rather
than silently folded into "added"/"removed", which would be wrong on its face.
"""

import argparse
import json
import pathlib
import re
import sys
from typing import Dict, List, Optional, Tuple
from urllib.parse import parse_qsl, urlsplit

from output_writer import UNIQUE_BY_SKU_MODES

# WHAT CAN ACTUALLY CHANGE ON THIS SITE, and nothing else.
#
# This tuple was inherited tracking `sold` and `sold_is_floor`, which are not
# columns in this repo's schema at all -- so those two comparisons were
# no-ops on every row of every diff, while `title`, `location` and
# `image_count`, which a poster edits routinely, were not watched at all. A
# diff tool watching columns that cannot change and ignoring ones that do is
# worse than no diff tool, because it reports "nothing changed" with
# confidence.
#
# Deliberately NOT here: `original_price`, `discount_pct`, `in_stock` and
# `rating`. Craigslist publishes none of them anywhere -- they are null on
# every row of every run, so tracking them would add four comparisons that
# can only ever be null-to-null. See output_writer.Product, where each says
# which measurement establishes that.
TRACKED_FIELDS = ("price", "currency", "title", "location", "image_count",
                  "updated_at")

# The subset of TRACKED_FIELDS whose comparability depends on price_source
# matching between the two runs — see diff_products.
PRICE_FIELDS = ("price", "original_price", "discount_pct")


def _load(path: str) -> List[dict]:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def _by_sku(products: List[dict]) -> Tuple[Dict[str, dict], int]:
    indexed = {}
    unmatchable = 0
    for p in products:
        sku = p.get("sku")
        if sku is None:
            unmatchable += 1
            continue
        # A run's own output can already hold a duplicate sku (two rows in the
        # same category, or a rerun of dedupe_by_sku's job on older output
        # written before it existed) — keep the first and count the rest as
        # unmatchable rather than letting one clobber the other silently.
        if sku in indexed:
            unmatchable += 1
            continue
        indexed[sku] = p
    return indexed, unmatchable


def _within_tolerance(before: dict, after: dict, changes: dict,
                      tolerance_pct: float) -> bool:
    """True if every differing price field moved by less than `tolerance_pct`.

    Inherited from this family rather than earned here, and said plainly
    because the alternative is a comment inventing a reason. A sibling repo
    needs it: that site converts prices for a cross-border visitor and the
    exchange rate ticks between two runs of the same command. NO EQUIVALENT
    CRAIGSLIST BEHAVIOUR WAS MEASURED — the currency follows the AREA, not
    the exit (eight areas fetched through one US residential exit returned
    USD, CAD, EUR, JPY and MXN, see the README), and a price is the one the
    seller typed, so a run has no conversion in it and any difference is a
    real price move.

    So the flag stays available and DEFAULTS TO ZERO, which makes it inert
    unless someone deliberately asks for it. Set it to something non-zero
    only with a reason you can state; a price monitor that silently swallows
    small moves is worse than one that cries wolf.

    A move is judged on the LARGEST relative change among the price fields,
    so a genuine 0.5% cut is not hidden by a 0.04% tolerance applied
    field-by-field.
    """
    if tolerance_pct <= 0:
        return False
    for field in PRICE_FIELDS:
        if field not in changes:
            continue
        was, now = before.get(field), after.get(field)
        if not isinstance(was, (int, float)) or not isinstance(now, (int, float)):
            return False  # a None appearing or disappearing is a real change
        if was == 0:
            return False
        if abs(now - was) / abs(was) * 100.0 > tolerance_pct:
            return False
    return True


def diff_products(old: List[dict], new: List[dict],
                  price_tolerance_pct: float = 0.0) -> dict:
    old_by_sku, old_unmatchable = _by_sku(old)
    new_by_sku, new_unmatchable = _by_sku(new)

    added = [new_by_sku[sku] for sku in new_by_sku.keys() - old_by_sku.keys()]
    removed = [old_by_sku[sku] for sku in old_by_sku.keys() - new_by_sku.keys()]

    changed, source_changed, within_tolerance = [], [], []
    for sku in old_by_sku.keys() & new_by_sku.keys():
        before, after = old_by_sku[sku], new_by_sku[sku]
        field_changes = {
            field: {"old": before.get(field), "new": after.get(field)}
            for field in TRACKED_FIELDS
            if before.get(field) != after.get(field)
        }
        if not field_changes:
            continue

        # A row whose price_source differs between runs is not comparable on
        # price: here that means one run had its structured price confirmed
        # against a rendered tile ("jsonld+dom") while the other did not
        # ("jsonld"), or fell back to reading the DOM alone ("dom"). The
        # figures should agree, and when they do not, the difference is in
        # how OUR two snapshots rendered, not in what the shop charges.
        # Reporting it as a price change would be a false alarm about the
        # site. Non-price fields still compare fine.
        sources = (before.get("price_source"), after.get("price_source"))
        if sources[0] != sources[1] and any(f in field_changes for f in PRICE_FIELDS):
            price_part = {f: v for f, v in field_changes.items() if f in PRICE_FIELDS}
            other_part = {f: v for f, v in field_changes.items() if f not in PRICE_FIELDS}
            source_changed.append({
                "sku": sku, "title": after.get("title"),
                "price_source": {"old": sources[0], "new": sources[1]},
                "changes": price_part,
            })
            field_changes = other_part
            if not field_changes:
                continue

        # An FX tick rather than a price change — see _within_tolerance. Only
        # when the ONLY differences are price fields: a currency or stock
        # change alongside is a real change whatever the size of the move.
        if (all(f in PRICE_FIELDS for f in field_changes)
                and _within_tolerance(before, after, field_changes,
                                      price_tolerance_pct)):
            within_tolerance.append({"sku": sku, "title": after.get("title"),
                                     "changes": field_changes})
            continue

        changed.append({"sku": sku, "title": after.get("title"),
                        "changes": field_changes})

    return {
        "added": added,
        "removed": removed,
        "changed": changed,
        "source_changed": source_changed,
        "within_tolerance": within_tolerance,
        "unmatchable_old": old_unmatchable,
        "unmatchable_new": new_unmatchable,
    }


def _print_summary(result: dict) -> None:
    print(f"[+] {len(result['added'])} added, {len(result['removed'])} removed, "
          f"{len(result['changed'])} changed, "
          f"{len(result['source_changed'])} not comparable on price, "
          f"{len(result.get('within_tolerance', []))} within the price "
          f"tolerance.")
    for p in result["added"]:
        print(f"  + {p.get('sku')}  {p.get('title')}  {p.get('price')} {p.get('currency')}")
    for p in result["removed"]:
        print(f"  - {p.get('sku')}  {p.get('title')}  {p.get('price')} {p.get('currency')}")
    for c in result["changed"]:
        deltas = ", ".join(f"{f}: {v['old']!r} -> {v['new']!r}" for f, v in c["changes"].items())
        print(f"  ~ {c['sku']}  {c['title']}  {deltas}")
    for c in result.get("within_tolerance", []):
        moves = ", ".join(
            f"{f}: {v['old']} -> {v['new']}" for f, v in c["changes"].items())
        print(f"  ~ {c['sku']}  {c['title']}  {moves}  [within --price-"
              f"tolerance-pct: an exchange-rate tick, not a price change]")
    for c in result["source_changed"]:
        src = c["price_source"]
        deltas = ", ".join(f"{f}: {v['old']!r} -> {v['new']!r}" for f, v in c["changes"].items())
        print(f"  ? {c['sku']}  {c['title']}  {deltas}  "
              f"[price_source {src['old']!r} -> {src['new']!r}: the two runs "
              f"rendered differently, so this is not a site-side price change]")
    unmatchable = result["unmatchable_old"] + result["unmatchable_new"]
    if unmatchable:
        print(f"[!] {unmatchable} row(s) across both files had no sku or a "
              f"duplicate sku, and could not be matched across runs.")


def _run_status(path: str) -> Tuple[Optional[str], Optional[dict]]:
    """Read the `<out>.meta.json` sidecar beside a run's JSON output.

    Returns (status, meta), or (None, None) when there is no sidecar — which
    is the normal case for output written before run metadata existed, or by
    `scraper_api_client.py` (single fetch, no pagination to cut short).
    """
    meta_path = re.sub(r"\.json$", "", path) + ".meta.json"
    try:
        with open(meta_path, "r", encoding="utf-8") as f:
            meta = json.load(f)
    except (OSError, json.JSONDecodeError):
        return None, None
    return meta.get("status"), meta


def _query_scope(url: Optional[str]) -> Optional[tuple]:
    """What a run ASKED for: host, path and query parameters, order-free.

    Craigslist's area is a path segment and its category and filters are
    query parameters, so `/search/area/newyork?cat=sss&max_price=25` and
    `/search/area/losangeles?cat=sss` are different listings that share one
    hostname, one currency and one mode. Two runs of them diffed against
    each other report every row as added or removed. The fragment is dropped:
    the site rewrites it as it renders, and it does not change what was asked.
    """
    if not url:
        return None
    parts = urlsplit(url)
    host = parts.netloc.lower()
    if host.startswith("www."):
        host = host[4:]
    path = parts.path.rstrip("/") or "/"
    return (host, path, tuple(sorted(parse_qsl(parts.query, keep_blank_values=True))))


def _hit_cap(meta: Optional[dict]) -> bool:
    walks = (meta or {}).get("walk") or {}
    return any(isinstance(w, dict) and w.get("hit_cap") for w in walks.values())


def _check_comparable(args) -> bool:
    """Refuse an assortment diff between runs that are not both complete.

    This is the failure mode the sidecar exists for: a run cut short on page
    3 of 10 is missing every product on pages 4-10, and diffing it against
    yesterday's full run reports all of them as `removed` — reading as "these
    products were delisted" when in fact they were simply never fetched.
    Prices of the SKUs both runs DID see are still comparable, which is why
    this is a refusal with a --force escape hatch rather than a hard error.
    """
    problems = []
    modes = {}
    scopes = {}
    for label, path in (("--old", args.old), ("--new", args.new)):
        status, meta = _run_status(path)
        if status is None:
            continue  # no sidecar: nothing to check, see _run_status
        scope = _query_scope((meta or {}).get("start_url"))
        if scope is not None:
            scopes[label] = ((meta or {}).get("start_url"), scope)
        if _hit_cap(meta):
            # Not a refusal: the rows both runs saw are still comparable. But
            # a capped run holds a slice of the listing, so `removed` here can
            # mean "outside this run's slice" rather than "delisted".
            print(f"[!] {label} ({path}) reached Craigslist's result ceiling: "
                  f"its rows are a slice of the listing, so `removed` may "
                  f"include adverts that were simply outside that slice.")
        mode = (meta or {}).get("mode")
        if mode:
            modes[label] = mode
        if mode and mode not in UNIQUE_BY_SKU_MODES:
            # This tool's whole premise is one row per `sku`, diffed on
            # price. A mode that produces many rows per sku would give a diff
            # whose every line is an artefact of two rows sharing an id, so
            # it is refused outright rather than answered. Both of this
            # repo's current modes qualify; the check is here so that adding
            # one that does not is caught rather than discovered.
            problems.append(
                f"{label} ({path}) is a {mode!r} run, which is not one row "
                f"per sku. This tool diffs one row per sku on price, so there "
                f"is nothing here it can compare.")
        if status != "complete":
            problems.append(
                f"{label} ({path}) was a {status!r} run — stopped after "
                f"{meta.get('pages_completed')} of {meta.get('pages_requested')} "
                f"page(s), reason {meta.get('stop_reason')!r}")
    if len(scopes) == 2 and len({v[1] for v in scopes.values()}) > 1:
        problems.append(
            f"the two runs asked for different listings "
            f"({scopes['--old'][0]} vs {scopes['--new'][0]}). The area, the "
            f"category and every filter are part of the URL here, so "
            f"`added`/`removed` would describe the change of question rather "
            f"than the listing.")
    if len(set(modes.values())) > 1:
        problems.append(
            f"the two runs are different modes ({modes}). A listing row and a "
            f"detail row carry different fields, so `added`/`removed` would "
            f"describe the mode change rather than the catalogue.")

    # A CURRENCY MISMATCH, which on this site should be impossible — and is
    # checked anyway.
    #
    # The sibling repos guard cross-storefront diffs with `source`: a country
    # hostname each, so a run of one against another is refused on the
    # hostname alone. That does not work here. Craigslist is ONE hostname
    # worldwide and the AREA is a path segment -- so `source` is
    # "craigslist.org" on both sides of every diff, including a diff of Tokyo
    # against Toronto.
    #
    # The currency is what catches it instead, and on this site it is a real
    # signal rather than a formality: the currency follows the area, measured
    # 2026-09-14 across eight areas fetched through one exit, which returned
    # USD, CAD, EUR, JPY and MXN. Two runs quoting different currencies are
    # two different areas, and every row's price is incomparable between
    # them.
    currencies = {}
    for label, path in (("--old", args.old), ("--new", args.new)):
        try:
            rows = json.loads(pathlib.Path(path).read_text(encoding="utf-8"))
        except (OSError, ValueError):
            continue
        seen = {r.get("currency") for r in rows if r.get("currency")}
        if len(seen) == 1:
            currencies[label] = seen.pop()
        elif len(seen) > 1:
            problems.append(
                f"{label} ({path}) holds more than one currency ({sorted(seen)}) "
                f"— that run was redirected mid-way and its own prices are not "
                f"comparable with each other, let alone with another run's.")
    if len(set(currencies.values())) > 1:
        problems.append(
            f"the two runs quote different currencies ({currencies}), which "
            f"means they are two different AREAS -- the currency follows the "
            f"area on this site, and eight areas fetched through one exit "
            f"returned five different ones. Every row's price is incomparable "
            f"between them. `source` cannot catch this: Craigslist is one "
            f"hostname worldwide, so it reads 'craigslist.org' on both sides.")

    if not problems:
        return True

    # A generic headline, because the reasons below are no longer only about
    # completeness: a mode mismatch and a reviews run are refused too, and a
    # message naming the wrong reason sends the reader looking in the wrong
    # place.
    print("[!] Refusing to diff these two runs:")
    for line in problems:
        print(f"      {line}")
    print("    Re-run the incomplete side (or the same URL), or pass --force "
          "to compare anyway (added/removed will include products that were "
          "simply never fetched).")
    return False


def parse_args():
    p = argparse.ArgumentParser(
        description="Diff two craigslist-scraper JSON outputs by sku.")
    p.add_argument("--old", required=True, help="Earlier run's JSON output.")
    p.add_argument("--new", required=True, help="Later run's JSON output.")
    p.add_argument("--out", default=None,
                   help="Write the full diff as JSON to this path too.")
    p.add_argument("--price-tolerance-pct", type=float, default=0.0,
                   metavar="PCT",
                   help="Treat a price move smaller than PCT%% as an exchange-"
                        "rate tick rather than a price change: reported "
                        "separately and ignored by --fail-on-change. Default 0 "
                        "(report every cent), which is what a Craigslist "
                        "run wants: a run covers one area and therefore one "
                        "currency, so there is no conversion drift to absorb. "
                        "The flag is inherited from this scraper family; set "
                        "it non-zero only with a reason you can state.")
    p.add_argument("--fail-on-change", action="store_true",
                   help="Exit 1 if anything was added, removed or changed — "
                        "for a cron job that should only notify on a real diff.")
    p.add_argument("--force", action="store_true",
                   help="Diff even when a run's .meta.json says it was partial "
                        "or failed, or when the two runs asked for different "
                        "listings (area, category or filters in the URL). "
                        "Products never fetched by one side will appear as "
                        "added/removed.")
    return p.parse_args()


def main() -> int:
    args = parse_args()
    if not args.force and not _check_comparable(args):
        return 2

    try:
        old = _load(args.old)
        new = _load(args.new)
    except (OSError, json.JSONDecodeError) as e:
        print(f"[!] Could not read one of the input files: {e}")
        return 2

    result = diff_products(old, new, price_tolerance_pct=args.price_tolerance_pct)
    _print_summary(result)

    if args.out:
        with open(args.out, "w", encoding="utf-8") as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
        print(f"[+] Full diff written to {args.out}")

    # Neither `source_changed` nor `within_tolerance` is a reason to fail.
    # The first means our own two snapshots rendered differently; the second
    # means an exchange rate moved. Neither says anything about the site, and
    # alerting on either would train whoever reads the alert to ignore it.
    if args.fail_on_change and (result["added"] or result["removed"] or result["changed"]):
        return 1
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        sys.exit(1)