Build a Zillow Price-Change Alert Bot for Telegram

curl -s "https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/getUpdates" | python3 -m json.tool
# Look for result[].message.chat.id and export it:
export TELEGRAM_CHAT_ID="<the id you found>"

"""Check a list of Zillow links and send new price or status changes to Telegram.

Run it on a schedule, for example once a day from cron. Requires Python 3.10+ and requests.
Environment: PROPERTIESAPI_KEY, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID.
"""
import argparse
import json
import os
import sqlite3
import sys
from pathlib import Path

import requests

BATCH_URL = "https://api.propertiesapi.com/v1/zillow/batch"
TELEGRAM_URL = "https://api.telegram.org/bot{token}/sendMessage"

SCHEMA = """
CREATE TABLE IF NOT EXISTS observations (
  property_key TEXT NOT NULL,
  fetched_at   TEXT NOT NULL,
  price        REAL,
  home_status  TEXT,
  currency     TEXT,
  address      TEXT,
  detail_url   TEXT,
  PRIMARY KEY (property_key, fetched_at)
);
CREATE TABLE IF NOT EXISTS sent_alerts (
  alert_key TEXT PRIMARY KEY,
  sent_at   TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""


def load_watchlist(path):
    lines = Path(path).read_text(encoding="utf-8").splitlines()
    links = [line.strip() for line in lines if line.strip() and not line.startswith("#")]
    if not links:
        raise ValueError("The watchlist is empty")
    return list(dict.fromkeys(links))


def fetch(links, api_key):
    results = []
    for start in range(0, len(links), 50):
        response = requests.post(BATCH_URL, json={"items": links[start:start + 50]},
                                 headers={"x-api-key": api_key}, timeout=(10, 330))
        body = response.json()
        if response.status_code != 200:
            code = (body.get("error") or {}).get("code", response.status_code)
            raise RuntimeError(f"Batch request refused: {code}")
        results.extend(body["data"])
    return results


def observation(data, meta):
    return {"property_key": data.get("zpid") or data.get("detailUrl"),
            "fetched_at": meta.get("fetched_at"), "price": data.get("price"),
            "home_status": data.get("homeStatus"), "currency": data.get("currency"),
            "address": data.get("address"), "detail_url": data.get("detailUrl")}


def changes(previous, current):
    found = []
    if previous["home_status"] and current["home_status"] \
            and previous["home_status"] != current["home_status"]:
        found.append(("status", previous["home_status"], current["home_status"]))
    elif previous["price"] is not None and current["price"] is not None \
            and previous["price"] != current["price"]:
        found.append(("price", previous["price"], current["price"]))
    return found


def show(value):
    return f"{value:,.0f}" if isinstance(value, (int, float)) else str(value)


def message(obs, field, old, new):
    unit = f" {obs['currency']}" if field == "price" and obs["currency"] else ""
    return (f"{field.title()} change: {obs['address']}\n{show(old)}{unit} -> {show(new)}{unit}\n"
            f"Collected {obs['fetched_at']}\n{obs['detail_url']}")


def send(text, dry_run):
    if dry_run:
        print(text, end="\n\n")
        return
    url = TELEGRAM_URL.format(token=os.environ["TELEGRAM_BOT_TOKEN"])
    response = requests.post(url, timeout=30, json={
        "chat_id": os.environ["TELEGRAM_CHAT_ID"], "text": text,
        "link_preview_options": {"is_disabled": True}})
    if not response.json().get("ok"):
        raise RuntimeError(f"Telegram refused the message: {response.text[:200]}")


def process(results, db, dry_run):
    sent = failed = unchanged = 0
    for item in results:
        if not item.get("ok"):
            failed += 1
            print(f"item {item['index']}: {item['error']['code']}", file=sys.stderr)
            continue
        obs = observation(item["data"], item["meta"])
        if not obs["property_key"] or not obs["fetched_at"]:
            failed += 1
            continue
        seen = db.execute("SELECT 1 FROM observations WHERE property_key = ? AND fetched_at = ?",
                          (obs["property_key"], obs["fetched_at"])).fetchone()
        if seen:
            unchanged += 1  # same collection as last time (for example, a cache hit)
            continue
        previous = db.execute(
            "SELECT * FROM observations WHERE property_key = ? AND fetched_at < ? "
            "ORDER BY fetched_at DESC LIMIT 1", (obs["property_key"], obs["fetched_at"])).fetchone()
        with db:
            for field, old, new in (changes(dict(previous), obs) if previous else []):
                alert_key = f"{obs['property_key']}|{field}|{old}|{new}|{obs['fetched_at']}"
                if db.execute("SELECT 1 FROM sent_alerts WHERE alert_key = ?", (alert_key,)).fetchone():
                    continue
                send(message(obs, field, old, new), dry_run)
                db.execute("INSERT INTO sent_alerts (alert_key) VALUES (?)", (alert_key,))
                sent += 1
            db.execute("INSERT INTO observations VALUES (:property_key, :fetched_at, :price, "
                       ":home_status, :currency, :address, :detail_url)", obs)
    return sent, failed, unchanged


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("watchlist", help="Text file with one Zillow link per line")
    parser.add_argument("--db", default="price_alerts.sqlite3")
    parser.add_argument("--from-file", help="Use a saved batch response instead of the API")
    parser.add_argument("--dry-run", action="store_true", help="Print alerts instead of sending")
    args = parser.parse_args()
    db = sqlite3.connect(args.db)
    db.row_factory = sqlite3.Row
    db.executescript(SCHEMA)
    if args.from_file:
        results = json.loads(Path(args.from_file).read_text(encoding="utf-8"))["data"]
    else:
        results = fetch(load_watchlist(args.watchlist), os.environ["PROPERTIESAPI_KEY"])
    sent, failed, unchanged = process(results, db, args.dry_run)
    print(f"alerts sent: {sent}, failed items: {failed}, no new collection: {unchanged}")
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())

# m h dom mon dow
17 7 * * * cd /opt/price-alerts && /usr/bin/python3 price_alerts.py watchlist.txt >> alerts.log 2>&1