Monitor for Sundance Film Festival Tickets

check_sundance_tickets.py

import os, re, time, yaml
import requests

from datetime import datetime
from zoneinfo import ZoneInfo

import pandas as pd
import numpy as np

def parse_input(filepath = "sundance-checks-20260121.yaml"):

	with open(filepath, "r") as file:
		data = yaml.safe_load(file)


	config = {
		"sleep": data.get("config", {}).get("sleep", 30),
		"alert_sounds": data.get("config", {}).get("alert_sounds", 3)
	}

	checks = []
	for check in data.get("checks", []):
		try:
			if "query" not in check:
				raise Exception("No query included...")
				print("ERROR: Skipping check... no query...")

			query = ""

			if isinstance(check["query"], str):
				query = check["query"]
			elif isinstance(check["query"], list):
				query_parts = []
				for query_part in check["query"]:
					if isinstance(query_part, str):
						query_parts.append(query_part)
					elif isinstance(query_part, dict):
						if query_part.get("fn") == "name_in":
							assert "list" in query_part, "No list of names included in query_part..."
							query_parts.append(f"name in {query_part['list']}")
						elif query_part.get("fn") == "name_not_in":
							assert "list" in query_part, "No list of names included in query_part..."
							query_parts.append(f"name not in {query_part['list']}")
						elif "fn" not in query_part:
							raise Exception("No 'fn' included in query_part")
						else:
							raise Exception(f"Unrecognized 'fn' {query_part.get('fn')} in query_part")
					else:
						raise Exception(f"Query part must be string or dict, not {type(query_part)}")
				query = " and ".join(query_parts)

			checks.append({
				"query": query,
				"description": check.get("description", "matches"),
				"loud": check.get("loud", False)
			})

		except Exception as e:
			print(f"ERROR: Failed to parse check: {e}")
			print(check)

	return config, checks

def get_sundance_data(auth_token):
	req = requests.get("https://api.eventive.org/event_buckets/67a0f8057aa6eac072bf986b/events",
	    headers = {
	        "Authorization": auth_token
	    }
	)
	assert req.status_code == 200, "Request to Sundance API failed..."
	return req.json()

def parse_screenings(data):
	all_df = pd.DataFrame([{
	    "index": i,
	    "screening_id": e.get("id"),
	    "name": e.get("name"),
	    "category": "|".join([t.get("name") for t in e.get("tags", []) if "[C]" in t.get("name")]).replace("[C] ", ""),
	    "start_time": datetime.fromisoformat(e.get("start_time").replace("Z", "+00:00")).astimezone(ZoneInfo("America/Denver")).strftime("%Y-%m-%d %H:%M"),
	    "venue": "Virtual" if e.get("is_virtual") else e.get("location"),
	    # "location": re.search(r"(Park City|Salt Lake City)", data["events"][9].get("venue", {}).get("address"), flags = re.IGNORECASE).group(1),
	    "premiere_screening": any(["[S] Premiere" in t.get("name") for t in e.get("tags", [])]),
	    "opi_screening": any(["[S] Press and Industry Screening - Online" in t.get("name") for t in e.get("tags", [])]),
	    "total_quantity": e.get("quantity", 0),
	    "quantity_allocated": sum([tb.get("quantity", 0) for tb in e.get("ticket_buckets", [])]),
	    "quantity_sold": sum([tb.get("quantity_sold", 0) for tb in e.get("ticket_buckets", [])]),
	 } for i, e in enumerate(data["events"]) if e.get("quantity", 0) > 0])
	all_df["quantity_held"] = all_df["total_quantity"] - all_df["quantity_allocated"]
	all_df["quantity_available"] = all_df["quantity_allocated"] - all_df["quantity_sold"]
	all_df["percent_held"] = all_df["quantity_held"] / all_df["total_quantity"]
	return all_df

def refresh_data(auth_token = "Basic YOUR_AUTH_TOKEN"):
	data = get_sundance_data(auth_token = auth_token)
	df = parse_screenings(data)
	return df

def check_df_w_query(df, query, 
	description = "matches", 
	columns = ["name", "start_time", "venue", "quantity_available"],
	loud = False
):
	sdf1 = df.query(query)
	n1 = sdf1.shape[0]
	sdf2 = sdf1.query("quantity_available > 0")
	n2 = sdf2.shape[0]
	if n2 > 0:
		matches = "\n\t".join(
			sdf2[columns].astype(str).agg(lambda row: " | ".join(row), axis = 1).tolist()
		)
		return {
			"status": True,
			"text": f"{n2} {description} with available tickets... ({n2} of {n1} overall)\n\t{matches}",
			"loud": loud
		}
	else:
		return {
			"status": False,
			"text": f"No {description} with available tickets ({n2} of {n1} overall)",
			"loud": loud
		}

##########################
##########################
##########################

if __name__ == "__main__":

	log_file = datetime.now().strftime("logs/check_log_%Y%m%d-%H%M%S.log")

	while True:

		try:
			df = refresh_data()

			config, checks = parse_input()

			checked_checks = [
				check_df_w_query(
					df = df,
					query = check.get("query"),
					description = check.get("description"),
					loud = check.get("loud")
				) for check in checks
			]

			update_msg = "=================================================\n"
			update_msg += f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} STATUS\n"
			for check in checked_checks:
				update_msg += f"{('PASS 🔊🔊🔊' if check['loud'] else 'PASS') if check['status'] else 'FAIL'}: "
				update_msg += check['text'] + "\n"

			print(update_msg)

			with open(log_file, "a+") as f:
				f.write(update_msg)

			if any([check["status"] for check in checked_checks if check["loud"]]):
				for i in range(config.get("alert_sounds", 3)):
					os.system("afplay /System/Library/Sounds/Ping.aiff")

			print(f"Sleeping for {config.get('sleep', 30)} seconds...")
			time.sleep(config.get("sleep", 30))
		except Exception as e:
			error_msg = f"ERROR during check loop: {e}\n"

			print(error_msg)

			with open(log_file, "a+") as f:
				f.write(error_msg)
				
			print(f"Sleeping for {config.get('sleep', 30)} seconds before retry...")
			time.sleep(config.get("sleep", 30))

sundance-checks-20260121.yaml

config:
  sleep: 15       # how long to sleep (seconds)
  alert_sounds: 3 # how many times to beep (if loud = True)
  
checks:
  # monitor for any premieres of the named films
  - query:
    - fn: name_in
      list: 
        - I Want Your Sex
        - The Gallerist
        - The Moment
        - Buddy
        - The Invite
        - The History of Concrete
        - The Shitheads
        - The Weight
        - Wicker
        - Gail Daughtry and the Celebrity Sex Pass
    - premiere_screening
  description: premiere screenings of major interest
  loud: True
      
  # monitor for any available screenings on 2026-01-29
  - query:
    - start_time < '2026-01-30 00:00'
    - start_time > '2026-01-29 00:00'
    description: screenings on Jan 29
    
  # monitor for any screenings of The History of Concrete
  - query:
    - fn: name_in
      list:
      - The History of Concrete
    description: screenings of The History of Concrete (any)

  # monitor for any available tickets for zi online screening
  - query:
    - name == "zi"
    - venue == "Virtual"
    - not opi_screening
    description: virtual screenings of zi
    loud: True