Autobooking for the gym

.gitignore

scratch
.idea
.vscode

book.py

import json
from html.parser import HTMLParser
import urllib.request
import urllib.parse
import http.cookiejar
import datetime
import argparse
import re
import logging
import time
from typing import Optional, Dict, List, Tuple

userAgent = "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:88.0) Gecko/20100101 Firefox/88.0"
httpHeaders = {"User-Agent": userAgent}

counter = 1


def write_log(html: bytes):
    global counter

    with open(f'/tmp/gym-{counter:02}.html', "w") as f:
        f.write(html.decode("utf-8"))
    counter = counter + 1


def log_state(data: Dict[str, any]):
    print("===============================================")
    for x in data:
        print(x, data[x])
    print("===============================================")


def get_attribute(attrs: list, name: str) -> Optional[str]:
    for x in attrs:
        if len(x) == 2 and x[0] == name:
            return x[1]


class SubmitLocator(HTMLParser):
    def __init__(self):
        super().__init__()
        self.submit = {}
        self.tags = ["h4", "td", "strong", "div"]
        self.strips = "\xa0\t\n\r "
        self.collecting = False
        self.strings = []

    def handle_starttag(self, tag: str, attrs: list):
        if tag == "input":
            the_type = get_attribute(attrs, "type")
            if the_type in ["submit", "button"]:
                name = get_attribute(attrs, "name")
                value = get_attribute(attrs, "value")
                self.submit[name] = value

        if tag in self.tags:
            self.collecting = True
        else:
            self.collecting = False

    def handle_endtag(self, tag: str):
        if tag in self.tags:
            self.collecting = False

    def handle_data(self, data):
        if self.collecting:
            value = data.strip(self.strips)
            if value:
                self.strings.append(value)


class ConfirmationParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.record = False
        self.counter = 0
        self.message = "#NA"

    def handle_starttag(self, tag: str, attrs: list):
        if tag == "div":
            aclass = get_attribute(attrs, "class")
            if aclass and aclass == "alert alert-info":
                self.record = True
                self.counter = 0

            if self.record:
                self.counter += 1

    def handle_data(self, data):
        if self.record:
            message = data.strip()
            if message:
                self.message = message

    def handle_endtag(self, tag: str):
        if tag == "div":
            if self.record:
                self.counter -= 1
                if self.counter == 0:
                    self.record = False


class StateParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.state = {}
        self.form = "aspnetForm"
        self.record = False
        self.submit = {}
        self.post_back = None
        self.set_status = None
        self.activity = None

    def handle_starttag(self, tag: str, attrs: list):
        if self.record:
            if tag == "input":
                name = get_attribute(attrs, "name")
                if name is not None:
                    the_type = get_attribute(attrs, "type")
                    if the_type == "checkbox":
                        checked = get_attribute(attrs, "checked")
                        if checked == "checked":
                            self.state[name] = "on"
                        else:
                            self.state[name] = "off"
                    else:
                        value = get_attribute(attrs, "value")
                        if value is not None:
                            self.state[name] = value
        else:
            if tag == "form":
                if get_attribute(attrs, "id") == self.form:
                    self.record = True

        if tag == "a":
            aid = get_attribute(attrs, "id")
            if aid:
                if "Classes" in aid:
                    if "lnkActivitySelect_lg" in aid:
                        href = get_attribute(attrs, "href")
                        m = re.search(r"javascript:__doPostBack\('(.*)',''\)", href)
                        if m:
                            self.post_back = m.group(1)
                    if "btnAvailability_lg" in aid:
                        self.set_status = True
                elif "Activities" in aid:
                    if "lnkActivitySelect_lg" in aid:
                        href = get_attribute(attrs, "href")
                        m = re.search(r"javascript:__doPostBack\('(.*)',''\)", href)
                        if m:
                            self.post_back = m.group(1)

    def handle_data(self, data):
        if self.post_back and data:
            activity = data.strip()
            self.activity = activity
            count = 1
            while self.activity in self.submit:
                self.activity = f'{activity}_{count}'
                count += 1
            self.submit[self.activity] = {'post_back': self.post_back}
            self.post_back = None
        if self.set_status:
            if data:
                self.submit[self.activity]['status'] = data.strip()
            self.set_status = False
            self.activity = None

    def handle_endtag(self, tag: str):
        if tag == "form":
            self.record = False
        if tag == "a":
            self.post_back = None
            self.set_status = False


