Skip to content

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:

  1. Initialization: Loads environment variables (DISCORD_SERVER_ID, BOT_TOKEN) and validates them.
  2. Extension Loading: Iterates through the src.oscar.cogs package and loads all available modules automatically using pkgutil.
  3. 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

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()

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)