nonsensebot/app/main.py

# import sys
import asyncio

import niobot
import tomllib
import platform
import logging
# import tortoise
from pathlib import Path

log = logging.getLogger(__name__)
logging.getLogger("nio.rooms").setLevel(logging.WARNING)
logging.getLogger("nio.crypto.log").setLevel(logging.WARNING)
logging.getLogger("peewee").setLevel(logging.WARNING)
logging.getLogger("nio.responses").setLevel(logging.WARNING)


with open("config.toml", "rb") as fd:
    config = tomllib.load(fd)


(store := (Path.cwd() / "store")).mkdir(parents=True, exist_ok=True)


class TortoiseIntegratedBot(niobot.NioBot):
    cfg: dict

    async def start(self, **kwargs):
        # url = config["database"].get("uri")
        # if not url:
        #     log.critical("No database URI specified in config.toml - using ephemeral sqlite.")
        #     url = "sqlite://:memory:"
        # sys.path.extend(("..", "."))
        # await tortoise.Tortoise.init(db_url=url, modules={"models": ["app.models"]})
        # await tortoise.Tortoise.generate_schemas()

        for file in (Path(__file__).parent / "./modules").glob("*.py"):
            if file.name.startswith("__"):
                log.warning("Skipping loading %s - dunder file.", file)
            name = "app.modules." + file.name[:-3]
            try:
                log.info("Loading %s...", name)
                bot.mount_module(name)
            except Exception:
                log.error("Failed to load %s!", name, exc_info=True)
            else:
                log.info("Loaded %s.", name)

        await super().start(**kwargs)


bot = TortoiseIntegratedBot(
    homeserver=config["bot"]["homeserver"],
    user_id=config["bot"]["user_id"],
    device_id=config["bot"].get("device_id", platform.node()),
    store_path=str(store.resolve()),
    command_prefix="h!",
    owner_id=config["bot"].get("owner_id") or "@nex:nexy7574.co.uk",
)
bot.cfg = config


@bot.on_event("ready")
async def on_ready(_):
    log.info("Bot has logged in.")


@bot.on_event("command")
async def on_command(ctx: niobot.Context):
    log.info("Command %s invoked by %s.", ctx.command, ctx.message.sender)


@bot.on_event("command_error")
async def on_command_error(ctx: niobot.Context, exc: Exception):
    exc = getattr(exc, "original", exc) or exc
    log.error("Command %s failed.", ctx.command, exc_info=exc)
    text = "\N{warning sign} Command failed: `%s`" % exc
    text = text.replace(bot.access_token, "REDACTED")
    await ctx.respond(
        text
    )


@bot.on_event("command_complete")
async def on_command_complete(ctx: niobot.Context, _):
    log.info("Command %s completed.", ctx.command)


if __name__ == "__main__":
    asyncio.run(bot.start(access_token=bot.cfg["bot"]["access_token"]))
    # tortoise.run_async(bot.start(access_token=bot.cfg["bot"]["access_token"]))