clist_webhook.py

import datetime
import json
import logging
import sched
import time
import urllib.parse

import humanize
import requests

CHECK_EVERY = 60 # seconds

CLIST_API_KEY = "<clist_username>:<clist_apikey>"
WEBHOOK_URL = "<discord_webhook_url>"

FILTER = True # customize filters at https://clist.by/settings/preferences/

REMIND_BEFORE = (
    datetime.timedelta(minutes=10),
    datetime.timedelta(minutes=30),
    datetime.timedelta(hours=1),
    datetime.timedelta(hours=3),
)


def get_contests(start_gte, start_lte):
    headers = {"Accept": "application/json", "Authorization": f"ApiKey {CLIST_API_KEY}"}
    params = {
        "start__gte": start_gte.isoformat(),
        "start__lte": start_lte.isoformat(),
        "order_by": "start",
        "filtered": "true" if FILTER else "false",
    }

    res = requests.get("https://clist.by/api/v1/contest", params=params, headers=headers)
    data = res.json()
    contests = data["objects"]

    while data["meta"]["next"]:
        res = requests.get(
            urllib.parse.urljoin("https://clist.by", data["meta"]["next"]),
            headers=headers,
        )
        data = res.json()
        contests.extend(data["objects"])

    for contest in contests:
        contest["start"] = datetime.datetime.fromisoformat(contest["start"] + "+00:00")
        contest["end"] = datetime.datetime.fromisoformat(contest["end"] + "+00:00")
        contest["duration"] = datetime.timedelta(seconds=contest["duration"])

    return contests


def send_embed(contest, delta=None):
    if delta is None:
        delta = contest["start"] - datetime.datetime.utcnow()

    message = {
        "embeds": [{
            "title": contest["event"],
            "description": f"Starts in {humanize.precisedelta(delta)}",
            "url": contest["href"],
            "author": {
                "name": contest["resource"]["name"],
                "url": f"http://{contest['resource']['name']}",
                "icon_url": urllib.parse.urljoin("https://clist.by", contest["resource"]["icon"]),
            },
            "footer": {
                "text": f"Duration: {humanize.precisedelta(contest['duration'])}",
            },
        }]
    }

    requests.post(WEBHOOK_URL, json=message)

    logging.info("Sent embed for '%s' with delta %s", contest["event"], delta)


def main():
    scheduler = sched.scheduler(time.time, time.sleep)

    contest_time = {}
    remind_events = {}

    while True:
        start_gte = datetime.datetime.now(datetime.timezone.utc)
        start_lte = start_gte + max(REMIND_BEFORE) + datetime.timedelta(minutes=10)

        for contest in get_contests(start_gte, start_lte):
            if contest["id"] in contest_time:
                if contest["start"] != contest_time[contest["id"]]:
                    for event in remind_events[contest["id"]]:
                        sched.cancel(event)
                    del contest_time[contest["id"]]
                    del remind_events[contest["id"]]
                    logging.info("Canceled all send_embed events for '%s'", contest["event"])
                else:
                    continue

            contest_time[contest["id"]] = contest["start"]
            remind_events[contest["id"]] = []

            for delta in REMIND_BEFORE:
                remind_time = contest["start"] - delta
                if remind_time > datetime.datetime.now(datetime.timezone.utc):
                    remind_events[contest["id"]].append(
                        scheduler.enterabs(remind_time.timestamp(), 0, send_embed, (contest, delta)))
                    logging.info(
                        "Scheduled send_embed event for '%s' with delta %s at %s",
                        contest["event"],
                        delta,
                        remind_time,
                    )

        scheduler.run(blocking=False)
        time.sleep(CHECK_EVERY)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    main()