F1 Event Notifications for Telegram and IRC

f1.py

import pickle
from collections import namedtuple
from datetime import datetime, timezone
import os.path
import pydle

import requests
from telegram import ParseMode
from telegram.ext import (Defaults, Updater)

F1_API_KEY = "[REDACTED]"
F1_API_ENDPOINT = "https://api.formula1.com/v1/event-tracker"
F1_SESSION_ENDPOINT = "https://livetiming.formula1.com/static/"

Driver = namedtuple("Driver", "position initials time difference team")

if os.path.isfile("sent.p"):
    sent = pickle.load(open("sent.p", "rb"))
else:
    sent = set()


def send_message(text: str):
    defaults = Defaults(parse_mode=ParseMode.HTML)
    updater = Updater(
        token='[REDACTED]', use_context=True, defaults=defaults
    )

    updater.bot.send_message(
        chat_id={YOUR_CHANNEL_ID},
        text=text
    )


def irc_send_message(irc_message: str):
    class F1Bot(pydle.Client):
        async def on_connect(self):
            await self.join('#f1')

        async def on_join(self, channel, _):
            await self.message(channel, irc_message)
            await self.quit()

    client = F1Bot('HamVerBot', realname='#f1 Bot',
                   sasl_username='HamVerBot', sasl_password='[REDACTED]')
    client.run('irc.libera.chat')


def results():
    r = requests.get(F1_SESSION_ENDPOINT + "SessionInfo.json")
    r.encoding = 'utf-8-sig'

    latest = r.json()
    if latest["ArchiveStatus"]["Status"] == "Complete" and latest["Path"] not in sent:
        r = requests.get(F1_SESSION_ENDPOINT + latest["Path"] + "SPFeed.json")
        r.encoding = 'utf-8-sig'

        result = r.json()["free"]["data"]
        drivers = []
        for driver in result["DR"]:
            d = driver["F"]
            drivers.append(
                Driver(int(d[3]), d[0], d[1], d[4], d[2])
            )

        text = f'<u>🏎️ {result["R"]}</u>\n'
        text += f'<b>{result["S"]} Results</b>'

        irc_message = f'🏎️ \x02{result["R"]}: {result["S"]} Results\x02:'

        for i, driver in enumerate(sorted(drivers, key=lambda x: x.position)):
            if i < 10:
                irc_message += f' {driver.position}. {driver.initials} - \x0303[{driver.time}]\x03'
            if i == 0:
                text += f'\n<pre> {driver.position}. {driver.initials} - {driver.time} <b>[+0.000]</b></pre>'
            elif i < 9:
                text += f'\n<pre> {driver.position}. {driver.initials} - {driver.time} <b>[{driver.difference}]</b></pre>'
            if i == 9:
                text += f'\n<pre>{driver.position}. {driver.initials} - {driver.time} <b>[{driver.difference}]</b></pre>'
                break

        send_message(text)
        irc_send_message(irc_message)
        sent.add(latest["Path"])


def next_event():
    headers = {
        "apikey": F1_API_KEY,
        "locale": "en",
    }

    data = requests.get(F1_API_ENDPOINT, headers=headers).json()

    for event in data["seasonContext"]["timetables"]:
        if event["state"] == 'upcoming':
            startTime = datetime.strptime(
                event["startTime"] + " " + event["gmtOffset"], "%Y-%m-%dT%H:%M:%S %z")
            difference = startTime - datetime.now(timezone.utc)
            if difference.days == 0 and difference.seconds // 60 < 5:
                text = f'🏎️ <b>{data["race"]["meetingOfficialName"]}</b>' \
                       f'\n{event["description"]} begins in 5 minutes.'
                send_message(text)

                irc_message = f'🏎️ \x02{data["race"]["meetingOfficialName"]}\x02: {event["description"]} begins in 5 minutes.'
                irc_send_message(irc_message)


next_event()
results()
pickle.dump(sent, open("sent.p", "wb"))