Skip to content

API: main

main

Bot

Bases: ExtendedClient

Discord client with database, managers, handlers and i18n.

Parameters:

Name Type Description Default
logger Logger

Root logger used to create component specific child loggers.

required

Attributes:

Name Type Description
session ClientSession

Shared aiohttp session for outbound HTTP requests.

db

MongoDBAsyncORM instance for persistence.

detailed_help

Optional mapping for extended help entries.

setting_cache

In‑memory cache for frequently accessed settings.

ready

Indicates whether the bot finished the first on_ready cycle.

Source code in main.py
 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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
class Bot(ExtendedClient): 
    """Discord client with database, managers, handlers and i18n.

    Args:
        logger: Root logger used to create component specific child loggers.

    Attributes:
        session: Shared aiohttp session for outbound HTTP requests.
        db: MongoDBAsyncORM instance for persistence.
        detailed_help: Optional mapping for extended help entries.
        setting_cache: In‑memory cache for frequently accessed settings.
        ready: Indicates whether the bot finished the first on_ready cycle.
    """
    def __init__(self, logger: logging.Logger):
        intents = Intents.default()
        intents.message_content = True 

        super().__init__(intents=intents, logger=logger, command_prefix="j!", owner_ids=OWNER_IDS, help_command=None)

        self.session: aiohttp.ClientSession = None
        self.db = None
        self.detailed_help = {}
        self.setting_cache = {}
        self.ready = False

    async def setup_hook(self):
        """Initialize external services and internal subsystems before the bot is ready.

        This connects to MongoDB, prepares HTTP session, instantiates managers and handlers,
        registers default permission namespaces, loads modules, then offers slash command sync.
        """
        # Initialize MongoDB connection
        self.logger.info("Connecting to MongoDB...")
        try:
            self.db = MongoDBAsyncORM(uri=MONGODB_URI, db_name="GigaJoyce-Test")
            await self.db.create_index("members", [("id", 1), ("guildId", 1)], unique=True)
            self.db.members = self.db.get_collection("members")
            self.db.guilds = self.db.get_collection("guilds")
            self.db.users = self.db.get_collection("users")
            self.logger.info("Successfully connected to the database and collected data.")
        except Exception as e:
            self.logger.error(f"Failed to connect to MongoDB: {e}")
            return

        # Initialize HTTP session+
        self.session = aiohttp.ClientSession()

        # Initialize Managers
        self.logger.info("Initializing Managers...")
        self.guild_manager= GuildManager(self, self.logger)
        self.member_manager = MemberManager(self, self.logger)
        self.settings_manager = SettingsManager(self, self.logger)
        self.permission_manager = PermissionsManager(self, self.logger)
        self.logger.info("Managers initialized.")

        # Register default permission namespaces
        self.logger.info("Registering default permission namespaces...")
        self.permission_manager.register_node("Role.*", RolesNamespace)
        self.permission_manager.register_node("User.*", UsersNamespace)
        self.permission_manager.register_node("Channel.*", ChannelsNamespace)
        self.logger.info("Default permission namespaces registered.")

        # Initialize EmojiManager
        self.logger.info("Initializing EmojiManager...")
        translations_path = Path("./shared/emojis")
        self.emoji_manager = EmojiManager(self, translations_path, self.logger )
        self.logger.info("EmojiManager initialized.")

        # Initialize Translator
        self.logger.info("Initializing Translator...")
        translations_path = Path("./shared/translations")
        self.translator = Translator(self, translations_path, self.logger )
        self.logger.info("Translator initialized.")

        # Initialize Handlers
        self.logger.info("Initializing Handlers...")
        self.command_handler = CommandHandler(self, self.logger)
        self.event_handler = EventHandler(self, self.logger)
        self.logger.info("Handlers initialized.")

        # Initialize and load modules
        self.logger.info("Initializing ModuleHandler...")
        self.module_handler = ModuleHandler(self, self.logger)
        await self.module_handler.load_modules()
        self.logger.info("ModuleHandler initialized.")

        # Sync slash commands
        await self.sync_slash_commands()

    async def sync_slash_commands(self):
        """Interactively synchronize slash commands.

        Offers options to sync globally, to specific guilds, to the configured test guild,
        or to skip syncing. Consider a non‑interactive flag for headless deployments.
        """

        print("\nChoose a sync option for slash commands:")
        print("1. Sync globally (all servers)")
        print("2. Sync to a specific guild")
        print(f"3. Sync to the test guild (ID: {TEST_GUILD_ID})")
        print("4. Do not sync (skip this step)")
        sync_choice = input("Enter your choice (1/2/3/4): ").strip()
        # sync_choice = ""

        try:
            if sync_choice == "1":
                await self.tree.sync()
                self.logger.info("Slash commands synced globally.")
            elif sync_choice == "2":
                guild_ids = input("Enter guild IDs separated by commas: ").strip().split(",")
                for guild_id in guild_ids:
                    guild = Object(id=int(guild_id.strip()))
                    await self.tree.sync(guild=guild)
                    self.logger.info(f"Slash commands synced to guild {guild_id.strip()}.")
            elif sync_choice == "3":
                guild = Object(id=TEST_GUILD_ID)
                await self.tree.sync(guild=guild)
                self.logger.info(f"Slash commands synced to test guild {TEST_GUILD_ID}.")
            elif sync_choice == "4":
                self.logger.info("Slash command synchronization skipped.")
            else:
                print("Invalid choice. No sync performed.")
        except Exception as e:
            self.logger.error(f"Failed to sync slash commands: {e}")

    async def _populate_language_cache(self):
        """Preload per‑guild language into the Translator cache."""
        self.logger.info("Populating language cache...")
        # self.logger.info(f"Guilds: {self.guilds}")
        try:
            for guild in self.guilds:
                guild_id = str(guild.id)
                language = await self.guild_manager.get_language(guild_id)
                self.translator.language_cache[guild_id] = language
                self.logger.debug(f"Cached language '{language}' for guild '{guild_id}'")
        except Exception as e:
            self.logger.error(f"Error while populating language cache: {e}")
        self.logger.info("Language cache populated.")

    def get_logger(self, name: str) -> logging.Logger:
        """Return a child logger derived from the bot's root logger.

        Args:
            name: Name suffix for the child logger.

        Returns:
            A configured child logger.
        """
        return self.logger.getChild(name)

    async def on_ready(self):
        """Run one‑time tasks after the bot connects for the first time.

        Sets presence, warms the language cache, refreshes translation data,
        and flips the internal ready flag.
        """
        if not self.ready:
            self.logger.info(f"Bot connected as {self.user}")
            await self.change_presence(activity=Activity(type=ActivityType.watching, name="TechJoyce"))
            await self._populate_language_cache()
            await self.translator.refresh_translation_cache()
            self.ready = True

    async def close(self):
        """
        Gracefully close the bot, including HTTP sessions and database connections.
        """
        self.logger.info("Shutting down bot...")
        if self.session:
            await self.session.close()
        if self.db:
            await self.db.close()
        await super().close()