def get_state(html: bytes) -> StateParser:
    parser = StateParser()
    parser.feed(html.decode("utf-8"))

    return parser


def get_submits(html: bytes) -> Tuple[Dict, List]:
    locator = SubmitLocator()
    locator.feed(html.decode("utf-8"))
    return locator.submit, locator.strings


def get_confirmation_message(html: bytes) -> str:
    confirmator = ConfirmationParser()
    confirmator.feed(html.decode("utf-8"))
    return confirmator.message


def add_login_data(data: Dict[str, any], user: str, pin: str):
    data["ctl00$MainContent$InputLogin"] = user
    data["ctl00$MainContent$InputPassword"] = pin
    data["ctl00$MainContent$btnLogin"] = "Login"


def add_search_data(data: Dict[str, any], date: datetime.date, source: str, links: str):
    str_date = date.strftime("%Y-%m-%d")

    data["ctl00$ScriptManager1"] = source

    data["ctl00$ctl11$SearchTextBox"] = ""
    data["ctl00$ctl11$searchTextBoxWaterMark_ClientState"] = ""
    data["ctl00$ctl11$SearchResult"] = ""

    data["ctl00$MainContent$_advanceSearchUserControl$_selectedSite"] = ""
    data["ctl00$MainContent$_advanceSearchUserControl$SitesSimple"] = "SAA"
    data["ctl00$MainContent$_advanceSearchUserControl$SitesAdvanced"] = "SAA"
    data["ctl00$MainContent$_advanceSearchUserControl$ActivityGroups"] = ""
    data["ctl00$MainContent$_advanceSearchUserControl$startDate"] = str_date
    data["ctl00$MainContent$_advanceSearchUserControl$_maskEditStartDate_ClientState"] = ""
    data["ctl00$MainContent$_advanceSearchUserControl$endDate"] = str_date
    data["ctl00$MainContent$_advanceSearchUserControl$_maskEditEndDate_ClientState"] = ""
    data["ctl00$MainContent$_advanceSearchUserControl$Activities"] = ""
    data["ctl00$MainContent$_advanceSearchUserControl$StartTime"] = "1"
    data["ctl00$MainContent$_advanceSearchUserControl$EndTime"] = "23"
    data["ctl00$MainContent$_advanceSearchUserControl$inputIsSearchByQuickLinks"] = links

    # if true, it updates the page
    # (which is harder to parse)
    # rather than returning a new one
    # data["__ASYNCPOST"] = "true"


def add_book_data(data: Dict[str, any]):
    data["ctl00$ctl11$SearchTextBox"] = ""
    data["ctl00$ctl11$searchTextBoxWaterMark_ClientState"] = ""
    data["ctl00$ctl11$SearchResult"] = ""


def download(url: str, data: Optional[Dict[str, any]] = None) -> bytes:
    request = urllib.request.Request(url, headers=httpHeaders)

    post = None
    if data:
        post = urllib.parse.urlencode(data).encode("ascii")

    response = urllib.request.urlopen(request, post)

    html = response.read()
    write_log(html)

    return html


def sleep_until(target_time: datetime.time):
    sod = datetime.datetime.combine(datetime.date.today(), target_time)
    now = datetime.datetime.now()
    if now < sod:
        time_to_sleep = sod - now
        sleep_seconds = time_to_sleep.total_seconds()
        logging.info(f'Sleeping for {sleep_seconds} seconds')
        time.sleep(sleep_seconds)


