mirror of
https://github.com/wassname/Open-Assistant.git
synced 2026-09-09 11:15:08 +08:00
initial bot structure
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
# -*- coding=utf-8 -*-
|
||||
"""The official Open-Assistant Discord Bot."""
|
||||
@@ -0,0 +1,17 @@
|
||||
# -*- coding=utf-8 -*-
|
||||
"""Entry point for the bot."""
|
||||
import logging
|
||||
import os
|
||||
|
||||
from bot.bot import bot
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
if __name__ == "__main__":
|
||||
if os.name != "nt":
|
||||
import uvloop
|
||||
|
||||
uvloop.install()
|
||||
|
||||
logger.info("Starting bot")
|
||||
bot.run()
|
||||
@@ -0,0 +1,37 @@
|
||||
# -*- coding=utf-8 -*-
|
||||
"""Bot logic."""
|
||||
import hikari
|
||||
|
||||
import aiosqlite
|
||||
import lightbulb
|
||||
import miru
|
||||
from bot.config import Config
|
||||
|
||||
config = Config.from_env()
|
||||
|
||||
bot = lightbulb.BotApp(
|
||||
token=config.token,
|
||||
logs="DEBUG",
|
||||
prefix="./",
|
||||
default_enabled_guilds=config.declare_global_commands,
|
||||
owner_ids=config.owner_ids,
|
||||
intents=hikari.Intents.ALL,
|
||||
)
|
||||
|
||||
|
||||
@bot.listen()
|
||||
async def on_starting(event: hikari.StartingEvent):
|
||||
"""Setup."""
|
||||
|
||||
miru.install(bot) # component handler
|
||||
bot.load_extensions_from("./bot/extensions") # load extensions
|
||||
|
||||
bot.d.db = await aiosqlite.connect(":memory:") # TODO: Update
|
||||
await bot.d.db.executescript(open("./bot/db/schema.sql").read())
|
||||
await bot.d.db.commit()
|
||||
|
||||
|
||||
@bot.listen()
|
||||
async def on_stopping(event: hikari.StoppingEvent):
|
||||
"""Cleanup."""
|
||||
await bot.d.db.close()
|
||||
@@ -0,0 +1,35 @@
|
||||
# -*- coding=utf-8 -*-
|
||||
"""Configuration for the bot."""
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from os import getenv
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
"""Configuration for the bot."""
|
||||
|
||||
token: str
|
||||
declare_global_commands: int
|
||||
owner_ids: list[int]
|
||||
|
||||
@classmethod
|
||||
def from_env(cls):
|
||||
token = getenv("TOKEN", None)
|
||||
|
||||
if token is None:
|
||||
logger.error("Invalid token, please set the TOKEN environment variable.")
|
||||
exit(1)
|
||||
|
||||
return cls(
|
||||
token=token,
|
||||
declare_global_commands=int(getenv("DECLARE_GLOBAL_COMMANDS", 0)),
|
||||
owner_ids=[int(x) for x in getenv("OWNER_IDS", "").split(",")],
|
||||
)
|
||||
@@ -0,0 +1,10 @@
|
||||
-- Sqlite3 schema for the bot
|
||||
CREATE TABLE IF NOT EXISTS guild_settings (
|
||||
guild_id BIGINT NOT NULL PRIMARY KEY,
|
||||
log_channel_id BIGINT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS example (
|
||||
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
|
||||
name VARCHAR(255) NOT NULL
|
||||
);
|
||||
@@ -0,0 +1,61 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Hot reload plugin."""
|
||||
from glob import glob
|
||||
|
||||
import hikari
|
||||
import lightbulb
|
||||
|
||||
plugin = lightbulb.Plugin(
|
||||
"HotReloadPlugin",
|
||||
)
|
||||
plugin.add_checks(lightbulb.owner_only)
|
||||
|
||||
EXTENSIONS_FOLDER = "bot/extensions"
|
||||
|
||||
|
||||
def _get_extensions() -> list[str]:
|
||||
# Recursively get all the .py files in the extensions directory.
|
||||
exts = glob("bot/extensions/**/*.py", recursive=True)
|
||||
# Turn the path into a plugin path ("path/to/extension.py" -> "path.to.extension")
|
||||
return [ext.replace("/", ".").replace("\\", ".").replace(".py", "") for ext in exts]
|
||||
|
||||
|
||||
async def _plugin_autocomplete(option: hikari.CommandInteractionOption, _: hikari.AutocompleteInteraction) -> list[str]:
|
||||
# Check that the option is a string.
|
||||
if not isinstance(option.value, str):
|
||||
raise TypeError(f"`option.value` must be of type `str`, it is currently a `{type(option.value)}`")
|
||||
|
||||
exts = _get_extensions()
|
||||
return [ext for ext in exts if option.value in ext]
|
||||
|
||||
|
||||
@plugin.command
|
||||
@lightbulb.option(
|
||||
"plugin",
|
||||
"The plugin to reload. Leave empty to reload all plugins.",
|
||||
autocomplete=_plugin_autocomplete,
|
||||
required=False,
|
||||
default=None,
|
||||
)
|
||||
@lightbulb.command("reload", "Reload a plugin")
|
||||
@lightbulb.implements(lightbulb.SlashCommand)
|
||||
async def reload(ctx: lightbulb.SlashContext):
|
||||
"""Reload a plugin or all plugins."""
|
||||
# If the plugin option is None, reload all plugins.
|
||||
if ctx.options.plugin is None:
|
||||
ctx.bot.reload_extensions(*_get_extensions())
|
||||
await ctx.respond("Reloaded all plugins.")
|
||||
# Otherwise, reload the specified plugin.
|
||||
else:
|
||||
ctx.bot.reload_extensions(ctx.options.plugin)
|
||||
await ctx.respond(f"Reloaded `{ctx.options.plugin}`.")
|
||||
|
||||
|
||||
def load(bot: lightbulb.BotApp):
|
||||
"""Add the plugin to the bot."""
|
||||
bot.add_plugin(plugin)
|
||||
|
||||
|
||||
def unload(bot: lightbulb.BotApp):
|
||||
"""Remove the plugin to the bot."""
|
||||
bot.remove_plugin(plugin)
|
||||
Reference in New Issue
Block a user