downgrangefixtures_ical/main.py

import pandas as pd
from icalendar import Calendar, Event
from datetime import datetime, timedelta
import uuid
import os
import csv
import requests
import lxml.html as lh

telegram_bot_api_key = USER = os.getenv('TELEGRAM_BOT_API_KEY')
telegram_bot_chat_id = USER = os.getenv('TELEGRAM_BOT_CHAT_ID')

def send_message(message:str)->None:
    """Send message to me on Telegram when updated.

    Args:
        message (str): String of message to send.
    """    
    requests.post(f'https://api.telegram.org/bot{telegram_bot_api_key}/sendMessage', json={'chat_id': telegram_bot_chat_id, 'text': message})  

def store_df_as_csv(df:pd.DataFrame, name:str)->None:
    """Store dataframe as a CSV file.

    Args:
        df (pd.DataFrame): Dataframe of fixtures.
        name (str): Name of the CSV file.
    """    
    df.to_csv(f'./{name}.csv', index=False)
def compare_dfs(df:pd.DataFrame, name:str)->bool:
    """Compare the latest DF with the stored DF for any changes.

    Args:
        df (pd.DataFrame): Latest copy of fixtures in dataframe
        name (str): Name of the CSV file.

    Returns:
        bool: True if match, False if no match.
    """    
    df2 = pd.read_csv(f'./{name}.csv')
    return df.equals(df2)

def write_calendar(cal:Calendar)->None:
    """Write the cal object to an ics file.

    Args:
        cal (Calendar): iCalendar object with all the ics details.
    """    
    f = open(os.path.join('./', 'fixtures.ics'), 'wb')
    f.write(cal.to_ical())
    f.close()

def does_csv_exist()->bool:
    """Check if the CSV file exists.

    Returns:
        bool: True if CSV file exists, False if not.
    """    
    return os.path.isfile('./fixtures.csv')

def make_ordinal(n:int)->str:
    '''
    Convert an integer into its ordinal representation::

        make_ordinal(0)   => '0th'
        make_ordinal(3)   => '3rd'
        make_ordinal(122) => '122nd'
        make_ordinal(213) => '213th'
    '''
    n = int(n)
    if 11 <= (n % 100) <= 13:
        suffix = 'th'
    else:
        suffix = ['th', 'st', 'nd', 'rd', 'th'][min(n % 10, 4)]
    return str(n) + suffix

def create_ical_file(df:pd.DataFrame, cal:Calendar, table:pd.DataFrame)->None:
    """Create an iCalendar file from a dataframe.

    Args:
        df (pd.DataFrame): Dataframe of fixtures.
        cal (Calendar): iCalendar object with all the ics details.
        table (pd.DataFrame): Dataframe of table details.
    """    
    for index, row in df.iterrows():
        event = Event()
        match_type = str(row['Type']) 
        home_team = str(row['Home Team'])
        if ("Tongham") not in home_team:
            home_team = str(row['Home Team']).replace(" U12","")
        away_team = str(row['Away Team.1'])            
        if ("Tongham") not in away_team:
            away_team = str(row['Away Team.1']).replace(" U12","")
        venue = str(row['Venue'])
        print(row['Date / Time'], home_team, away_team, venue)
        if row['Date / Time'] == 'TBC':
            continue
        start_date_time = datetime.strptime(row['Date / Time'], '%d/%m/%y %H:%M')
        # Set default 8am start time to normal 930 kickoff time.
        if start_date_time.hour == 8:
            start_date_time = start_date_time + timedelta(hours=1, minutes=30)
        # Arrival time is 30 mins before kickoff time.
        arrival_time = start_date_time + timedelta(minutes=-30)
        if match_type == 'L':
            summary = "(League) " + home_team + f" ({make_ordinal(table.loc[table['Team'] == home_team, 'POS'].iloc[0])})" + f" {str(row['Unnamed: 4'])} " + away_team + f" ({make_ordinal(table.loc[table['Team'] == away_team, 'POS'].iloc[0])})"
        else:
            summary = "(Cup) " + home_team + f" {str(row['Unnamed: 4'])} " + away_team            
        event.add('summary', summary)
        notes = row['Status / Notes']
        if pd.isna(notes):
            notes = 'No Match Notes'
        elif notes == 'Postponed':
            continue
        event.add('description', "Arrive by - " + str(arrival_time) + "\n" + notes + "\nTable -\n" + "https://fulltime.thefa.com/table.html?selectedSeason=19010414&selectedDivision=165601607")
        event.add('dtstart', start_date_time)
        # End 2 hours after start_date_time
        event.add('dtend', start_date_time + timedelta(hours=2))
        event.add('dtstamp', start_date_time)
        event.add('uid', str(uuid.uuid4()))
        event.add('location', venue)
        cal.add_component(event)
    write_calendar(cal)
    
