Bot Core System
The Oscar class is the heart of the application. It inherits from discord.ext.commands.Bot and orchestrates the startup process, extension loading, and environment configuration.
The Oscar Class
graph TD
A["Bot Started"] --> B["Load Cogs"]
B --> C["Sync Commands"]
C --> D["Bot Ready"]
D --> E["Wait for Command"]
E --> F{Command Type?}
F -->|/module| G["Search Module"]
F -->|/filter| H["Open Filter"]
F -->|/semesterplan| I["View Plan"]
F -->|/feedback| J["Open Feedback"]
F -->|/review_feedback| K["Admin: View Feedback"]
F -->|/help| L["Display Help"]
F -->|/ping| M["Show Latency"]
G --> G1["Query Tables API"]
G1 --> G2["Display Module Info"]
G2 --> G3{User Action?}
G3 -->|Add to Plan| G4["Save to DB"]
G3 -->|More Info| G5["Show Details"]
G3 -->|Back| E
G4 --> E
G5 --> E
H --> H1["Show Filter Options"]
H1 --> H2["Apply CP/SWS Filter"]
H2 --> H3["Show Results"]
H3 --> E
I --> I1["Fetch User Plan"]
I1 --> I2["Display Modules"]
I2 --> I3{Action?}
I3 -->|Remove| I4["Delete from Plan"]
I3 -->|View Detail| I5["Show Module Info"]
I4 --> E
I5 --> E
J --> J1["Open Feedback Form"]
J2["Rate 3 Dimensions"]
J1 --> J2
J2 --> J3["Add Optional Text"]
J3 --> J4["Submit & Save"]
J4 --> E
K --> K1["Generate Plots"]
K1 --> K2["Select Time Range"]
K2 --> K3["Update Charts"]
K3 --> K4{Export?}
K4 -->|Yes| K5["Download JSON"]
K4 -->|No| E
K5 --> E
L --> L1["Select Topic"]
L1 --> L2["Show Help Text"]
L2 --> E
M --> M1["Calculate Latency"]
M1 --> M2["Display ms"]
M2 --> E
style A fill:#5f6af0,color:#fff
style D fill:#5f6af0,color:#fff
style E fill:#4a9bc4,color:#fff
style F fill:#6b7280,color:#fff
style G fill:#5ab5e0,color:#000
style H fill:#5ab5e0,color:#000
style I fill:#5ab5e0,color:#000
style J fill:#6ac5e8,color:#000
style K fill:#d4af37,color:#000
style L fill:#5ab5e0,color:#000
style M fill:#6b7280,color:#fff
Responsibilities:
- Initialization: Loads environment variables (
DISCORD_SERVER_ID, BOT_TOKEN) and validates them.
- Extension Loading: Iterates through the
src.oscar.cogs package and loads all available modules automatically using pkgutil.
- Command Sync: Synchronizes the Application Command Tree (Slash Commands) with the Discord API upon startup (
on_ready).
src.oscar.oscar.Oscar
Oscar(*args: tuple[Any], **kwargs: dict[str, Any])
Bases: Bot
Custom Bot class for abstraction
Source code in src/oscar/oscar.py
| def __init__(
self,
*args: tuple[Any], # pyright: ignore[reportExplicitAny]
**kwargs: dict[str, Any] # pyright: ignore[reportExplicitAny]
):
# Forward all arguments, and keyword-only arguments to commands.Bot
self.server_id: str
self.bot_token: str
self._startup_done: bool = False
super().__init__(*args, **kwargs)
# self.db = None # To Do: establish database connection (maybe somewhere else?)
if not load_dotenv():
logger.warning("No Environment Variables File found.")
try:
self.server_id = os.environ[
"DISCORD_SERVER_ID"
] # Hole dir die Server-ID aus der Umgebungsvariablen
except KeyError as exc:
logger.exception(
"DISCORD_SERVER_ID Environment Variable not set, terminating."
)
raise RuntimeError("Constructor failed") from exc
try:
self.bot_token = os.environ["BOT_TOKEN"]
except KeyError as exc:
logger.exception("BOT_TOKEN Environment Variable not set, terminating.")
raise RuntimeError("Constructor failed") from exc
# ensure db is ready, and semesterplans are loaded:
_ = get_database()
_ = get_semesterplans_manager()
|
setup_bot
Sets up (automatically loads all available cogs) and runs the bot instance
Source code in src/oscar/oscar.py
| def setup_bot(self):
"""Sets up (automatically loads all available cogs) and runs the bot instance"""
self._startup_done: bool = False
@self.event
async def on_ready(): # pyright: ignore[reportUnusedFunction]
revision = os.environ.get("OSCAR_REVISION", "unknown (local run)")
logger.info(f"Logged in as {self.user}, running build {revision}")
# on_ready fires again after every reconnect
if self._startup_done:
return
for _, name, _ in pkgutil.walk_packages(
oscar_cogs.__path__, oscar_cogs.__name__ + "."
):
try:
await self.load_extension(name)
except Exception: # pylint: disable=broad-exception-caught
logger.exception(f"Couldn't load '{name}'")
# without a translator discord.py sends the names in one language only
await self.tree.set_translator(GermanNames())
hidden = self.hide_commands()
if hidden:
logger.info(f"Switched off, see util.command_surface.HIDDEN: {hidden}")
_ = await self.sync_commands()
self._startup_done = True
# The faculty renames a BookStack book whenever an edition rolls over, and
# the module handbook button then opens a 404.
_ = asyncio.create_task(self.keep_handbook_links_fresh())
# Nobody should be the one who waits for the module table.
_ = asyncio.create_task(self.keep_tables_warm())
async def on_app_command_error(
interaction: discord.Interaction, error: app_commands.AppCommandError
):
# without this, any exception shows up as a silent "application did not respond"
if isinstance(error, app_commands.CheckFailure):
message = ("Du darfst diesen Befehl nicht nutzen. "
"/ You are not allowed to use this command.")
else:
name = interaction.command.name if interaction.command else "?"
logger.opt(exception=error).error(f"Command '/{name}' failed")
message = ("Da ist leider etwas schiefgelaufen, versuch es bitte später nochmal. "
"/ Something went wrong, please try again later.")
try:
if interaction.response.is_done():
_ = await interaction.followup.send(message, ephemeral=True)
else:
_ = await interaction.response.send_message(message, ephemeral=True)
except discord.HTTPException:
logger.exception("Couldn't send the error message to the user")
self.tree.on_error = on_app_command_error
async def operator_only(ctx: commands.context.Context[commands.Bot]) -> bool:
# `has_permissions(administrator=True)` let the administrator of any server
# in, and anybody can invite a public bot to a server of their own
permissions = getattr(ctx.author, "guild_permissions", None)
if is_operator(
user_id=ctx.author.id,
guild_id=ctx.guild.id if ctx.guild else None,
is_administrator=bool(permissions is not None and permissions.administrator),
):
return True
return await self.is_owner(ctx.author)
# Ping command to test deployment latency
@self.command()
@commands.check(operator_only)
async def ping( # pyright: ignore[reportUnusedFunction]
ctx: commands.context.Context[commands.Bot]
):
latency = round(self.latency, 2)
logger.info(f"Latency is `{latency}ms`")
_ = await ctx.send(f"Latency is {latency}ms")
@self.command()
@commands.check(operator_only)
async def resync( # pyright: ignore[reportUnusedFunction]
ctx: commands.context.Context[commands.Bot]
):
synced = await self.sync_commands()
names = ", ".join(f"/{cmd.name}" for cmd in synced)
# a prefix command needs no slash sync, so this always names the running build
revision = os.environ.get("OSCAR_REVISION", "unknown (local run)")
_ = await ctx.send(
f"🔄 Build `{revision}` synced {len(synced)} commands "
f"to {len(self.guilds)} server(s): {names}"
)
|
Main Entry Point
The main.py script serves as the bootstrapper. It configures the discord.Intents (specifically enabling message_content) and instantiates the Oscar bot.
src.oscar.main.main
Main method
Source code in src/oscar/main.py
| def main():
"""Main method"""
intents = discord.Intents.default()
intents.message_content = True
try:
oscar: Oscar = Oscar(
command_prefix="!", # pyright: ignore[reportArgumentType]
intents=intents, # pyright: ignore[reportArgumentType]
# OSCAR prints text students typed. Nothing it says pings anybody unless
# the message asks for it, the study group thread is the one that does.
allowed_mentions=discord.AllowedMentions.none(), # pyright: ignore[reportArgumentType]
)
except Exception: # pylint: disable=broad-exception-caught
sys.exit(1)
oscar.setup_bot()
oscar.run(oscar.bot_token)
|