close() async

Gracefully close the bot, including HTTP sessions and database connections.

Source code in main.py
229
230
231
232
233
234
235
236
237
238
async def close(self):
    """
    Gracefully close the bot, including HTTP sessions and database connections.
    """
    self.logger.info("Shutting down bot...")
    if self.session:
        await self.session.close()
    if self.db:
        await self.db.close()
    await super().close()

get_logger(name)

Return a child logger derived from the bot's root logger.

Parameters:

Name Type Description Default
name str

Name suffix for the child logger.

required

Returns:

Type Description
Logger

A configured child logger.

Source code in main.py
205
206
207
208
209
210
211
212
213
214
def get_logger(self, name: str) -> logging.Logger:
    """Return a child logger derived from the bot's root logger.

    Args:
        name: Name suffix for the child logger.

    Returns:
        A configured child logger.
    """
    return self.logger.getChild(name)

on_ready() async

Run one‑time tasks after the bot connects for the first time.

Sets presence, warms the language cache, refreshes translation data, and flips the internal ready flag.

Source code in main.py
216
217
218
219
220
221
222
223
224
225
226
227
async def on_ready(self):
    """Run one‑time tasks after the bot connects for the first time.

    Sets presence, warms the language cache, refreshes translation data,
    and flips the internal ready flag.
    """
    if not self.ready:
        self.logger.info(f"Bot connected as {self.user}")
        await self.change_presence(activity=Activity(type=ActivityType.watching, name="TechJoyce"))
        await self._populate_language_cache()
        await self.translator.refresh_translation_cache()
        self.ready = True

setup_hook() async

Initialize external services and internal subsystems before the bot is ready.

This connects to MongoDB, prepares HTTP session, instantiates managers and handlers, registers default permission namespaces, loads modules, then offers slash command sync.

Source code in main.py
 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
