Get alerted when a website adds new pages

import json
import os
import time
from pathlib import Path

import requests

API = "https://api.zenrows.com/v1/crawls"
HEADERS = {"X-API-Key": "YOUR_ZENROWS_API_KEY"}
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")  # optional
STATE = Path("known_urls.json")

# The section to watch
CRAWL = {
    "url": "https://www.scrapingcourse.com/ecommerce/",
    "include_patterns": ["/product/"],
    "depth": 1,
    "max_items": 1000,
}

def crawl_section() -> tuple[dict, set[str]]:
    """Run one crawl and return its final state and every URL it kept."""
    response = requests.post(API, headers=HEADERS, json=CRAWL, timeout=30)
    response.raise_for_status()
    crawl_id = response.json()["crawl_id"]

    while True:
        crawl = requests.get(f"{API}/{crawl_id}", headers=HEADERS, params={"limit": 1}, timeout=30).json()
        if crawl["status"] != "running":
            break
        time.sleep(5)

    download = requests.get(f"{API}/{crawl_id}/download", headers=HEADERS, timeout=120)
    download.raise_for_status()
    urls = {json.loads(line)["url"] for line in download.text.splitlines() if line.strip()}
    return crawl, urls

def alert(message: str) -> None:
    if SLACK_WEBHOOK_URL:
        requests.post(SLACK_WEBHOOK_URL, json={"text": message}, timeout=30)
    else:
        print(message)

def main() -> None:
    crawl, current = crawl_section()
    if crawl["status"] == "failed":
        raise SystemExit(f"Crawl {crawl['crawl_id']} failed: {crawl['error']}")
    print(f"Crawl {crawl['crawl_id']}: {len(current)} URLs, {crawl['status']}")

    if not STATE.exists():
        STATE.write_text(json.dumps(sorted(current), indent=2))
        print("Baseline saved")
        return

    previous = set(json.loads(STATE.read_text()))
    complete = "stop_reason" not in crawl  # the crawl covered the whole section
    added = sorted(current - previous)
    removed = sorted(previous - current) if complete else []

    if added:
        alert(f"{len(added)} new page(s) on {CRAWL['url']}\n" + "\n".join(f"+ {url}" for url in added))
    if removed:
        alert(f"{len(removed)} page(s) gone from {CRAWL['url']}\n" + "\n".join(f"- {url}" for url in removed))
    if not added and not removed:
        print("No new or removed pages")

    # Keep URLs a partial crawl did not reach, so they don't count as new next time
    STATE.write_text(json.dumps(sorted(current if complete else current | previous), indent=2))

if __name__ == "__main__":
    main()

#!/bin/sh
set -e
START_URL="https://www.scrapingcourse.com/ecommerce/"

# 1. Crawl the section and wait until it ends
CRAWL_ID=$(zenrows crawl create "$START_URL" --include-patterns /product/ --depth 1 \
  --max-items 1000 --follow --json | jq -r .crawl_id)
[ "$CRAWL_ID" != null ] || exit 1

# 2. Save the sorted URL list
zenrows crawl download "$CRAWL_ID" --out latest.jsonl
jq -r .url latest.jsonl | sort > current.txt

# 3. Compare with the previous run
if [ -f known.txt ]; then
  comm -13 known.txt current.txt | sed 's/^/+ /'   # new pages
  comm -23 known.txt current.txt | sed 's/^/- /'   # removed pages
else
  echo "Baseline saved: $(wc -l < current.txt | tr -d ' ') URLs"
fi
mv current.txt known.txt