def process_table(table_df:pd.DataFrame)->pd.DataFrame:
    table_df = table_df[:-1]
    table_df.drop(table_df.columns[len(table_df.columns)-1], axis=1, inplace=True)
    table_df['POS'] = table_df['POS'].astype('int')
    table_df['P'] = table_df['P'].astype('int')
    table_df['W'] = table_df['W'].astype('int')
    table_df['D'] = table_df['D'].astype('int')
    table_df['L'] = table_df['L'].astype('int')
    table_df['PTS'] = table_df['PTS'].astype('int')
    store_df_as_csv(table_df, "table")
    return table_df

def process_results()->None:
    req = requests.get("https://fulltime.thefa.com/results.html?selectedSeason=19010414&selectedFixtureGroupAgeGroup=11&selectedFixtureGroupKey=1_579285719&selectedRelatedFixtureOption=3&selectedClub=&selectedTeam=466317969&selectedDateCode=all&previousSelectedFixtureGroupAgeGroup=11&previousSelectedFixtureGroupKey=1_579285719&previousSelectedClub=")

    doc = lh.fromstring(req.text)
    headers = ['Date', 'Home Team', 'Score', 'Away Team']

    with open('results.csv', 'w', newline='') as fp:
        file = csv.writer(fp)    
        file.writerow(headers)    
        for idx,row in enumerate(doc.xpath("//div[contains(@id,'fixture')]"), start=1):
            date = row.xpath(f'/html[1]/body[1]/main[1]/div[2]/section[1]/div[1]/div[3]/div[1]/div[2]/div[{idx}]/div[1]/div[3]/a[1]/span[1]//text()')[0]
            home_team = row.xpath(f'/html[1]/body[1]/main[1]/div[2]/section[1]/div[1]/div[3]/div[1]/div[2]/div[{idx}]/div[1]/div[4]/div[1]/a[1]//text()')[0].strip()
            score = row.xpath(f'/html[1]/body[1]/main[1]/div[2]/section[1]/div[1]/div[3]/div[1]/div[2]/div[{idx}]/div[1]/div[5]//text()')[0].strip()
            if score == 'X - X':
                continue
            away_team = row.xpath(f'/html[1]/body[1]/main[1]/div[2]/section[1]/div[1]/div[3]/div[1]/div[2]/div[{idx}]/div[1]/div[6]/div[2]/a[1]//text()')[0].strip()
            file.writerow([date,home_team,score,away_team])

def compare_table():
    table_df = pd.read_html("https://fulltime.thefa.com/table.html?league=9268728&selectedSeason=19010414&selectedDivision=165601607&selectedCompetition=0&selectedFixtureGroupKey=1_579285719")[0]    
    store_df_as_csv(table_df, "base_table")
    return table_df

cal = Calendar()
cal.add('prodid', 'Down Grange Pumas Fixtures')
cal.add('version', '2.0')
fixtures_df = pd.read_html("https://fulltime.thefa.com/fixtures.html?selectedSeason=19010414&selectedFixtureGroupAgeGroup=11&selectedFixtureGroupKey=1_579285719&selectedDateCode=all&selectedClub=&selectedTeam=466317969&selectedRelatedFixtureOption=3&selectedFixtureDateStatus=&selectedFixtureStatus=&previousSelectedFixtureGroupAgeGroup=11&previousSelectedFixtureGroupKey=1_579285719&previousSelectedClub=&itemsPerPage=25")[0]
fixtures_df.head()
process_results()
table = compare_table()
exists = does_csv_exist()
if exists:
    fixtures_change = compare_dfs(fixtures_df, "fixtures")
    table_change = compare_dfs(table, "base_table")
    if not table_change:
        send_message("Table has updated")
    if not all([fixtures_change, table_change]):
        print("Data Updated, ical updated")
        store_df_as_csv(fixtures_df, "fixtures")
        create_ical_file(fixtures_df, cal, process_table(table))
        send_message("Fixtures updated, ical updated")
    else:
        print("No Data Updated, No update to ical")
else:
    store_df_as_csv(fixtures_df, "fixtures")
    create_ical_file(fixtures_df, cal, process_table(table))
    send_message("New ical file created")
    print("New ical file created")