async def setup_hook(self):
    """Initialize external services and internal subsystems before the bot is ready.

    This connects to MongoDB, prepares HTTP session, instantiates managers and handlers,
    registers default permission namespaces, loads modules, then offers slash command sync.
    """
    # Initialize MongoDB connection
    self.logger.info("Connecting to MongoDB...")
    try:
        self.db = MongoDBAsyncORM(uri=MONGODB_URI, db_name="GigaJoyce-Test")
        await self.db.create_index("members", [("id", 1), ("guildId", 1)], unique=True)
        self.db.members = self.db.get_collection("members")
        self.db.guilds = self.db.get_collection("guilds")
        self.db.users = self.db.get_collection("users")
        self.logger.info("Successfully connected to the database and collected data.")
    except Exception as e:
        self.logger.error(f"Failed to connect to MongoDB: {e}")
        return

    # Initialize HTTP session+
    self.session = aiohttp.ClientSession()

    # Initialize Managers
    self.logger.info("Initializing Managers...")
    self.guild_manager= GuildManager(self, self.logger)
    self.member_manager = MemberManager(self, self.logger)
    self.settings_manager = SettingsManager(self, self.logger)
    self.permission_manager = PermissionsManager(self, self.logger)
    self.logger.info("Managers initialized.")

    # Register default permission namespaces
    self.logger.info("Registering default permission namespaces...")
    self.permission_manager.register_node("Role.*", RolesNamespace)
    self.permission_manager.register_node("User.*", UsersNamespace)
    self.permission_manager.register_node("Channel.*", ChannelsNamespace)
    self.logger.info("Default permission namespaces registered.")

    # Initialize EmojiManager
    self.logger.info("Initializing EmojiManager...")
    translations_path = Path("./shared/emojis")
    self.emoji_manager = EmojiManager(self, translations_path, self.logger )
    self.logger.info("EmojiManager initialized.")

    # Initialize Translator
    self.logger.info("Initializing Translator...")
    translations_path = Path("./shared/translations")
    self.translator = Translator(self, translations_path, self.logger )
    self.logger.info("Translator initialized.")

    # Initialize Handlers
    self.logger.info("Initializing Handlers...")
    self.command_handler = CommandHandler(self, self.logger)
    self.event_handler = EventHandler(self, self.logger)
    self.logger.info("Handlers initialized.")

    # Initialize and load modules
    self.logger.info("Initializing ModuleHandler...")
    self.module_handler = ModuleHandler(self, self.logger)
    await self.module_handler.load_modules()
    self.logger.info("ModuleHandler initialized.")

    # Sync slash commands
    await self.sync_slash_commands()

sync_slash_commands() async

Interactively synchronize slash commands.

Offers options to sync globally, to specific guilds, to the configured test guild, or to skip syncing. Consider a non‑interactive flag for headless deployments.

Source code in main.py
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
async def sync_slash_commands(self):
    """Interactively synchronize slash commands.

    Offers options to sync globally, to specific guilds, to the configured test guild,
    or to skip syncing. Consider a non‑interactive flag for headless deployments.
    """

    print("\nChoose a sync option for slash commands:")
    print("1. Sync globally (all servers)")
    print("2. Sync to a specific guild")
    print(f"3. Sync to the test guild (ID: {TEST_GUILD_ID})")
    print("4. Do not sync (skip this step)")
    sync_choice = input("Enter your choice (1/2/3/4): ").strip()
    # sync_choice = ""

    try:
        if sync_choice == "1":
            await self.tree.sync()
            self.logger.info("Slash commands synced globally.")
        elif sync_choice == "2":
            guild_ids = input("Enter guild IDs separated by commas: ").strip().split(",")
            for guild_id in guild_ids:
                guild = Object(id=int(guild_id.strip()))
                await self.tree.sync(guild=guild)
                self.logger.info(f"Slash commands synced to guild {guild_id.strip()}.")
        elif sync_choice == "3":
            guild = Object(id=TEST_GUILD_ID)
            await self.tree.sync(guild=guild)
            self.logger.info(f"Slash commands synced to test guild {TEST_GUILD_ID}.")
        elif sync_choice == "4":
            self.logger.info("Slash command synchronization skipped.")
        else:
            print("Invalid choice. No sync performed.")
    except Exception as e:
        self.logger.error(f"Failed to sync slash commands: {e}")

parse_args()

Parse command line arguments.

Returns:

Type Description

Parsed arguments including the debug flag used to control logging verbosity.

Source code in main.py
28
29
30
31
32
33
34
35
36
def parse_args():
    """Parse command line arguments.

    Returns:
        Parsed arguments including the debug flag used to control logging verbosity.
    """
    parser = argparse.ArgumentParser(description="Discord Bot")
    parser.add_argument('--debug', action='store_true', help='Enable debug mode')
    return parser.parse_args()