Divent/divent/bot.py

import json
import logging
from datetime import datetime, timedelta
from os import environ, path
from typing import Optional

from disnake import Client, Guild
from dotenv import load_dotenv
from ics import Calendar, Event
from ics.alarm import DisplayAlarm
from quart import Quart, redirect, render_template, request, url_for
import sentry_sdk
from sentry_sdk.integrations.quart import QuartIntegration

load_dotenv()

QUART_DEBUG = environ.get("QUART_DEBUG", False)
DISCORD_TOKEN = environ.get("DISCORD_TOKEN")
if not DISCORD_TOKEN:
    raise Exception("Missing DISCORD_TOKEN")

if QUART_DEBUG:
    logging.basicConfig(level=logging.DEBUG)

SENTRY_DSN = environ.get("SENTRY_DSN")
if SENTRY_DSN:
    sentry_sdk.init(SENTRY_DSN, integrations=[QuartIntegration()])


class Discord(Client):
    async def on_ready(self):
        print(f"Logged on as {self.user}!")


client = Discord()
app = Quart(__name__)


def get_guild_by_id(guild_id: str) -> Optional[Guild]:
    if guild_id:
        for guild in client.guilds:
            if str(guild.id) == guild_id or guild.vanity_url_code == guild_id:
                return guild
    return None


CATALOG_CACHE = {}


@app.errorhandler(500)
async def errorhandler(error: Exception):
    print(f"\33[31m{error}\33[m")
    return await render_template("error.html.j2", error=str(error)), 500


@app.errorhandler(404)
async def not_found(error: Exception):
    return await render_template("error.html.j2", error=str(error)), 404


def i18n(str: str) -> str:
    lang = request.accept_languages.best_match(["en", "fr"])

    if lang not in CATALOG_CACHE:
        catalog_file = f"{path.dirname(__file__)}/translations/{lang}.json"
        if path.exists(catalog_file):
            with open(catalog_file) as catalog_json:
                catalog = json.load(catalog_json)
                CATALOG_CACHE[lang] = catalog

    if lang in CATALOG_CACHE and str in CATALOG_CACHE[lang]:
        return CATALOG_CACHE[lang][str]

    return str


def days_before_failure() -> int:
    nextYear = datetime.today().year + 5 - ((datetime.today().year + 5) % 5)
    nextDate = datetime(year=nextYear, month=6, day=3)
    nextDelta = nextDate - datetime.now()

    return nextDelta.days


@app.context_processor
def context_processor():
    return dict(_=i18n, client=client, days_before_failure=days_before_failure())


@app.route("/")
async def index():
    guild_id = request.args.get("guild")
    guild = get_guild_by_id(guild_id)

    if guild:
        return redirect(url_for(".subscribe", guild_id=guild_id))

    return await render_template("index.html.j2")


@app.route("/subscribe/<guild_id>")
async def subscribe(guild_id: str):
    guild = get_guild_by_id(guild_id)
    if guild is None:
        return redirect(url_for(".index"))

    return await render_template("subscribe.html.j2", guild=guild)


@app.route("/<guild_id>.ics")
async def ical(guild_id: str):
    guild = get_guild_by_id(guild_id)
    if guild is None:
        return redirect(url_for(".index"))

    calendar = Calendar()

    for scheduled_event in guild.scheduled_events:
        event = Event()
        event.name = scheduled_event.name
        event.begin = scheduled_event.scheduled_start_time
        event.end = (
            scheduled_event.scheduled_end_time
            or scheduled_event.scheduled_start_time + timedelta(hours=2)
        )
        event.uid = str(scheduled_event.id)
        event.description = scheduled_event.description
        event.url = f"https://discord.com/events/{guild_id}/{scheduled_event.id}"
        event.location = (
            scheduled_event.entity_metadata.location
            if scheduled_event.entity_metadata
            else None
        )

        alarm = DisplayAlarm()
        alarm.trigger = timedelta(hours=1)
        event.alarms.append(alarm)

        calendar.events.append(event)

    return str(calendar)


quart_task = client.loop.create_task(app.run_task("0.0.0.0"))
quart_task.add_done_callback(lambda f: client.loop.stop())
client.run(DISCORD_TOKEN)