Canada Citizenship Application Tracker Tracker

import yaml, time, os, dbus, logging, nio, asyncio, traceback
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from bs4 import BeautifulSoup
from datetime import datetime
import subprocess

def desktop_notify(date: datetime, summary: str) -> None:
    dir = os.path.dirname(os.path.realpath(__file__))
    icon = os.path.join(dir, 'Canada.png')
    if not os.path.exists(icon):
        icon = 'dialog-information'
    date_format = date.strftime('%d %B')
    if date_format.startswith('0'):
        date_format = date_format[1:]
    session = dbus.SessionBus()
    obj = session.get_object('org.freedesktop.Notifications', '/org/freedesktop/Notifications')
    interface = dbus.Interface(obj, 'org.freedesktop.Notifications')
    interface.Notify('Citizenship tracker', 0, icon, date_format + ' update', summary, [], {'urgency': 2}, 0)
    session.close()

def send_matrix_message(message: str, config: dict) -> None:
    async def main():
        client = nio.AsyncClient(f'https://' + config['matrix-home-server'], f"@{config['matrix-username']}:{config['matrix-home-server']}")
        await client.login(config['matrix-password'])
        await client.join(config['matrix-room'])
        await client.room_send(
            room_id=config['matrix-room'],
            message_type='m.room.message',
            content={'msgtype': 'm.text', 'body': message},
        )
        await client.close()
    asyncio.run(main())

if __name__ == '__main__':
    url = 'https://tracker-suivi.apps.cic.gc.ca/en/login'
    dir = os.path.dirname(os.path.realpath(__file__))
    with open(os.path.join(dir, 'config.yaml'), 'r') as f:
        config = yaml.safe_load(f)
    logging.basicConfig(filename=os.path.join(dir, 'tracker.log'), level=logging.INFO, format='%(asctime)s %(message)s')
    logging.info(f'Starting tracker scraper')

    if os.path.exists(os.path.join(dir, 'STOP')):
        logging.info(f'Exiting because of STOP file')
        raise SystemExit

    try:
        options = FirefoxOptions()
        options.add_argument('-headless')
        with webdriver.Firefox(options=options) as driver:
            driver.get(url)
            time.sleep(config['wait'])
            element = driver.find_element(By.CSS_SELECTOR, '#uci')
            element.send_keys(config['uci'])
            element = driver.find_element(By.CSS_SELECTOR, '#password')
            element.send_keys(config['password'])
            element = driver.find_element(By.CSS_SELECTOR, 'button[type="submit"]')
            element.click()
            time.sleep(config['wait'])
            html = driver.page_source

        soup = BeautifulSoup(html, 'html.parser')
        history = soup.select_one('#history-timeline')
        items: dict[datetime, str] = {}
        for item in history.children:
            if item.name == 'li' and 'activity' in item['class']:
                date = item.select_one('div.date').text
                date = datetime.strptime(date, '%b %d, %Y')
                summary = item.select_one('div.info div.activity-title').text.strip()
                items[date] = summary
        ceremony_status = ''
        try:
            h3 = soup.select_one('h3[aria-label="Citizenship ceremony"]')
            summary = h3.parent
            p = summary.select_one('p.chip-text')
            ceremony_status = p.text.strip()
        except:
            pass
    except Exception as e:
        tb = traceback.extract_tb(e.__traceback__)
        logging.info(f'Failed scraping due to {type(e).__name__} at line {tb[-1].lineno}')
        raise SystemExit

    updated = False
    for date, summary in items.items():
        if date.date() > config['last-updated']:
            logging.info(f'New update: {date.date()} {summary}')
            updated = True
            if config.get('notification-desktop'):
                desktop_notify(date, summary)
            if config.get('notification-matrix'):
                send_matrix_message(f'Citizenship tracker update: {date.date()} {summary}', config)

    if config.get('ceremony-started-check') and ceremony_status and ceremony_status != 'Not started':
        summary = f'Ceremony status changed: "{ceremony_status}"'
        logging.info(summary)
        updated = True
        if config.get('notification-desktop'):
            desktop_notify(datetime.now().date(), summary)
        if config.get('notification-matrix'):
            send_matrix_message(summary, config)

    if updated:
        with open(os.path.join(dir, 'STOP'), 'w') as f:
            pass
        if config.get('notification-sound'):
            try:
                subprocess.run([config['sound-player'], config['sound']])
            except:
                logging.info(f'Failed to play sound')
    else:
        item_count = len(items)
        if item_count:
            logging.info(f'No updates: {item_count} items found, the most recent update is from {max(items.keys()).date()}')
        else:
            logging.info(f'No updates: no items found')
    logging.info(f'Finished')