kuh-no-bowls/attractions_api_logger.py

#!/usr/bin/env python3

import os
from datetime import datetime, date, timedelta
import json, requests, psycopg2, logging, sys
from psycopg2 import sql

ATTRACTIONS_IO_ENDPOINT = "https://live-data.attractions.io/72f0ea9e-d196-508a-bee8-cce62c3228c7.json"
DISCORD_WEBHOOK = os.environ.get('DISCORD_WEBHOOK', '').strip()

DB_PASSWORD = os.environ.get('DB_PASSWORD')
if not DB_PASSWORD:
    sys.stderr.write("ERROR: DB_PASSWORD env var is required\n")
    sys.exit(1)

db_params = {
    'dbname': 'gp0',
    'user': os.environ.get('DB_USER', 'kuhnobowls'),
    'password': DB_PASSWORD,
    'host': os.environ.get('DB_HOST', '192.168.88.9'),
    'port': os.environ.get('DB_PORT', '5432'),
    'options': '-c search_path=knoebels'
}

LOG_PATH = os.environ.get('LOG_PATH', '/var/log/kuh-no-bowls.log')

park_hours_query = """
SELECT *
FROM knoebels."LZ_attractions_io_resort"
ORDER BY datestamp DESC
LIMIT 1
"""

def send_to_discord(message):
    if not DISCORD_WEBHOOK:
        # Discord notifications disabled — skip silently.
        return

    data = {
        "content": message
    }

    try:
        response = requests.post(DISCORD_WEBHOOK, json=data)

        if response.status_code == 204:
            print("Message sent successfully.")
        else:
            print(f"Failed to send message. Status code: {response.status_code}, Response: {response.text}")
    except requests.exceptions.RequestException as e:
        print(f"Error sending message: {e}")

def log_to_database(table_name, data):
    with psycopg2.connect(**db_params) as conn:
        with conn.cursor() as cursor:
            columns = len(data[0])
            query = sql.SQL("INSERT INTO {table} VALUES ({values})").format(
            table=sql.Identifier(table_name),
            values=sql.SQL(', ').join(sql.Placeholder() * columns)
            )

            cursor.executemany(query, data)
        conn.commit()

def generic_query(q):
    ret = []
    with psycopg2.connect(**db_params) as conn:
        with conn.cursor() as cursor:
            cursor.execute(q)
            rows = cursor.fetchall()
            col_names = [desc[0] for desc in cursor.description]

            for row in rows:
                ret.append(row)

    return ret

def fetch_webpage(url):
    HEADERS = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:124.0) Gecko/20100101 Firefox/124.0',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.5',
        'DNT': '1',
        'Sec-GPC': '1',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1',
        'Sec-Fetch-Dest': 'document',
        'Sec-Fetch-Mode': 'navigate',
        'Sec-Fetch-Site': 'none',
        'Sec-Fetch-User': '?1',
    }
    try:
        response = requests.get(url, headers=HEADERS)

        if response.status_code == 200:
            return response.json()
        else:
            return f"Error: {response.status_code} - Unable to fetch webpage"
    except Exception as e:
        return f"Error: {str(e)} - Unable to fetch webpage"

def pull_data_and_log(log_park_hours):
    json_data = fetch_webpage(ATTRACTIONS_IO_ENDPOINT)

    logging.debug('Parsing park hours...')
    try:
        park_opens = json.loads(json_data['entities']['Resort']['records'][0]['OpeningTimes'])['start']
        park_closes = json.loads(json_data['entities']['Resort']['records'][0]['OpeningTimes'])['end']
    except KeyError as e:
        logging.error('Unable to pull park hours. Is the park open today?')
        return

    logging.debug('Checking that park hours from API are actually occuring today...')
    if not datetime.today().date() == datetime.strptime(park_opens, "%Y-%m-%d %H:%M:%S").date():
        logging.error('API not returning park hours for today yet. Exiting...')
        return

    resort_data = [[datetime.now(), json_data['entities']['Resort']['records'][0]['_id'], park_opens, park_closes]]
    if log_park_hours:
        logging.info('Inserting park hours into LZ_attractions_io_resort...')
        log_to_database('LZ_attractions_io_resort', resort_data)
        if not (datetime.strptime(park_opens, "%Y-%m-%d %H:%M:%S") - timedelta(minutes=5)) <= datetime.now() <= (datetime.strptime(park_closes, "%Y-%m-%d %H:%M:%S") + timedelta(minutes=5)):
            logging.info('Park not open now. Exiting...')
            return

    attraction_data = []

    for attraction in json_data['entities']['Item']['records']:
        try:
            isOperational = attraction['IsOperational']
        except:
            isOperational = None
        try:
            QueueTime = attraction['QueueTime']
        except:
            QueueTime = None
        try:
            QueueStatusMessage = attraction['QueueStatusMessage']
        except:
            QueueStatusMessage = None
        try:
            IsOpen = attraction['IsOpen']
        except:
            IsOpen = None
        try:
            OpeningTimes = json.loads(attraction['OpeningTimes'])
            start = OpeningTimes['start']
            end = OpeningTimes['end']
        except:
            start = None
            end = None

        row = [attraction['_id'], isOperational, QueueTime, QueueStatusMessage, IsOpen, start, end, datetime.now()]
        attraction_data.append(row)

    queueline_data = []

    for queue in json_data['entities']['QueueLine']['records']:
        row = [datetime.now(), queue['_id'], queue['QueueTime']]
        queueline_data.append(row)

    logging.debug('Inserting rows into LZ_attractions_io...')
    log_to_database('LZ_attractions_io', attraction_data)

    logging.debug('Inserting rows into LZ_attractions_io_queuetimes...')
    log_to_database('LZ_attractions_io_queuetimes', queueline_data)


logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.DEBUG, filename=LOG_PATH, filemode="a")

logging.info('--------------------------------------------------------------------------------')
logging.info('Fetching most recent park hours entry from DB...')

park_hours_pull = generic_query(park_hours_query)[0]
logging.debug('%s - %s', park_hours_pull[2], park_hours_pull[3])

if park_hours_pull[2].date() == date.today():
    logging.info('Park hours for today already in DB')
    park_opens = park_hours_pull[2]
    park_closes = park_hours_pull[3]
    if (park_opens - timedelta(minutes=5)) <= datetime.now() <= (park_closes + timedelta(minutes=5)):
        logging.info('Park is open! Pulling data from API...')
        pull_data_and_log(False)
    else:
        logging.info('Park not currently open. Exiting...')
else:
    logging.info('Park hours for today not in DB')
    pull_data_and_log(True)