1
Fork 0
mirror of https://github.com/RGBCube/VReplBot synced 2025-05-14 05:24:58 +00:00
VReplBot/v_repl_bot/__init__.py
RGBCube 271706db7e B
2023-02-03 21:52:47 +03:00

89 lines
2.7 KiB
Python

from __future__ import annotations
__all__ = ("ReplBot",)
import asyncio
from inspect import cleandoc as strip
from itertools import chain
from pathlib import Path
from time import time as get_time
from traceback import format_exc as format_exception
from aiohttp import ClientSession as AIOHTTPSession
from discord import AllowedMentions, Game, Intents, Webhook
from discord.ext.commands import (
Bot as CommandsBot,
ExtensionFailed,
NoEntryPointError,
when_mentioned_or,
)
class ReplBot(CommandsBot):
ready_timestamp: float
log_webhook: Webhook
session: AIOHTTPSession
def __init__(self, *, token: str, webhook_url: str) -> None:
self.__token = token
self.__webhook_url = webhook_url
super().__init__(
command_prefix = when_mentioned_or("&"),
strip_after_prefix = True,
case_insensitive = True,
status = Game("in the V playground"),
allowed_mentions = AllowedMentions.none(),
max_messages = 100,
intents = Intents(
guilds = True,
messages = True,
message_content = True,
),
help_attrs = dict(
brief = "Sends help.",
help = "Sends all the commands of the bot, or help of a specific command or module.",
),
)
async def on_ready(self) -> None:
print(
strip(
f"""
Connected to Discord!
User: {self.user}
ID: {self.user.id}
"""
)
)
self.ready_timestamp = get_time()
async def load_extensions(self) -> None:
cogs = Path(__file__).parent / "cogs"
for file_name in chain(
map(
lambda file_path: ".".join(file_path.relative_to(cogs.parent.parent).parts)[:-3],
cogs.rglob("*.py"),
),
("jishaku",),
):
try:
await self.load_extension(file_name)
print(f"Loaded {file_name}")
except (ExtensionFailed, NoEntryPointError):
print(f"Couldn't load {file_name}:\n{format_exception()}")
def run(self) -> None:
async def runner() -> None:
async with self, AIOHTTPSession() as self.session:
self.log_webhook = Webhook.from_url(
self.__webhook_url, session = self.session, bot_token = self.__token
)
await self.load_extensions()
await self.start(self.__token)
try:
asyncio.run(runner())
except KeyboardInterrupt:
pass