cellar scraper.py

import os
import re
from datetime import datetime, timedelta
from typing import Optional, List

import pytz
import requests
import sendgrid
from bs4 import BeautifulSoup
import json

from tabulate import tabulate

from app import kv
from app import create_app
from app.util import log_run, JsonDict, retrying_with_backoff
from app.log import log

FAVORITE_PERFORMERS = {
    "Andrew Schulz",
    "Bill Burr",
}
from_email = sendgrid.Email("notifications@email.co", "Notifications")


# via https://chat.openai.com/share/75f43b95-023b-4dca-975e-85022280dd20
def get_show_data(date: str, venue: str, show_type: str) -> Optional[JsonDict]:
    url = "https://www.comedycellar.com/lineup/api/"
    payload = {"action": "cc_get_shows", "json": json.dumps({"date": date, "venue": venue, "type": show_type})}
    headers = {"Content-Type": "application/x-www-form-urlencoded"}

    response = retrying_with_backoff(requests.post, fkwargs={"url": url, "data": payload, "headers": headers})

    # Checking if request was successful
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Error: {response.status_code}")
        print(response.content)
        return None


def format_show_data(api_response: Optional[JsonDict]) -> List[str]:
    if not api_response:
        return []

    # Getting HTML content from the response
    html_content = api_response.get("show", {}).get("html", "")

    # Parsing HTML content using BeautifulSoup
    soup = BeautifulSoup(html_content, "html.parser")
    to_return = []

    # Iterating over each set-header (show time and title)
    for set_header, make_reservation in zip(
        soup.find_all(class_="set-header"), soup.find_all(class_="make-reservation")
    ):
        time = set_header.find(class_="bold").text.strip().replace(" show", "")
        title = set_header.find(class_="title").text.strip()
        href_value = make_reservation.find("a").get("href")
        # while here it says 'showid' that's not actually what is later used for reservations -- in that page, CC
        # terms what we consider here a 'showid' as a 'show-timestamp'. Then, each of those are mapped to the actual
        # showid that is used in the HTTP POST call to make a reservation. As a concrete example:
        # 2023-10-10 @ 10:30pm show
        # URL from <a> link above: <div class="make-reservation"><a aria-label="Make a reservation for 10:30 pm Hot Soup in the FBPC." href="/reservations-newyork/?showid=1696991400">Make A Reservation</a></div>
        # Inspected element on actual reservation page: <li data-show-timestamp="1696991400" data-show-id="20037254"><p class="description">10:30pm Hot Soup</p><p class="cover">$20 Cover Charge</p></li>
        # Actual HTTP POST: curl 'https://www.comedycellar.com/reservations/api/addReservation' --data-raw '{"guest":{"email":"kevyn_baracuda@aleeas.com","firstName":"Kevyn","lastName":"Bacon","size":2,"phone":"704-404-7921","howHeard":"Other","smsOk":"No"},"showId":20037254,"date":"2023-10-10","settime":"22:30:00"}'
        show_timestamp = int(re.search(r"showid=(\d+)", href_value).group(1))  # type: ignore  # noqa: F841

        # Finding the lineup associated with the current set-header
        lineup_id = set_header.find(class_="lineup-toggle")["data-lineup-id"]
        lineup = soup.find(attrs={"data-set-content": lineup_id})

        fav_performers = []
        # Printing each performer in the lineup
        for performer in lineup.find_all(class_="name"):
            if performer.text.strip() in FAVORITE_PERFORMERS:
                fav_performers.append(performer.text.strip())

        if len(fav_performers) > 0:
            to_return.append(f"{api_response.get('date')}; {time}; {title}; {', '.join(fav_performers)}")

    return to_return


def main() -> None:
    now = datetime.utcnow()
    after_notifs = []

    today_nyc = datetime.now(pytz.timezone("America/New_York")).date()
    log(f"Beginning run for dates between {now} and {now + timedelta(days=29)}.")
    for i in range(30):
        date = today_nyc + timedelta(days=i)
        api_response = get_show_data(date.strftime("%Y-%m-%d"), "newyork", "lineup")
        after_notifs.extend(format_show_data(api_response))

    existing_notifs = (kv.get("COMEDIAN_NOTIFICATIONS") or "").split("\n")
    has_new_notifs = len(set(after_notifs).difference(existing_notifs)) > 0
    log(f"Found {len(existing_notifs)} existing notifications. Were there new notifications? {has_new_notifs}")

    if has_new_notifs:
        sg = sendgrid.SendGridAPIClient(api_key=kv.get("SENDGRID_API_KEY"))
        content = sendgrid.Content(
            "text/html",
            tabulate(
                [aa.split(";") for aa in after_notifs],
                headers=["Date", "Time", "Location", "Comedian"],
                tablefmt="html",
            ),
        )
        mail = sendgrid.Mail(
            from_email=from_email,
            to_emails=sendgrid.To("rsanek@gmail.com"),
            subject="🎭 Comedy Cellar Notification",
            html_content=content,
        )
        sg.client.mail.send.post(request_body=mail.get())
        kv.put("COMEDIAN_NOTIFICATIONS", "\n".join(after_notifs))
        log("Successfully sent notification email and updated the kv store key.")
    else:
        log("No notification email necessary.")

    log_run(os.path.basename(__file__), now, datetime.utcnow(), "SUCCESS")


if __name__ == "__main__":
    with create_app().app_context():
        main()