def add_button(data: Dict[str, any], submits: Dict, targets: List[str], name: str) -> bool:
    found = False
    for k, v in submits.items():
        if v in targets:
            found = True
        else:
            del data[k]

    if not found:
        logging.warning(f"Cannot find {targets} on '{name}'")

    return found


class Gym:
    opening_time = datetime.time(8, 0, 1)

    def __init__(self, args: argparse.Namespace):
        self.need_sleep = args.wait and (args.ahead == 7)

        for i in range(args.attempts):
            try:
                html = self.login(args)
                break
            except Exception as e:
                logging.warning(f"Attempt {i} failed {e}")
                pass
        else:
            # one more try which will raise the exception if it fails
            html = self.login(args)

        if not html:
            raise RuntimeError(f"Cannot login")

        self.login_data = get_state(html).state
        self.parser = None
        self.date = None

    def search(self, date: datetime.date, wait: bool):
        # === search by date
        logging.info(f"Searching by date: {date.strftime('%c')}")

        data = self.login_data.copy()

        add_search_data(data, date, "ctl00$ScriptManager1", "0")
        data["__EVENTTARGET"] = "ctl00$MainContent$_advanceSearchUserControl$_searchBtn"

        if self.need_sleep and wait:
            sleep_until(self.opening_time)

        # === extract activities
        logging.info("Getting list")
        html = download("https://booking.1life.co.uk/Connect/memberHomePage.aspx", data)

        self.parser = get_state(html)
        self.date = date

    def __enter__(self):
        return self

    def __exit__(self, exception_type, exception_value, traceback):
        self.logout()

    @staticmethod
    def login(args: argparse.Namespace) -> Optional[bytes]:
        # === get login page with cookies
        logging.info("Connecting")
        html = download("https://booking.1life.co.uk/Connect/mrmlogin.aspx?Culture=en-GB")

        # === post login and password
        logging.info(f"Login as '{args.username}'")
        data = get_state(html).state
        add_login_data(data, args.username, args.password)

        html = download("https://booking.1life.co.uk/Connect/mrmlogin.aspx", data)
        if "Invalid Email Address or PIN" in html.decode():
            logging.error(f"Cannot login as '{args.username}'")
            return None
        return html

    @staticmethod
    def logout():
        # === logout
        logging.info("Logout")
        html = download("https://booking.1life.co.uk/Connect/logout.aspx")

    def list(self):
        logging.info("Available activities:")
        for x, v in self.parser.submit.items():
            status = v.get('status', None)
            logging.info(f"'{x}' -> {status}")

    def get_activity(self, name: str) -> bytes:
        found = self.parser.submit[name]

        post_back = found['post_back']
        status = found.get('status', None)
        logging.info(f"Found: '{name}' -> {status}, {post_back}")

        data = self.parser.state

        add_search_data(data, self.date, "ctl00$ScriptManager1|" + post_back, "0")
        data["__EVENTTARGET"] = post_back

        html = download("https://booking.1life.co.uk/Connect/memberHomePage.aspx", data)
        return html

    # returns True if we need to show the list at the end
    def book(self, name: str, hour: Optional[str], dry_run: bool) -> bool:
        if name not in self.parser.submit:
            logging.warning(f"Cannot find '{name}'")
            return True

        html = self.get_activity(name)
        if not html:
            return True

        found = self.parser.submit[name]
        if 'status' in found:
            # redirected to https://booking.1life.co.uk/Connect/mrmClassStatus.aspx
            self.book_class(name, hour, dry_run, html)
        else:
            # redirected to https://booking.1life.co.uk/Connect/mrmProductStatus.aspx
            self.book_activity(name, hour, dry_run, html)

        return False

    def book_activity(self, name: str, hour: Optional[str], dry_run: bool, html: bytes):
        if hour is None:
            logging.warning(f"Activity 'name' needs a time")
            return

        submits, strings = get_submits(html)
        logging.info(f"Strings: {strings}")

        # === book
        logging.info(f"Booking: {name} @ {hour}")
        data = get_state(html).state
        add_book_data(data)

        found = add_button(data, submits, [hour], name)
        if not found:
            return

        html = download("https://booking.1life.co.uk/Connect/mrmProductStatus.aspx", data)
        # redirects to https://booking.1life.co.uk/Connect/mrmConfirmBooking.aspx

        self.confirm(dry_run, html)

    def book_class(self, name: str, hour: Optional[str], dry_run: bool, html: bytes):
        submits, strings = get_submits(html)
        logging.info(f"Strings: {strings}")

        if hour:
            if not any(hour in x for x in strings):
                logging.warning(f"Cannot match '{hour}' on '{name}'")
                return

        # === book
        msg = ", ".join(submits.values())
        logging.info(f"Booking: {msg}")
        data = get_state(html).state
        add_book_data(data)

        found = add_button(data, submits, ['Book', 'Add to waiting list'], name)
        if not found:
            logging.warning("Not open")
            return

        html = download("https://booking.1life.co.uk/Connect/mrmClassStatus.aspx", data)
        # redirects to https://booking.1life.co.uk/Connect/mrmConfirmBooking.aspx

        self.confirm(dry_run, html)

    def confirm(self, dry_run: bool, html: bytes):
        submits, strings = get_submits(html)
        logging.info(f"Strings: {strings}")

        confirm_button_name = "ctl00$MainContent$btnBasket"
        cancel_button_name = "ctl00$MainContent$btnCancel"

        if confirm_button_name not in submits:
            logging.warning("Cannot confirm")
            return

        # === confirm booking
        if dry_run:
            # one could cancel sending
            # ctl00$MainContent$btnCancel: Cancel
            return

        action = submits[confirm_button_name]

        logging.info(f"Confirm: {action}")
        data = get_state(html).state
        add_book_data(data)
        del data[cancel_button_name]
        del data[confirm_button_name]
        data["ctl00$MainContent$hidSort"] = ""
        data["__EVENTTARGET"] = confirm_button_name

        html = download("https://booking.1life.co.uk/Connect/mrmConfirmBooking.aspx", data)
        # redirects to https://booking.1life.co.uk/Connect/mrmBookingConfirmed.aspx

        message = get_confirmation_message(html)
        logging.info(f'Confirmation message: {message}')


