World Cup Slack Bot

cup.py

import requests
import os
import pathlib
import json
import pendulum
import hashlib
from time import sleep

UPDATE_INTERVAL = 60
SETTINGS = {
  "slack_instances": [
    {
      "webhook": "<slack_webhook_url>",
    }
  ],
  "slack_payload": {
    "username": "Sportsball 2.0",
    "icon_emoji": ":soccer:",
    "link_names": 1,
    "channel": "#worldcup"
  }
}

def num_to_word(num):
    nums = {
        0: ':zero:',
        1: ':one:',
        2: ':two:',
        3: ':three:',
        4: ':four:',
        5: ':five:',
        6: ':six:',
        7: ':seven:',
        8: ':eight:',
        9: ':nine:',
        10: ':ten:',
    }

    return nums[num] if num in nums else num


def code_to_flag(code):
    codes = {
        'EGY': ':flag-eg:', # egypt
        'RUS': ':flag-ru:', # russia
        'KSA': ':flag-sa:', # saudi arabia
        'URU': ':flag-uy:', # uruguay
        'IRN': ':flag-ir:', # iran
        'MAR': ':flag-ma:', # morocco
        'POR': ':flag-pt:', # portugal
        'ESP': ':flag-es:', # spain
        'AUS': ':flag-au:', # australia
        'DEN': ':flag-dk:', # denmark
        'FRA': ':flag-fr:', # france
        'PER': ':flag-pe:', # peru
        'ARG': ':flag-ar:', # argentina
        'CRO': ':flag-hr:', # croatia
        'ISL': ':flag-is:', # iceland
        'NGA': ':flag-ng:', # nigeria
        'BRA': ':flag-br:', # brazil
        'CRC': ':flag-cr:', # costa rica
        'SRB': ':flag-rs:', # serbia
        'SUI': ':flag-ch:', # switzerland
        'GER': ':flag-de:', # germany
        'MEX': ':flag-mx:', # mexico
        'KOR': ':flag-kr:', # south korea
        'SWE': ':flag-se:', # sweden
        'BEL': ':flag-be:', # belgium
        'ENG': ':flag-england:', # england
        'PAN': ':flag-pa:', # panama
        'TUN': ':flag-tn:', # tunisia
        'COL': ':flag-co:', # colombia
        'JPN': ':flag-jp:', # japan
        'POL': ':flag-pl:', # poland
        'SEN': ':flag-sn:', # senegal
    }

    return codes[code] if code in codes else code

def generate_output(matches):
    now = pendulum.now("UTC")
    now_day = now.day
    now_month = now.month
    now_hour = now.hour

    for match in matches:
        datetime = pendulum.parse(match["datetime"])
        day = datetime.day
        month = datetime.month
        hour = datetime.hour
        start_time = datetime.format("HH:mm")

        if now_month == month and now_day == day:
            venue = f'{match["location"]}, {match["venue"]}'

            hteam = match["home_team"]["country"]
            hcode = match["home_team"]["code"]
            hteamgoals = match["home_team"]["goals"]
            hgoals = num_to_word(hteamgoals)
            hflag = code_to_flag(hcode)

            ateam = match["away_team"]["country"]
            acode = match["away_team"]["code"]
            ateamgoals = match["away_team"]["goals"]
            agoals = num_to_word(ateamgoals)
            aflag = code_to_flag(acode)

            time = match["time"]

            yield(f':timer_clock: {start_time} UTC: {hflag} {hteam} vs {aflag} {ateam} @ {venue}')

            if now_hour == hour:
                yield(f':timer_clock: {hflag} {hteam} vs {aflag} {ateam} just started!')

            if hteamgoals > 0 or ateamgoals > 0:
                yield(f':recycle: Score update: {hflag} {hteam} {hgoals} vs {agoals} {aflag} {ateam}')

            if time == "half-time":
                yield(f':stopwatch: Half-time: {hflag} {hteam} {hgoals} vs {agoals} {aflag} {ateam}')

            if time == "full-time":
                yield(f':checkered_flag: Final score: {hflag} {hteam} {hgoals} vs {agoals} {aflag} {ateam}')

            for event in match["home_team_events"]:
                yield format_event_output(event, hflag, hteam)

            for event in match["away_team_events"]:
                yield format_event_output(event, aflag, ateam)



def format_event_output(event, flag, country):
    event_output = get_event_string(event)
    event_output = event_output.replace("[flag]", f'{flag}')
    event_output = event_output.replace("[country]", f'{country}')
    event_output = f':stopwatch: {event["time"]} {event_output}'
    return event_output


def get_event_string(event):
    player = event["player"]
    event_type = event["type_of_event"]

    event_types = {
        "goal": f'[flag] [country]: {player} just scored a goooooooal! :soccer:',
        "goal-own": f'[flag] [country]: Oh no, {player} just scored a goal on the wrong side of the field! :face_palm:',
        "goal-penalty": f'[flag] [country]: {player} gets a goal penalty :dart:',
        "yellow-card": f'[flag] [country]: {player} just received a yellow card :warning:',
        "yellow-card-second": f'[flag] [country]: {player} just received a SECOND yellow card :rotating_light:',
        "red-card": f'[flag] [country]: {player} just received a red card :rotating_light:',
        "substitution-out": f'[flag] [country]: {player} is coming out and going to be replaced :arrow_backward:',
        "substitution-in": f'[flag] [country]: {player} is coming onto the field as a substitute :arrow_forward:',
    }

    if event_type in event_types:
        return event_types[event_type]

    return ""

def monitor_output():
    matches = requests.get("http://worldcup.sfg.io/matches").json()
    outputs_filename = f'output.{pendulum.now("UTC").format("MM-DD")}.json'

    try:
        previous_outputs = json.loads(pathlib.Path(outputs_filename).read_text())
    except FileNotFoundError:
        previous_outputs = []

    for output in generate_output(matches):
        output_hash = sha1(output)
        if output_hash not in previous_outputs:
            send_to_slack(output)
            previous_outputs.append(output_hash)

    record_outputs(previous_outputs)


def record_outputs(outputs):
    outputs_filename = f'output.{pendulum.now("UTC").format("MM-DD")}.json'
    f = open(outputs_filename, "w")
    f.write(json.dumps(outputs))
    f.close()


def sha1(string):
    h = hashlib.sha1()
    h.update(string.encode())
    return h.hexdigest()


def send_to_slack(output):
    slacks = SETTINGS["slack_instances"]
    payload = dict(SETTINGS["slack_payload"])
    payload["text"] = output

    for slack in slacks:
        requests.post(slack["webhook"], data=json.dumps(payload), headers={'Content-Type': 'application/json'})

    print(output)

while True:
    monitor_output()
    sleep(UPDATE_INTERVAL)