astrocade.py

"""Entrypoint for the Astrocade Discord bot."""

import asyncio
import logging
from os import name as OS_NAME
from pathlib import Path
from sys import stdout
from typing import Final

from arc import GatewayClient
from hikari import (
    Activity,
    ActivityType,
    ApplicationContextType,
    GatewayBot,
    GatewayConnectionError,
    Intents,
    Permissions,
    Status,
)
from loguru import logger
from loguru_discord import DiscordSink
from loguru_discord.intercept import Intercept

from core.consts import (
    DATABASE_PATH,
    DISCORD_BOT_TOKEN,
    DISCORD_SERVER_IDS,
    IS_DEBUG,
    LOG_DISCORD_WEBHOOK_LEVEL,
    LOG_DISCORD_WEBHOOK_URL,
    LOG_LEVEL,
)
from core.database import AsyncEngine, Database
from core.hooks import Hooks


async def start() -> None:
    """Initialize the Astrocade Discord Bot."""
    logger.info("Astrocade - Discord Activities Enhancements")
    logger.info("https://github.com/EthanC/Astrocade")

    if LOG_LEVEL:
        logger.remove()
        logger.add(stdout, level=LOG_LEVEL)

        logger.success(f"Set console logging level to {LOG_LEVEL}")

    Intercept.setup({"TRACE_HIKARI": "TRACE"})

    if LOG_DISCORD_WEBHOOK_URL:
        logger.add(
            DiscordSink(LOG_DISCORD_WEBHOOK_URL, suppress=[GatewayConnectionError]),
            level=LOG_DISCORD_WEBHOOK_LEVEL,
            backtrace=False,
        )

        logger.success(
            f"Enabled logging to Discord webhook at level {LOG_DISCORD_WEBHOOK_LEVEL}"
        )

    # Replace default asyncio event loop with libuv on UNIX
    # https://github.com/hikari-py/hikari#uvloop
    if OS_NAME != "nt":
        try:
            import uvloop  # type: ignore

            uvloop.install()

            logger.success("Installed libuv event loop")
        except Exception as e:
            logger.opt(exception=e).debug("Defaulted to asyncio event loop")

    database: AsyncEngine = await Database.setup(DATABASE_PATH)
    bot: GatewayBot = GatewayBot(
        DISCORD_BOT_TOKEN,
        allow_color=False,
        banner=None,
        suppress_optimization_warning=IS_DEBUG,
        intents=Intents.GUILD_MESSAGES
        | Intents.MESSAGE_CONTENT
        | Intents.GUILD_PRESENCES,
    )
    client: GatewayClient = GatewayClient(
        bot,
        default_enabled_guilds=DISCORD_SERVER_IDS,
        default_permissions=Permissions.VIEW_CHANNEL
        | Permissions.READ_MESSAGE_HISTORY
        | Permissions.SEND_MESSAGES,
        invocation_contexts=[ApplicationContextType.GUILD],
    )

    client.set_type_dependency(AsyncEngine, database)
    client.set_type_dependency(GatewayBot, bot)

    client.load_extensions_from("extensions")

    client.add_startup_hook(Hooks.client_startup)
    client.add_shutdown_hook(Hooks.client_shutdown)

    try:
        await bot.start(
            activity=Activity(
                # Later overriden by recurring presence task
                name="Pardon our space dust...",
                type=ActivityType.WATCHING,
            ),
            check_for_updates=False,
            status=Status.DO_NOT_DISTURB,
        )
        await bot.join()
    finally:
        await Database.close(database)


if __name__ == "__main__":
    try:
        asyncio.run(start())
    except KeyboardInterrupt:
        logger.debug("Exiting due to keyboard interrupt")
    except Exception as e:
        logger.opt(exception=e).critical("Fatal error occurred during runtime")