def get_activities(args: argparse.Namespace, day: str) -> Dict[str, Optional[str]]:
    activities = {}
    if args.activity:
        for a in args.activity:
            activities[a] = None
    if args.conf:
        with open(args.conf, 'r') as f:
            conf = json.load(f)
            if day in conf:
                logging.info(f"Found configuration for '{day}'")
                activities.update(conf[day])
    return activities


def run(args: argparse.Namespace):
    date = datetime.date.today() + datetime.timedelta(days=args.ahead)
    day = date.strftime('%A').lower()
    activities = get_activities(args, day)

    if activities or args.list:
        https_handler = urllib.request.HTTPSHandler()

        cookie_jar = http.cookiejar.CookieJar()
        cookies = urllib.request.HTTPCookieProcessor(cookie_jar)

        grabber = urllib.request.build_opener(cookies, https_handler)

        urllib.request.install_opener(grabber)

        with Gym(args) as gym:
            show_list = args.list

            gym.search(date, bool(activities))

            if activities:
                logging.info(f'Processing {activities}')

                # === process a single activity
                for a, t in activities.items():
                    if gym.book(a.strip(), t, args.dry_run):
                        # if missing, remember and show list at the end
                        show_list = True
            else:
                show_list = True

            if show_list:
                gym.list()
    else:
        logging.info(f"Nothing to do for '{args.username}' on {day}")


