Overview
Festival resale tickets come and go fast, and good prices don’t last; manually refreshing Pukkelpop’s resale page all day isn’t realistic. This script does it instead: it polls the official resale pages for both combi and VIP combi tickets, parses out current listings and prices, and pushes a notification the moment something worth acting on appears, whether that’s a ticket under a target price, or the lowest available price dropping significantly.
Key features
- Polls Pukkelpop’s resale ticket pages on a randomized, jittered interval to avoid predictable, bot-like request patterns
- Parses listings and prices directly out of the page HTML, handling European number formatting (comma vs. period as decimal separator)
- Match alerts push a notification (via ntfy.sh to phone, and a native macOS notification) when a listing appears at or below a configured max price, with an option to auto-open the ticket page in the browser
- Price drop alerts track the lowest price seen per ticket type and send a dedicated alert when a new lowest price drops by at least a configured threshold (default: €10)
- Price history logging appends every newly seen listing to a local CSV, building a running history of resale pricing over time
- Daily summary, one push notification per day recapping how many listings were seen and the lowest price of the day
- Heartbeat notification every 6 hours confirming the watcher is still running
- State persistence for seen tickets, seen alerts, and price baselines, stored locally so restarting the script doesn’t cause duplicate CSV entries or repeated alerts
- Built-in test mode (
--test) that simulates the entire notification pipeline (match alert, price drop, heartbeat, and daily summary) without waiting for a real trigger
Challenges
- Reliably parsing prices out of loosely structured HTML, with inconsistent number formatting between sources (comma as decimal separator vs. thousands separator)
- Avoiding duplicate or noisy alerts across script restarts, which needed a small local state system layered on top of the CSV price log
- Balancing polling frequency against not hammering the site or looking like a bot; solved with randomized jitter on top of the base polling interval
What I learned
- Lightweight web scraping and HTML parsing with BeautifulSoup and regex on real-world, inconsistently formatted pages
- Building a small but resilient local automation tool: state persistence, deduplication, and a self-test mode
- Push notification integration via ntfy.sh, alongside native macOS notifications through terminal-notifier/osascript
- Designing polling behavior (jitter, heartbeats, daily summaries) that stays timely without being disrespectful to the target site
The code
The whole watcher is a single self-contained Python script. It’s folded away below so it stays out of the way; expand it to read the full source.
View the full source
#!/usr/bin/env python3
import argparse
import csv
import json
import os
import random
import re
import subprocess
import sys
import time
import webbrowser
from datetime import datetime
from urllib.parse import urljoin
import requests
from bs4 import BeautifulSoup
WATCH_URLS = [
"https://tickets.pukkelpop.be/nl/meetup/demand/?type=combi&camping=a&price=all",
"https://tickets.pukkelpop.be/nl/meetup/demand/?type=vip_combi&camping=a&price=all#tickets",
]
MAX_PRICE = 250
DROP_THRESHOLD = 10.0
POLL_SECONDS = 120
JITTER_SECONDS = 30
BASE = "https://tickets.pukkelpop.be"
NTFY_ENABLED = True
NTFY_TOPIC = "pkp-CHANGE-ME"
MACOS_NOTIFY_ENABLED = True
AUTO_OPEN_BROWSER = False
HEARTBEAT_HOURS = 6
CSV_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "price_history.csv")
COOKIE = ""
STATE_FILE = os.path.expanduser("~/.pukkelpop_watcher_seen.json")
DEBUG_HTML_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "last_page.html")
HEADERS = {
"User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0 Safari/537.36"),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "nl-BE,nl;q=0.9,en;q=0.8",
}
if COOKIE:
HEADERS["Cookie"] = COOKIE
PRICE_RE = re.compile(
r"€\s*([0-9]{1,4}(?:[.,][0-9]{2})?)"
r"|([0-9]{1,4}(?:[.,][0-9]{2})?)\s*(?:€|EUR)",
re.IGNORECASE,
)
def log(msg):
print(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {msg}", flush=True)
def to_float(raw):
raw = raw.strip().replace(" ", "")
if "," in raw and "." in raw:
if raw.rfind(",") > raw.rfind("."):
raw = raw.replace(".", "").replace(",", ".")
else:
raw = raw.replace(",", "")
elif "," in raw:
raw = raw.replace(",", ".") if re.search(r",\d{2}$", raw) else raw.replace(",", "")
try:
return float(raw)
except ValueError:
return None
def prices_in(text):
out = []
for m in PRICE_RE.finditer(text):
val = to_float(m.group(1) or m.group(2) or "")
if val is not None and 10 <= val <= 5000:
out.append(val)
return out
def parse_listings(html):
soup = BeautifulSoup(html, "html.parser")
listings = []
seen_urls = set()
for a in soup.find_all("a", href=True):
href = a["href"]
if "/meetup/buy/" not in href:
continue
if href in seen_urls:
continue
seen_urls.add(href)
container = a
found = []
for _ in range(4):
found = prices_in(container.get_text(" ", strip=True))
if found:
break
if container.parent is None:
break
container = container.parent
price = min(found) if found else None
label = a.get_text(" ", strip=True)[:80] or "listing"
listings.append({
"price": price,
"item_id": href,
"buy_url": urljoin(BASE, href),
"label": label,
})
return listings
def load_seen():
try:
with open(STATE_FILE) as f:
return set(json.load(f))
except (FileNotFoundError, json.JSONDecodeError):
return set()
def save_seen(seen):
try:
with open(STATE_FILE, "w") as f:
json.dump(sorted(seen), f)
except OSError as e:
log(f"warn: could not save state: {e}")
def load_tracker_seen():
seen = set()
if os.path.exists(CSV_FILE):
try:
with open(CSV_FILE, "r", encoding="utf-8") as f:
reader = csv.reader(f)
next(reader, None)
for row in reader:
if len(row) >= 4:
seen.add(row[3])
except Exception:
pass
return seen
def load_lowest_prices():
lowest = {}
if os.path.exists(CSV_FILE):
try:
with open(CSV_FILE, "r", encoding="utf-8") as f:
reader = csv.reader(f)
next(reader, None)
for row in reader:
if len(row) >= 3:
t_type = row[1]
try:
price = float(row[2])
if t_type not in lowest or price < lowest[t_type]:
lowest[t_type] = price
except ValueError:
pass
except Exception:
pass
return lowest
def log_price_to_csv(ticket_type, price, item_id, label):
file_exists = os.path.exists(CSV_FILE)
try:
with open(CSV_FILE, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(["timestamp", "type", "price", "item_id", "label"])
writer.writerow([datetime.now().strftime("%Y-%m-%d %H:%M:%S"), ticket_type, price, item_id, label])
except Exception as e:
log(f"warn: failed to write to CSV: {e}")
def notify_ntfy(price, page_url, label):
if not (NTFY_ENABLED and NTFY_TOPIC and "CHANGE-ME" not in NTFY_TOPIC):
return
try:
requests.post(
f"https://ntfy.sh/{NTFY_TOPIC}",
data=f"€{price:.0f}: {label}\nTap to open the tickets page, then press the price to reserve.".encode("utf-8"),
headers={
"Title": f"Pukkelpop ticket at {price:.0f} EUR",
"Priority": "5",
"Tags": "tickets,rotating_light",
"Click": page_url,
"Actions": f"view, Open tickets page, {page_url}, clear=true",
},
timeout=15,
)
log(" -> ntfy push sent (tickets page)")
except requests.RequestException as e:
log(f" -> ntfy failed: {e}")
def notify_price_drop(ticket_type, new_price, old_price, target_url):
drop = old_price - new_price
title = f"Price Drop Alert: {ticket_type.upper()}"
body = f"New lowest price: €{new_price:.0f} (Dropped by €{drop:.0f} from €{old_price:.0f})."
if MACOS_NOTIFY_ENABLED and sys.platform == "darwin":
if _has("terminal-notifier"):
subprocess.run(["terminal-notifier", "-title", f"📉 {title}", "-message", body, "-open", target_url], check=False)
else:
script = f'display notification "{body}" with title "📉 {title}" sound name "Glass"'
subprocess.run(["osascript", "-e", script], check=False)
if NTFY_ENABLED and NTFY_TOPIC and "CHANGE-ME" not in NTFY_TOPIC:
try:
requests.post(
f"https://ntfy.sh/{NTFY_TOPIC}",
data=body.encode("utf-8"),
headers={
"Title": title,
"Priority": "4",
"Tags": "chart_with_downwards_trend,moneybag",
"Click": target_url,
"Actions": f"view, Open demand page, {target_url}, clear=true",
},
timeout=15,
)
log(" -> Price drop push notification sent")
except requests.RequestException as e:
log(f" -> Price drop ntfy failed: {e}")
def notify_heartbeat():
if not (NTFY_ENABLED and NTFY_TOPIC and "CHANGE-ME" not in NTFY_TOPIC):
return
try:
requests.post(
f"https://ntfy.sh/{NTFY_TOPIC}",
data="Watcher is running smoothly and monitoring for tickets.".encode("utf-8"),
headers={"Title": "Pukkelpop Watcher Heartbeat", "Tags": "green_heart"},
timeout=15
)
log(" -> Sent heartbeat push")
except requests.RequestException:
pass
def send_daily_summary():
if not os.path.exists(CSV_FILE):
return
today_str = datetime.now().strftime("%Y-%m-%d")
count_combi = 0
count_vip = 0
lowest_price = float('inf')
try:
with open(CSV_FILE, "r", encoding="utf-8") as f:
reader = csv.reader(f)
next(reader, None)
for row in reader:
if len(row) >= 5:
dt_str, t_type, price_str, item_id, label = row
if dt_str.startswith(today_str):
if "vip" in t_type.lower():
count_vip += 1
else:
count_combi += 1
try:
p = float(price_str)
if p < lowest_price: lowest_price = p
except ValueError:
pass
except Exception as e:
log(f"CSV read error: {e}")
return
total = count_combi + count_vip
if total == 0:
return
msg = f"Tracked {total} new listings today (Combi: {count_combi}, VIP: {count_vip}). Lowest price seen today: €{lowest_price:.0f}."
if not (NTFY_ENABLED and NTFY_TOPIC and "CHANGE-ME" not in NTFY_TOPIC):
log(f"Daily Summary: {msg}")
return
try:
requests.post(
f"https://ntfy.sh/{NTFY_TOPIC}",
data=msg.encode("utf-8"),
headers={"Title": "Pukkelpop Daily 20h Summary", "Tags": "bar_chart"},
timeout=15
)
log(" -> Sent daily 20h summary push")
except requests.RequestException:
pass
def notify_macos(price, page_url, label):
if not (MACOS_NOTIFY_ENABLED and sys.platform == "darwin"):
return
title = f"Pukkelpop ticket €{price:.0f}"
if _has("terminal-notifier"):
subprocess.run(["terminal-notifier", "-title", title,
"-message", label, "-open", page_url], check=False)
else:
script = f'display notification "{label}" with title "{title}" sound name "Glass"'
subprocess.run(["osascript", "-e", script], check=False)
log(" -> macOS notification shown")
def _has(cmd):
return subprocess.run(["which", cmd], capture_output=True).returncode == 0
def check_url(url, alert_seen, tracker_seen, lowest_prices):
try:
r = requests.get(url, headers=HEADERS, timeout=25)
except requests.RequestException as e:
log(f"fetch error [{url}]: {e}")
return
if r.status_code != 200:
log(f"HTTP {r.status_code} on {url}.")
return
listings = parse_listings(r.text)
ticket_type = url.split("type=")[1].split("&")[0] if "type=" in url else "unknown"
if not listings:
log(f"[{ticket_type}] no listings parsed")
return
priced = [l for l in listings if l["price"] is not None]
cheapest = min((l["price"] for l in priced), default=None)
log(f"[{ticket_type}] {len(listings)} listings, {len(priced)} priced, "
f"cheapest = {('€%.0f' % cheapest) if cheapest else 'n/a'}")
if cheapest is not None:
prev_lowest = lowest_prices.get(ticket_type)
if prev_lowest is None:
lowest_prices[ticket_type] = cheapest
elif prev_lowest - cheapest >= DROP_THRESHOLD:
log(f"PRICE DROP WARNING [{ticket_type}]: €{cheapest:.0f} (was €{prev_lowest:.0f})")
notify_price_drop(ticket_type, cheapest, prev_lowest, url)
lowest_prices[ticket_type] = cheapest
for l in priced:
if l["item_id"] not in tracker_seen:
log_price_to_csv(ticket_type, l["price"], l["item_id"], l["label"])
tracker_seen.add(l["item_id"])
unique_id = f"{url}#{l['item_id']}"
if l["price"] <= MAX_PRICE and unique_id not in alert_seen:
log(f"MATCH €{l['price']:.0f} on [{ticket_type}]: {url}")
if AUTO_OPEN_BROWSER:
log(" -> Auto-opening tickets page (safe, does NOT reserve)...")
webbrowser.open(url)
notify_ntfy(l["price"], url, l["label"])
notify_macos(l["price"], url, l["label"])
alert_seen.add(unique_id)
save_seen(alert_seen)
def fetch_one_real_listing():
for url in WATCH_URLS:
try:
r = requests.get(url, headers=HEADERS, timeout=25)
except requests.RequestException as e:
log(f" -> fetch error while looking for a real listing: {e}")
continue
if r.status_code != 200:
log(f" -> HTTP {r.status_code} while looking for a real listing")
continue
listings = parse_listings(r.text)
ticket_type = url.split("type=")[1].split("&")[0] if "type=" in url else "unknown"
priced = [l for l in listings if l["price"] is not None]
pick = priced[0] if priced else (listings[0] if listings else None)
if pick:
return url, pick["price"], pick["label"], ticket_type
return None
def run_test():
log("=== RUNNING ALL NOTIFICATION TESTS ===")
test_url = WATCH_URLS[0]
log("1/7 Logging test entry to CSV...")
log_price_to_csv("combi", 150.0, "/test/simulated-ticket-1", "Test Ticket Simulation")
time.sleep(1)
log("2/7 Testing Auto-Open Browser (tickets page only)...")
if AUTO_OPEN_BROWSER:
webbrowser.open(test_url)
time.sleep(1.5)
else:
log(" -> AUTO_OPEN_BROWSER is off; skipping (safe default, nothing reserved).")
log("3/7 Testing Ticket Match Alert (macOS & ntfy, tickets page)...")
notify_macos(150.0, test_url, "TEST MATCH: Combi Ticket €150")
notify_ntfy(150.0, test_url, "TEST MATCH: Combi Ticket €150")
time.sleep(2)
log("4/7 Testing Price Drop Warning Alert...")
notify_price_drop("combi", 180.0, 200.0, test_url)
time.sleep(2)
log("5/7 Testing System Heartbeat Alert...")
notify_heartbeat()
time.sleep(2)
log("6/7 Testing Daily 20h Summary Alert...")
send_daily_summary()
time.sleep(2)
log("7/7 Fetching a REAL listing to test the tickets-page notification...")
real = fetch_one_real_listing()
if real:
page_url, price, label, ticket_type = real
price_str = f"€{price:.0f}" if price is not None else "€?"
log(f" -> Found real listing: {price_str} [{ticket_type}] -> {page_url}")
notify_ntfy(price if price is not None else 0.0, page_url, f"[REAL TEST] {label}")
notify_macos(price if price is not None else 0.0, page_url, f"[REAL TEST] {label}")
log(" -> Real push sent. Tap it ON YOUR PHONE to confirm it opens the tickets page.")
else:
log(" -> No real listings available right now (page empty / sold out). "
"Re-run the test when listings exist.")
log("=== TEST COMPLETE: Check your phone, Mac notifications, and price_history.csv ===")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--once", action="store_true", help="check a single time and exit")
ap.add_argument("--test", action="store_true", help="run system test")
args = ap.parse_args()
if args.test:
run_test()
return
log(f"Watching {len(WATCH_URLS)} ticket pages:")
for url in WATCH_URLS:
log(f" - {url}")
log(f"Alerting at <= €{MAX_PRICE:.0f}. Price drop threshold: >= €{DROP_THRESHOLD:.0f}. Ctrl-C to stop.")
alert_seen = load_seen()
tracker_seen = load_tracker_seen()
lowest_prices = load_lowest_prices()
if args.once:
for url in WATCH_URLS:
check_url(url, alert_seen, tracker_seen, lowest_prices)
return
last_heartbeat_time = time.time()
last_summary_date = datetime.now().date()
while True:
now = datetime.now()
if now.hour == 20 and last_summary_date != now.date():
send_daily_summary()
last_summary_date = now.date()
if time.time() - last_heartbeat_time > (HEARTBEAT_HOURS * 3600):
notify_heartbeat()
last_heartbeat_time = time.time()
for url in WATCH_URLS:
check_url(url, alert_seen, tracker_seen, lowest_prices)
wait = POLL_SECONDS + random.randint(-JITTER_SECONDS, JITTER_SECONDS)
time.sleep(max(30, wait))
if __name__ == "__main__":
main()