Monitor Website Changes with Python

import sqlite3
import difflib
import hashlib
from datetime import datetime

DB_PATH = "website_monitor.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            url TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            markdown TEXT NOT NULL,
            captured_at TEXT NOT NULL
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_url_time ON snapshots(url, captured_at)")
    conn.commit()
    conn.close()

def get_latest_snapshot(url: str) -> tuple[str, str] | None:
    """Returns (content_hash, markdown) or None if no snapshot exists."""
    conn = sqlite3.connect(DB_PATH)
    row = conn.execute(
        "SELECT content_hash, markdown FROM snapshots WHERE url = ? ORDER BY captured_at DESC LIMIT 1",
        (url,)
    ).fetchone()
    conn.close()
    return row

def save_snapshot(url: str, markdown: str):
    content_hash = hashlib.sha256(markdown.encode()).hexdigest()
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "INSERT INTO snapshots (url, content_hash, markdown, captured_at) VALUES (?, ?, ?, ?)",
        (url, content_hash, markdown, datetime.utcnow().isoformat())
    )
    conn.commit()
    conn.close()
    return content_hash

import httpx
import os

UNWEB_API_KEY = os.environ["UNWEB_API_KEY"]

async def fetch_markdown(url: str) -> str:
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.get(
            "https://api.unweb.info/v1/convert",
            params={"url": url, "format": "markdown"},
            headers={"Authorization": f"Bearer {UNWEB_API_KEY}"},
        )
        response.raise_for_status()
        markdown = response.json().get("markdown", "")
    if len(markdown) < 100:
        raise ValueError(f"Page content too short — possible bot block: {url}")
    return markdown

def compute_diff(old_markdown: str, new_markdown: str) -> str:
    """Returns a unified diff of the two Markdown versions."""
    old_lines = old_markdown.splitlines(keepends=True)
    new_lines = new_markdown.splitlines(keepends=True)
    diff_lines = list(difflib.unified_diff(
        old_lines, new_lines,
        fromfile="previous",
        tofile="current",
        n=3  # context lines
    ))
    return "".join(diff_lines)

import anthropic
import json

def classify_change(url: str, diff: str, watch_for: str = "any meaningful content change") -> dict:
    """
    Returns {"significant": bool, "summary": str, "details": str}
    watch_for: natural language description of what to watch for
    Examples: "price changes", "availability changes", "new job postings"
    """
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    message = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""Analyze this diff from a website ({url}) and determine if it represents a significant change.

I'm watching for: {watch_for}

Diff:
{diff[:4000]}

Return a JSON object with:
- "significant": true if this is a meaningful change I should know about, false if it's noise (minor rewording, whitespace, dynamic content like ads or timestamps)
- "summary": one sentence describing what changed (or "No significant change" if not significant)
- "details": 2-3 sentences with specifics if significant, empty string if not

Return only the JSON object."""
        }]
    )

    raw = message.content[0].text.strip()
    if raw.startswith("```"):
        raw = raw.split("```")[1]
        if raw.startswith("json"):
            raw = raw[4:]
    return json.loads(raw.strip())

import smtplib
from email.mime.text import MIMEText

def send_slack_alert(webhook_url: str, message: str):
    import urllib.request
    payload = json.dumps({"text": message}).encode()
    req = urllib.request.Request(
        webhook_url,
        data=payload,
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)

def send_email_alert(to_email: str, subject: str, body: str):
    smtp_host = os.environ.get("SMTP_HOST", "smtp.gmail.com")
    smtp_port = int(os.environ.get("SMTP_PORT", "587"))
    smtp_user = os.environ["SMTP_USER"]
    smtp_pass = os.environ["SMTP_PASS"]

    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["From"] = smtp_user
    msg["To"] = to_email

    with smtplib.SMTP(smtp_host, smtp_port) as server:
        server.starttls()
        server.login(smtp_user, smtp_pass)
        server.send_message(msg)

import asyncio
import schedule
import time

# Configure what to monitor
WATCH_LIST = [
    {
        "url": "https://competitor.com/pricing",
        "watch_for": "pricing changes or new plan tiers",
        "alert_email": "team@company.com",
    },
    {
        "url": "https://jobs.company.com/engineering",
        "watch_for": "new job postings added or positions removed",
        "slack_webhook": os.environ.get("SLACK_WEBHOOK_URL"),
    },
    {
        "url": "https://status.dependency.com",
        "watch_for": "service status changes or new incidents",
        "alert_email": "oncall@company.com",
    },
]

async def check_url(config: dict):
    url = config["url"]
    watch_for = config.get("watch_for", "any meaningful content change")

    try:
        new_markdown = await fetch_markdown(url)
    except Exception as e:
        print(f"[FETCH ERROR] {url}: {e}")
        return

    previous = get_latest_snapshot(url)
    new_hash = hashlib.sha256(new_markdown.encode()).hexdigest()

    if previous is None:
        # First run — save baseline
        save_snapshot(url, new_markdown)
        print(f"[BASELINE] Saved initial snapshot for {url}")
        return

    prev_hash, prev_markdown = previous

    if prev_hash == new_hash:
        print(f"[NO CHANGE] {url}")
        return

    # Content changed — compute diff and classify
    diff = compute_diff(prev_markdown, new_markdown)
    result = classify_change(url, diff, watch_for)

    if result["significant"]:
        # Save new snapshot
        save_snapshot(url, new_markdown)

        # Build alert message
        alert_msg = f"🔔 Change detected: {url}\n\n{result['summary']}\n\n{result['details']}"
        print(f"[ALERT] {alert_msg}")

        if "alert_email" in config:
            send_email_alert(config["alert_email"], f"Change detected: {url}", alert_msg)

        if "slack_webhook" in config and config["slack_webhook"]:
            send_slack_alert(config["slack_webhook"], alert_msg)
    else:
        print(f"[NOISE] {url}: {result['summary']}")
        # Don't save snapshot — wait for a significant change to update baseline

async def run_all_checks():
    tasks = [check_url(config) for config in WATCH_LIST]
    await asyncio.gather(*tasks)

def scheduled_check():
    asyncio.run(run_all_checks())

if __name__ == "__main__":
    init_db()
    print("Starting website monitor...")

    # Run immediately on startup
    scheduled_check()

    # Then run every 30 minutes
    schedule.every(30).minutes.do(scheduled_check)

    while True:
        schedule.run_pending()
        time.sleep(60)