def main():
    parser = argparse.ArgumentParser(description="Gym booking", fromfile_prefix_chars="@")

    parser.add_argument("--username", action="store", required=True, help="username")
    parser.add_argument("--password", action="store", required=True, help="password")
    parser.add_argument("--attempts", action="store", default=4, help="login attempts")
    parser.add_argument("--dry-run", action="store_true", default=False, help="do not confirm booking")
    parser.add_argument("--ahead", action="store", default=7, type=int, help="days ahead [7]")
    parser.add_argument("--conf", action="store", help="configuration file (.json)")
    parser.add_argument("--list", action="store_true", help="get list of activities")
    parser.add_argument("--wait", action="store_true", help="wait opening time")
    parser.add_argument("activity", nargs="*", help="activities")

    args = parser.parse_args()

    log_format = '%(asctime)s %(process)d %(levelname)s %(message)s'
    logging.basicConfig(format=log_format, level=logging.DEBUG)

    run(args)


if __name__ == "__main__":
    main()

first.json

{
  "monday": {
    "Body Blast": "09:00",
    "Indoor Cycling": "09:30",
    "Indoor Cycling_1": "09:30",
    "Indoor Cycling_2": "09:30",
    "Pilates": ":",
    "Pilates_1": ":"
  },
  "tuesday": {
    "Indoor Cycling": "10:30",
    "Indoor Cycling_1": "10:30",
    "Pilates": "13:30"
  },
  "wednesday": {
    "Body Pump": "09:30",
    "Body Pump_1": "09:30",
    "Pilates": "10:45",
    "Pilates_1": "10:45"
  },
  "thursday": {
    "Ashtanga Yoga": "12:00",
    "Hatha Yoga": "20:00"
  },
  "friday": {
    "The Firm": "09:30",
    "Pilates": "09:30",
    "Pilates_1": "09:30",
    "Pilates_2": "09:30",
    "Yin Yoga": "12:45"
  },
  "saturday": {
    "Indoor Cycling": "08:00",
    "Body Pump": "09:00"
  },
  "sunday": {
    "Pilates": "12:45"
  }
}

second.json

{
  "monday": {
    "Health Suite Session": "07:30"
  },
  "tuesday": {
    "Lane Swimming - 50 Mins": "19:30",
    "Lane Swimming - 50 Mins_1": "19:30",
    "Lane Swimming - 50 Mins_2": "19:30",
    "Lane Swimming - 50 Mins_3": "19:30",
    "Lane Swimming - 50 Mins_4": "19:30",
    "Lane Swimming - 50 Mins_5": "19:30",
    "Lane Swimming - 50 Mins_6": "19:30",
    "Lane Swimming - 50 Mins_7": "19:30",
    "Lane Swimming - 50 Mins_8": "19:30",
    "Lane Swimming - 50 Mins_9": "19:30"
  },
  "wednesday": {
    "Health Suite Session": "08:00"
  },
  "thursday": {
    "Lane Swimming - 50 Mins": "19:30",
    "Lane Swimming - 50 Mins_1": "19:30",
    "Lane Swimming - 50 Mins_2": "19:30",
    "Lane Swimming - 50 Mins_3": "19:30",
    "Lane Swimming - 50 Mins_4": "19:30",
    "Lane Swimming - 50 Mins_5": "19:30",
    "Lane Swimming - 50 Mins_6": "19:30",
    "Lane Swimming - 50 Mins_7": "19:30",
    "Lane Swimming - 50 Mins_8": "19:30",
    "Lane Swimming - 50 Mins_9": "19:30"
  },
  "friday": {
    "Health Suite Session": "07:30"
  },
  "saturday": {
    "Health Suite Session": "09:00"
  },
  "sunday": {
    "Lane Swimming - 50 Mins": "08:00",
    "Lane Swimming - 50 Mins_1": "08:00",
    "Lane Swimming - 50 Mins_2": "08:00",
    "Lane Swimming - 50 Mins_3": "08:00",
    "Lane Swimming - 50 Mins_4": "08:00"
  }
}