Website change monitoring API for website change detection and page change monitoring tools

import hashlib, json, os, requests

BASE = "https://api.clawengine.ai/v1"
headers = {"Authorization": f"Bearer {os.environ['CLAWENGINE_API_KEY']}"}

def capture(url, schema):
    r = requests.post(f"{BASE}/extract", headers=headers, timeout=90, json={
        "url": url,
        "format": "json",
        "render": True,              # content is built client-side
        "schema": schema,
    })
    r.raise_for_status()
    return r.json()["data"]

NOISE = {"session_id", "csrf_token", "rendered_at", "view_count", "ab_variant"}

def fingerprint(data):
    # only the fields we actually watch, with volatile keys dropped
    clean = {k: v for k, v in data.items() if k not in NOISE}
    for k, v in clean.items():
        if isinstance(v, str):
            clean[k] = " ".join(v.split())      # collapse whitespace
        if isinstance(v, list):
            clean[k] = sorted(v)                # order is not meaningful
    blob = json.dumps(clean, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(blob.encode()).hexdigest(), clean

url = "https://example.com/pricing"
data = capture(url, {
    "plan_names": "string[]",
    "plan_prices": "number[]",
    "effective_date": "string",
})

digest, clean = fingerprint(data)
prev = load_snapshot(url)                       # your store: Postgres, S3, whatever

if prev is None:
    save_snapshot(url, digest, clean)           # first run: baseline only, no alert
elif prev["digest"] != digest:
    changed = {k: (prev["clean"].get(k), clean[k])
               for k in clean if prev["clean"].get(k) != clean[k]}
    save_snapshot(url, digest, clean)           # append, never overwrite
    notify(url, changed)                        # old value -> new value, per field

# cron: run the fast tier hourly, the rest once a day
#   0 * * * *  python sweep.py --tier hourly
#   15 6 * * * python sweep.py --tier daily
import datetime, os, requests

BASE = "https://api.clawengine.ai/v1"
headers = {"Authorization": f"Bearer {os.environ['CLAWENGINE_API_KEY']}"}

WATCHLIST = [
    # tier the list by how fast the page really moves, not by habit
    {"url": "https://competitor.example/pricing",  "tier": "hourly",
     "schema": {"plan_names": "string[]", "plan_prices": "number[]"}},
    {"url": "https://vendor.example/legal/terms",  "tier": "daily",
     "schema": {"effective_date": "string", "section_titles": "string[]"}},
    {"url": "https://supplier.example/sku/44120",  "tier": "hourly",
     "schema": {"price": "number", "in_stock": "boolean"}},
    {"url": "https://agency.example/rules/2026-07", "tier": "daily",
     "schema": {"title": "string", "last_updated": "string", "status": "string"}},
]

def sweep(tier):
    now = datetime.datetime.now(datetime.timezone.utc).isoformat()
    for item in WATCHLIST:
        if item["tier"] != tier:
            continue
        try:
            r = requests.post(f"{BASE}/extract", headers=headers, timeout=90, json={
                "url": item["url"], "format": "json",
                "render": True, "schema": item["schema"],
            })
            if r.status_code == 404:
                write_diff(item["url"], now, status="gone", delta=None)
                continue                        # a page vanishing is an event too
            r.raise_for_status()
            digest, clean = fingerprint(r.json()["data"])
        except requests.RequestException as e:
            record_failure(item["url"], now, str(e))   # alert on 3 in a row, not 1
            continue

        prev = load_snapshot(item["url"])
        if prev and prev["digest"] != digest:
            write_diff(item["url"], now, status="changed",
                       delta={k: (prev["clean"].get(k), clean[k])
                              for k in clean if prev["clean"].get(k) != clean[k]})
        save_snapshot(item["url"], digest, clean, checked_at=now)  # every run

import crypto from "node:crypto";

const BASE = "https://api.clawengine.ai/v1";
const headers = {
  Authorization: `Bearer ${process.env.CLAWENGINE_API_KEY}`,
  "Content-Type": "application/json",
};

// one call walks the whole tree instead of you writing a frontier queue
const res = await fetch(`${BASE}/crawl`, {
  method: "POST",
  headers,
  body: JSON.stringify({
    url: "https://vendor.example/docs",
    include: ["/docs/*", "/legal/*"],   // scope it, or you crawl the blog too
    max_pages: 800,
    format: "json",
    schema: { title: "string", body_text: "string", last_updated: "string" },
  }),
});
const job = await res.json();

// markdown-level normalization: kill the bits that move on every render
const normalize = (s = "") =>
  s.replace(/\b20\d{2}-\d{2}-\d{2}T[\d:.]+Z?\b/g, "")  // ISO timestamps
   .replace(/\?(utm_[^"\s]+|v=\d+)/g, "")              // cache busters, tracking
   .replace(/\s+/g, " ")
   .trim();

const hash = (s) => crypto.createHash("sha256").update(s).digest("hex");

const previous = await loadPreviousRun();   // { url: digest } from the last crawl
const added = [], changed = [], removed = new Set(Object.keys(previous));

for (const page of job.pages) {
  const digest = hash(normalize(page.data.title) + "|" + normalize(page.data.body_text));
  removed.delete(page.url);
  if (!(page.url in previous)) added.push(page.url);
  else if (previous[page.url] !== digest) changed.push(page.url);
  await saveSnapshot(page.url, digest, page.data);   // keep the dated capture
}

console.log({ added, changed, removed: [...removed], checked: job.pages.length });