1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
|
import asyncio
import logging
import socket
from typing import Optional
import discord
from aiohttp import AsyncResolver, ClientSession, TCPConnector
from async_rediscache import RedisSession
from discord import DiscordException, Embed
from discord.ext import commands
from bot import constants
log = logging.getLogger(__name__)
__all__ = ("Bot", "bot")
class Bot(commands.Bot):
"""
Base bot instance.
While in debug mode, the asset upload methods (avatar, banner, ...) will not
perform the upload, and will instead only log the passed download urls and pretend
that the upload was successful. See the `mock_in_debug` decorator for further details.
"""
name = constants.Client.name
def __init__(self, redis_session: RedisSession, **kwargs):
super().__init__(**kwargs)
self.http_session = ClientSession(
connector=TCPConnector(resolver=AsyncResolver(), family=socket.AF_INET)
)
self._guild_available = asyncio.Event()
self.redis_session = redis_session
self.loop.create_task(self.check_channels())
self.loop.create_task(self.send_log(self.name, "Connected!"))
@property
def member(self) -> Optional[discord.Member]:
"""Retrieves the guild member object for the bot."""
guild = self.get_guild(constants.Client.guild)
if not guild:
return None
return guild.me
async def close(self) -> None:
"""Close Redis session when bot is shutting down."""
await super().close()
if self.http_session:
await self.http_session.close()
if self.redis_session:
await self.redis_session.close()
def add_cog(self, cog: commands.Cog) -> None:
"""
Delegate to super to register `cog`.
This only serves to make the info log, so that extensions don't have to.
"""
super().add_cog(cog)
log.info(f"Cog loaded: {cog.qualified_name}")
def add_command(self, command: commands.Command) -> None:
"""Add `command` as normal and then add its root aliases to the bot."""
super().add_command(command)
self._add_root_aliases(command)
def remove_command(self, name: str) -> Optional[commands.Command]:
"""
Remove a command/alias as normal and then remove its root aliases from the bot.
Individual root aliases cannot be removed by this function.
To remove them, either remove the entire command or manually edit `bot.all_commands`.
"""
command = super().remove_command(name)
if command is None:
# Even if it's a root alias, there's no way to get the Bot instance to remove the alias.
return
self._remove_root_aliases(command)
return command
async def on_command_error(self, context: commands.Context, exception: DiscordException) -> None:
"""Check command errors for UserInputError and reset the cooldown if thrown."""
if isinstance(exception, commands.UserInputError):
context.command.reset_cooldown(context)
else:
await super().on_command_error(context, exception)
async def check_channels(self) -> None:
"""Verifies that all channel constants refer to channels which exist."""
await self.wait_until_guild_available()
if constants.Client.debug:
log.info("Skipping Channels Check.")
return
all_channels_ids = [channel.id for channel in self.get_all_channels()]
for name, channel_id in vars(constants.Channels).items():
if name.startswith("_"):
continue
if channel_id not in all_channels_ids:
log.error(f'Channel "{name}" with ID {channel_id} missing')
async def send_log(self, title: str, details: str = None, *, icon: str = None) -> None:
"""Send an embed message to the devlog channel."""
await self.wait_until_guild_available()
devlog = self.get_channel(constants.Channels.devlog)
if not devlog:
log.info(f"Fetching devlog channel as it wasn't found in the cache (ID: {constants.Channels.devlog})")
try:
devlog = await self.fetch_channel(constants.Channels.devlog)
except discord.HTTPException as discord_exc:
log.exception("Fetch failed", exc_info=discord_exc)
return
if not icon:
icon = self.user.avatar_url_as(format="png")
embed = Embed(description=details)
embed.set_author(name=title, icon_url=icon)
await devlog.send(embed=embed)
async def on_guild_available(self, guild: discord.Guild) -> None:
"""
Set the internal `_guild_available` event when PyDis guild becomes available.
If the cache appears to still be empty (no members, no channels, or no roles), the event
will not be set.
"""
if guild.id != constants.Client.guild:
return
if not guild.roles or not guild.members or not guild.channels:
log.warning("Guild available event was dispatched but the cache appears to still be empty!")
return
self._guild_available.set()
async def on_guild_unavailable(self, guild: discord.Guild) -> None:
"""Clear the internal `_guild_available` event when PyDis guild becomes unavailable."""
if guild.id != constants.Client.guild:
return
self._guild_available.clear()
async def wait_until_guild_available(self) -> None:
"""
Wait until the PyDis guild becomes available (and the cache is ready).
The on_ready event is inadequate because it only waits 2 seconds for a GUILD_CREATE
gateway event before giving up and thus not populating the cache for unavailable guilds.
"""
await self._guild_available.wait()
def _add_root_aliases(self, command: commands.Command) -> None:
"""Recursively add root aliases for `command` and any of its subcommands."""
if isinstance(command, commands.Group):
for subcommand in command.commands:
self._add_root_aliases(subcommand)
for alias in getattr(command, "root_aliases", ()):
if alias in self.all_commands:
raise commands.CommandRegistrationError(alias, alias_conflict=True)
self.all_commands[alias] = command
def _remove_root_aliases(self, command: commands.Command) -> None:
"""Recursively remove root aliases for `command` and any of its subcommands."""
if isinstance(command, commands.Group):
for subcommand in command.commands:
self._remove_root_aliases(subcommand)
for alias in getattr(command, "root_aliases", ()):
self.all_commands.pop(alias, None)
_allowed_roles = [discord.Object(id_) for id_ in constants.MODERATION_ROLES]
_intents = discord.Intents.default() # Default is all intents except for privileged ones (Members, Presences, ...)
_intents.bans = False
_intents.integrations = False
_intents.invites = False
_intents.typing = False
_intents.webhooks = False
redis_session = RedisSession(
address=(constants.RedisConfig.host, constants.RedisConfig.port),
password=constants.RedisConfig.password,
minsize=1,
maxsize=20,
use_fakeredis=constants.RedisConfig.use_fakeredis,
global_namespace="sir-lancebot"
)
loop = asyncio.get_event_loop()
loop.run_until_complete(redis_session.connect())
bot = Bot(
redis_session=redis_session,
command_prefix=constants.Client.prefix,
activity=discord.Game(name=f"Commands: {constants.Client.prefix}help"),
allowed_mentions=discord.AllowedMentions(everyone=False, roles=_allowed_roles),
intents=_intents,
)
|
