142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
"""The actual appointment checking bot."""
|
|
|
|
import asyncio
|
|
import datetime
|
|
import logging
|
|
from typing import List
|
|
|
|
from selenium import webdriver
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.firefox.options import Options
|
|
from selenium.webdriver.support import expected_conditions as ec
|
|
from selenium.webdriver.support import wait
|
|
|
|
from .config import Config
|
|
from .matrixbot import MatrixBot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TLSAppointmentBot:
|
|
"""A bot for periodically checking the tls appointment website."""
|
|
|
|
def __init__(self, matrix_bot: MatrixBot, config: Config):
|
|
"""Set up the TLSAppointmentBot."""
|
|
self.matrix_bot = matrix_bot
|
|
self.config = config
|
|
|
|
options = Options()
|
|
options.add_argument("--headless")
|
|
options.add_argument("--disable-gpu")
|
|
self.driver = webdriver.Firefox(options=options)
|
|
|
|
async def run(self):
|
|
"""Run the TLSAppointmentBot."""
|
|
# Wait for the matrix bot to be ready
|
|
while not self.matrix_bot.initial_sync_done:
|
|
await asyncio.sleep(1)
|
|
|
|
while True:
|
|
try:
|
|
logger.info("checking for appointment slots")
|
|
while not self._is_logged_in():
|
|
logger.info("relogging since login expired")
|
|
self._login()
|
|
|
|
appointments = sorted(
|
|
filter(
|
|
lambda app: self.config.latest_appointment is None
|
|
or app < self.config.latest_appointment,
|
|
self._get_appointments(),
|
|
)
|
|
)
|
|
if appointments:
|
|
logger.info("found appointments, notifying")
|
|
await self.matrix_bot.send_appointments(appointments)
|
|
|
|
except Exception:
|
|
await self.matrix_bot.send_warning()
|
|
|
|
await asyncio.sleep(2 * 60)
|
|
|
|
def _is_logged_in(self) -> bool:
|
|
self.driver.get(f"{self.config.tls_instance}/BJS/index.php")
|
|
button = wait.WebDriverWait(self.driver, 10).until(
|
|
ec.visibility_of_element_located((By.CSS_SELECTOR, "a.nav-link.login"))
|
|
)
|
|
return button.text == "LOG OUT"
|
|
|
|
def _login(self):
|
|
self.driver.get(f"{self.config.tls_instance}/BJS/login.php")
|
|
[email, pwd, btn] = wait.WebDriverWait(self.driver, 10).until(
|
|
ec.visibility_of_all_elements_located(
|
|
(
|
|
By.XPATH,
|
|
"//input[@id='email']|//input[@id='pwd']|//input[@type='button']",
|
|
)
|
|
)
|
|
)
|
|
email.send_keys(self.config.tls_user)
|
|
pwd.send_keys((self.config.credentials_dir / "tls-password").read_text())
|
|
btn.click()
|
|
|
|
def _get_appointments(self) -> List[datetime.datetime]:
|
|
self.driver.get(f"{self.config.tls_instance}/BJS/myapp.php")
|
|
|
|
# Make sure we're on the page, and that we can see the
|
|
# appointment table
|
|
wait.WebDriverWait(self.driver, 10).until(
|
|
ec.visibility_of_element_located((By.XPATH, "//tr[@class='amend']"))
|
|
)
|
|
|
|
# If the user has already made an appointment, and is just
|
|
# trying to reschedule, there will be an "Amend" button we
|
|
# need to press first.
|
|
amend_button = self.driver.find_elements(
|
|
By.XPATH, "//input[@type='button' and @value='Amend']"
|
|
)
|
|
|
|
if amend_button:
|
|
wait.WebDriverWait(self.driver, 10).until(
|
|
ec.visibility_of_element_located((By.CLASS_NAME, "osano-cm-dialog"))
|
|
)
|
|
|
|
# Make sure the cookie banner doesn't potentially
|
|
# interfere with the page
|
|
self.driver.execute_script(
|
|
"""
|
|
let banner = document.getElementsByClassName("osano-cm-dialog")[0];
|
|
banner.parentNode.removeChild(banner);
|
|
"""
|
|
)
|
|
|
|
amend_button[0].click()
|
|
|
|
timetable = wait.WebDriverWait(self.driver, 10).until(
|
|
ec.visibility_of_element_located((By.ID, "timeTable"))
|
|
)
|
|
year = timetable.find_element(By.CLASS_NAME, "year-month-title").text.split()[1]
|
|
|
|
appointments: List[datetime.datetime] = []
|
|
for day in timetable.find_elements_by_class_name("inner_timeslot"):
|
|
date = day.find_element_by_class_name("appt-table-d")
|
|
slots = day.find_elements_by_css_selector(".appt-table-btn.dispo")
|
|
|
|
[month, num, _] = date.text.split()
|
|
|
|
logger.debug(f"{len(slots)} appointment slots on {month} {num}, {year}")
|
|
|
|
appointments.extend(
|
|
datetime.datetime.strptime(
|
|
f"{year}-{month}-{num}T{slot.text}", "%Y-%B-%dT%H:%M"
|
|
)
|
|
for slot in slots
|
|
)
|
|
|
|
return appointments
|
|
|
|
def quit(self):
|
|
"""Quit the TLSAppointmentBot, including cleaning up cookies."""
|
|
self.driver.delete_all_cookies()
|
|
self.driver.quit()
|