kuh-no-bowls/attractions_api_logger.py

#!/usr/bin/env python3

import json
import logging
import os
import sys
from datetime import date, datetime, timedelta

import psycopg2
import requests
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()
LOG_PATH = os.environ.get('LOG_PATH', '/var/log/kuh-no-bowls.log')

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',
}

PARK_HOURS_QUERY = """
SELECT datestamp, _id, open, close, closed
FROM knoebels."LZ_attractions_io_resort"
ORDER BY datestamp DESC
LIMIT 1
"""

API_TIME_FMT = "%Y-%m-%d %H:%M:%S"


def send_to_discord(message):
    if not DISCORD_WEBHOOK:
        return
    try:
        response = requests.post(DISCORD_WEBHOOK, json={"content": message})
        if response.status_code != 204:
            logging.warning("Discord webhook returned %s: %s", response.status_code, response.text)
    except requests.exceptions.RequestException as e:
        logging.warning("Discord webhook error: %s", e)


def log_to_database(conn, table_name, data):
    if not data:
        return
    query = sql.SQL("INSERT INTO {table} VALUES ({values})").format(
        table=sql.Identifier(table_name),
        values=sql.SQL(', ').join(sql.Placeholder() * len(data[0])),
    )
    with conn.cursor() as cur:
        cur.executemany(query, data)
    conn.commit()


def fetch_attractions_payload():
    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',
    }
    response = requests.get(ATTRACTIONS_IO_ENDPOINT, headers=headers)
    response.raise_for_status()
    return response.json()


def _attraction_opening_times(raw):
    if not raw:
        return None, None
    try:
        parsed = json.loads(raw)
    except (json.JSONDecodeError, TypeError):
        return None, None
    return parsed.get('start'), parsed.get('end')


def pull_data_and_log(conn, log_park_hours):
    json_data = fetch_attractions_payload()

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

    opens_dt = datetime.strptime(park_opens, API_TIME_FMT)
    closes_dt = datetime.strptime(park_closes, API_TIME_FMT)

    if datetime.today().date() != opens_dt.date():
        logging.error('API not returning park hours for today yet. Exiting...')
        return

    if log_park_hours:
        logging.info('Inserting park hours into LZ_attractions_io_resort...')
        log_to_database(conn, 'LZ_attractions_io_resort',
                        [[datetime.now(), resort_id, park_opens, park_closes]])
        if not (opens_dt - timedelta(minutes=5)) <= datetime.now() <= (closes_dt + timedelta(minutes=5)):
            logging.info('Park not open now. Exiting...')
            return

    attraction_data = []
    for attraction in json_data['entities']['Item']['records']:
        start, end = _attraction_opening_times(attraction.get('OpeningTimes'))
        attraction_data.append([
            attraction['_id'],
            attraction.get('IsOperational'),
            attraction.get('QueueTime'),
            attraction.get('QueueStatusMessage'),
            attraction.get('IsOpen'),
            start,
            end,
            datetime.now(),
        ])

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

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

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


def main():
    logging.basicConfig(
        format='%(asctime)s %(levelname)s %(message)s',
        level=logging.DEBUG,
        filename=LOG_PATH,
        filemode="a",
    )
    logging.info('-' * 80)
    logging.info('Fetching most recent park hours entry from DB...')

    with psycopg2.connect(**db_params) as conn:
        with conn.cursor() as cur:
            cur.execute(PARK_HOURS_QUERY)
            row = cur.fetchone()

        if not row:
            logging.info('No park hours rows in DB — fetching from API')
            pull_data_and_log(conn, log_park_hours=True)
            return

        datestamp, _id, open_ts, close_ts, _closed = row
        logging.debug('%s - %s', open_ts, close_ts)

        if datestamp != date.today():
            logging.info('Park hours for today not in DB')
            pull_data_and_log(conn, log_park_hours=True)
            return

        logging.info('Park hours for today already in DB')
        if (open_ts - timedelta(minutes=5)) <= datetime.now() <= (close_ts + timedelta(minutes=5)):
            logging.info('Park is open! Pulling data from API...')
            pull_data_and_log(conn, log_park_hours=False)
        else:
            logging.info('Park not currently open. Exiting...')


if __name__ == '__main__':
    main()