LoL-Calendar/main.py

#!/usr/bin/env python

import os
from datetime import datetime, timedelta

import requests
import bs4
from bs4 import BeautifulSoup, Tag

from init_api import get_service

import dotenv
import asyncio
import re


class Match:

    def __init__(self, team1: str, team2: str, score1: int, score2: int,
                 raw_time: str, stage: str, roundName: str, vods: list[str]):
        self.team1 = team1
        self.team2 = team2
        self.score1 = score1
        self.score2 = score2
        self.stage = stage
        self.roundName = roundName
        self.vods = vods
        self.time = datetime(*list(map(lambda v: int(v), raw_time.split(","))))
        self.eventId = None

    def get_format(self) -> int:
        if self.stage == "Play-In":
            if self.roundName == "Qualifiers" or self.roundName == "EMEA vs NA":
                return 5
            return 3
        if self.stage == "Swiss":
            if self.roundName in ["Round 1", "Round 2"]:
                return 1
            if self.roundName == "Round 3" and self.time.day == 6:
                return 1
            return 3
        if self.stage == "Knockout":
            return 5

    def event_data(self):
        def shorten(link: str):
            # remove all query parameters except t=
            return re.sub(r'([?&](?!t=))([^&=]+=[^&=]+)', r'\1', link)

        vods = "\r\n".join([f'Game {i + 1}: {shorten(link)}' for i, link in enumerate(self.vods)])

        return {
            "start": {
                "dateTime": self.time.isoformat() + "Z"
            },
            "end": {
                "dateTime": (self.time + timedelta(hours=self.get_format())).isoformat() + "Z"
            },
            "summary": f"{self.team1} - {self.team2}",
            "description": f"{self.stage} {self.roundName} BO{self.get_format()}\n{self.score1} - {self.score2}\n{vods}"
        }

    async def update_event(self):
        api = get_service()
        api.events().update(
            calendarId=os.environ["CALENDAR_ID"],
            eventId=self.eventId,
            body=self.event_data()
        ).execute()
        print(self)
        print(f"Event updated: {self.eventId}")
        print("")

    async def create_event(self):
        api = get_service()
        api.events().insert(
            calendarId=os.environ["CALENDAR_ID"],
            body=self.event_data()
        ).execute()
        print(self)
        print(f"Event created")
        print("")

    def is_in_calendar(self, calendar_events):
        for event in calendar_events:
            if datetime.fromisoformat(event["start"]["dateTime"][:-1]) == self.time:
                self.eventId = event["id"]
                return True
        return False

    def __repr__(self):
        return f"{self.team1:>3} - {self.time} - {self.team2:<3} | VODs: {len(self.vods)}"


def fetch_matches():
    sources = [
        "https://lol.fandom.com/wiki/2024_Season_World_Championship/Play-In",
        "https://lol.fandom.com/wiki/2024_Season_World_Championship/Main_Event"
    ]
    stages = ["Play-In", "Swiss", "Knockout"]

    matches: list[Match] = []

    for src in sources:
        fetched_data = requests.get(src).text
        soup = BeautifulSoup(fetched_data, "html.parser")

        match_lists: list[Tag] = soup.find_all(attrs={"class": "matchlist"})
        vodsTable: Tag = soup.find(attrs={"id", "md-table"}).find("tbody")

        for match_list in match_lists:
            roundName = match_list.find("tr").find("th").find(recursive=False, string=True).text
            raw_matches: list[Tag] = match_list.find_all(attrs={"class", "ml-row"})

            # somehow identify stage...
            stage = stages[0] if (src == sources[0]) else \
                (stages[1] if "Round" in roundName else stages[2])

            for raw_match in raw_matches:
                raw_time = raw_match.find(attrs={"class", "TimeInLocal"}).text
                scores = raw_match.find_all(attrs={"class", "matchlist-score"})
                scores = [0, 0] if not scores else tuple(map(lambda v: int(v.text), scores))
                score1 = scores[0]
                score2 = scores[1]
                team1 = raw_match.find(attrs={"class", "matchlist-team1"}).find(attrs={"class", "teamname"}).text
                team2 = raw_match.find(attrs={"class", "matchlist-team2"}).find(attrs={"class", "teamname"}).text

                vods = []
                if "TBD" not in [team1, team2]:
                    def is_current_match_row(tr):
                        both = tr.find_all(attrs={"class", "teamname"})
                        if len(both) == 2:
                            return both[0].text == team1 and both[1].text == team2
                        return False
                    roundStartRow: Tag = vodsTable.find(string=roundName).parent.parent
                    currentRow: Tag = roundStartRow.find_next_sibling(is_current_match_row)
                    for i in range(5):
                        fullLink = currentRow.find("a", string="PB")
                        if fullLink is not None:
                            vods.append(fullLink["href"])
                            currentRow = currentRow.find_next_sibling("tr")
                            if currentRow is None or len(currentRow.contents) > 5:
                                break
                        else:
                            break

                matches.append(Match(team1, team2, score1, score2, raw_time, stage, roundName, vods))

    return matches


def get_calendar_events():
    api = get_service()
    start = "2024-04-01T00:00:00Z"
    end = "2024-12-01T00:00:00Z"
    calendarId = os.environ.get("CALENDAR_ID")
    events = api.events().list(
        calendarId=calendarId, timeMin=start, timeMax=end,
        maxResults=2500, singleEvents=True,
        orderBy='startTime'
    ).execute()["items"]
    return events


def delete_calendar_events():
    api = get_service()
    calendarId = os.environ.get("CALENDAR_ID")
    events = get_calendar_events()
    for event in events:
        api.events().delete(calendarId=calendarId, eventId=event["id"]).execute()
    print("Deleted all calendar events")


async def update():
    dotenv.load_dotenv()

    events = get_calendar_events()

    tasks = []

    for match in fetch_matches():
        if match.is_in_calendar(events):
            tasks.append(match.update_event())
        else:
            tasks.append(match.create_event())

    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(update())