mirror of
https://github.com/nexy7574/LCC-bot.git
synced 2024-09-20 02:26:32 +01:00
115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
import asyncio
|
|
import sys
|
|
from asyncio import Lock
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Dict, Optional, Union
|
|
|
|
import config
|
|
import discord
|
|
from discord.ext import commands
|
|
|
|
if TYPE_CHECKING:
|
|
from asyncio import Task
|
|
|
|
from uvicorn import Config, Server
|
|
|
|
|
|
__all__ = ("Bot", "bot")
|
|
|
|
|
|
# noinspection PyAbstractClass
|
|
class Bot(commands.Bot):
|
|
if TYPE_CHECKING:
|
|
web: Optional[Dict[str, Union[Server, Config, Task]]]
|
|
|
|
def __init__(self, intents: discord.Intents, guilds: list[int], extensions: list[str], prefixes: list[str]):
|
|
from .console import console
|
|
from .db import registry
|
|
|
|
super().__init__(
|
|
command_prefix=commands.when_mentioned_or(*prefixes),
|
|
debug_guilds=guilds,
|
|
allowed_mentions=discord.AllowedMentions.none(),
|
|
intents=intents,
|
|
max_messages=5000,
|
|
case_insensitive=True,
|
|
)
|
|
self.loop.run_until_complete(registry.create_all())
|
|
self.training_lock = Lock()
|
|
self.started_at = datetime.now(tz=timezone.utc)
|
|
self.console = console
|
|
self.incidents = {}
|
|
for ext in extensions:
|
|
try:
|
|
self.load_extension(ext)
|
|
except discord.ExtensionNotFound:
|
|
console.log(f"[red]Failed to load extension {ext}: Extension not found.")
|
|
except (discord.ExtensionFailed, OSError) as e:
|
|
console.log(f"[red]Failed to load extension {ext}: {e}")
|
|
if getattr(config, "dev", False):
|
|
console.print_exception()
|
|
else:
|
|
console.log(f"Loaded extension [green]{ext}")
|
|
|
|
if getattr(config, "CONNECT_MODE", None) == 2:
|
|
|
|
async def connect(self, *, reconnect: bool = True) -> None:
|
|
self.console.log("Exit target 2 reached, shutting down (not connecting to discord).")
|
|
return
|
|
|
|
async def on_error(self, event: str, *args, **kwargs):
|
|
e_type, e, tb = sys.exc_info()
|
|
if isinstance(e, discord.NotFound) and e.code == 10062: # invalid interaction
|
|
return
|
|
if isinstance(e, discord.CheckFailure) and "The global check once functions failed." in str(e):
|
|
return
|
|
await super().on_error(event, *args, **kwargs)
|
|
|
|
async def close(self) -> None:
|
|
await self.http.close()
|
|
if getattr(self, "web", None) is not None:
|
|
self.console.log("Closing web server...")
|
|
await self.web["server"].shutdown()
|
|
if hasattr(self, "web"):
|
|
self.web["task"].cancel()
|
|
self.console.log("Web server closed.")
|
|
try:
|
|
await self.web["task"]
|
|
except asyncio.CancelledError:
|
|
pass
|
|
del self.web["server"]
|
|
del self.web["config"]
|
|
del self.web["task"]
|
|
del self.web
|
|
try:
|
|
await super().close()
|
|
except asyncio.TimeoutError:
|
|
self.console.log("Timed out while closing, forcing shutdown.")
|
|
sys.exit(1)
|
|
self.console.log("Finished shutting down.")
|
|
|
|
|
|
try:
|
|
from config import intents as _intents
|
|
except ImportError:
|
|
_intents = discord.Intents.all()
|
|
|
|
try:
|
|
from config import extensions as _extensions
|
|
except ImportError:
|
|
_extensions = [
|
|
"jishaku",
|
|
]
|
|
for file in Path("cogs").glob("*.py"):
|
|
if file.name.startswith(("_", ".")):
|
|
continue
|
|
_extensions.append(f"cogs.{file.stem}")
|
|
|
|
try:
|
|
from config import prefixes as _prefixes
|
|
except ImportError:
|
|
_prefixes = ("h!", "r!")
|
|
|
|
|
|
bot = Bot(_intents, config.guilds, _extensions, list(_prefixes))
|