diff --git a/setup.cfg b/setup.cfg
index 033c898..8f000f7 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -11,14 +11,16 @@ packages = find_namespace:
python_requires = >=3.7
;todo: remove flatten_dict (core/config.py)
install_requires =
- appdirs==1.4.4
- Babel==2.8.0
+ appdirs>=1.4.4
+ Babel>=2.8.0
discord.py==1.4.1
discord_flags==2.1.1
- flatten_dict==0.3.0
- jishaku==1.19.1.200
- PyYAML==5.3.1
- rich==6.0.0
+ flatten_dict>=0.3.0
+ jishaku>=1.19.1.200
+ psutil>=5.7.2
+ PyYAML>=5.3.1
+ rich>=6.0.0
+ structured_config>=4.12
[options.entry_points]
console_scripts =
diff --git a/tuxbot/__main__.py b/tuxbot/__main__.py
index 14f8e33..3d626a2 100644
--- a/tuxbot/__main__.py
+++ b/tuxbot/__main__.py
@@ -1,318 +1,26 @@
-import argparse
-import asyncio
-import json
-import logging
-import signal
-import sys
-import os
-from argparse import Namespace
from typing import NoReturn
-import discord
-import pip
-import tracemalloc
-from rich.columns import Columns
from rich.console import Console
-from rich.panel import Panel
from rich.traceback import install
-from rich.table import Table, box
-from rich.text import Text
-from rich import print
-
-import tuxbot.logging
-from tuxbot.core import data_manager
-from tuxbot.core.bot import Tux
-from . import __version__, version_info, ExitCodes
-
-log = logging.getLogger("tuxbot.main")
+from tuxbot import ExitCodes
console = Console()
install(console=console)
-tracemalloc.start()
-
-
-def list_instances() -> NoReturn:
- """List all available instances
-
- """
- with data_manager.config_file.open() as fs:
- data = json.load(fs)
-
- console.print(
- Panel("[bold green]Instances", style="green"),
- justify="center"
- )
- console.print()
-
- columns = Columns(expand=True, padding=2, align="center")
- for instance, details in data.items():
- is_running = details.get('IS_RUNNING')
-
- table = Table(
- style="dim", border_style="not dim",
- box=box.HEAVY_HEAD
- )
- table.add_column("Name")
- table.add_column(("Running" if is_running else "Down") + " since")
- table.add_row(instance, "42")
- table.title = Text(
- instance,
- style="green" if is_running else "red"
- )
- columns.add_renderable(table)
- console.print(columns)
- console.print()
-
- sys.exit(os.EX_OK)
-
-
-def debug_info() -> NoReturn:
- """Show debug info relatives to the bot
-
- """
- python_version = sys.version.replace("\n", "")
- pip_version = pip.__version__
- tuxbot_version = __version__
- dpy_version = discord.__version__
-
- uptime = os.popen('uptime').read().strip().split()
-
- console.print(
- Panel("[bold blue]Debug Info", style="blue"),
- justify="center"
- )
- console.print()
-
- columns = Columns(expand=True, padding=2, align="center")
-
- table = Table(
- style="dim", border_style="not dim",
- box=box.HEAVY_HEAD
- )
- table.add_column(
- "Bot Info",
- )
- table.add_row(f"[u]Tuxbot version:[/u] {tuxbot_version}")
- table.add_row(f"[u]Major:[/u] {version_info.major}")
- table.add_row(f"[u]Minor:[/u] {version_info.minor}")
- table.add_row(f"[u]Micro:[/u] {version_info.micro}")
- table.add_row(f"[u]Level:[/u] {version_info.releaselevel}")
- table.add_row(f"[u]Last change:[/u] {version_info.info}")
- columns.add_renderable(table)
-
- table = Table(
- style="dim", border_style="not dim",
- box=box.HEAVY_HEAD
- )
- table.add_column(
- "Python Info",
- )
- table.add_row(f"[u]Python version:[/u] {python_version}")
- table.add_row(f"[u]Python executable path:[/u] {sys.executable}")
- table.add_row(f"[u]Pip version:[/u] {pip_version}")
- table.add_row(f"[u]Discord.py version:[/u] {dpy_version}")
- columns.add_renderable(table)
-
- table = Table(
- style="dim", border_style="not dim",
- box=box.HEAVY_HEAD
- )
- table.add_column(
- "Server Info",
- )
- table.add_row(f"[u]System:[/u] {os.uname().sysname}")
- table.add_row(f"[u]System arch:[/u] {os.uname().machine}")
- table.add_row(f"[u]Kernel:[/u] {os.uname().release}")
- table.add_row(f"[u]User:[/u] {os.getlogin()}")
- table.add_row(f"[u]Uptime:[/u] {uptime[2]}")
- table.add_row(
- f"[u]Load Average:[/u] {' '.join(map(str, os.getloadavg()))}"
- )
- columns.add_renderable(table)
-
- console.print(columns)
- console.print()
-
- sys.exit(os.EX_OK)
-
-
-def parse_cli_flags(args: list) -> Namespace:
- """Parser for cli values.
-
- Parameters
- ----------
- args:list
- Is a list of all passed values.
- Returns
- -------
- Namespace
- """
- parser = argparse.ArgumentParser(
- description="Tuxbot - OpenSource bot",
- usage="tuxbot <instance_name> [arguments]",
- )
- parser.add_argument(
- "--version", "-V", action="store_true",
- help="Show tuxbot's used version"
- )
- parser.add_argument(
- "--debug", action="store_true",
- help="Show debug information."
- )
- parser.add_argument(
- "--list-instances", "-L", action="store_true",
- help="List all instance names"
- )
- parser.add_argument("--token", "-T", type=str,
- help="Run Tuxbot with passed token")
- parser.add_argument(
- "instance_name",
- nargs="?",
- help="Name of the bot instance created during `tuxbot-setup`.",
- )
-
- args = parser.parse_args(args)
-
- return args
-
-
-async def shutdown_handler(tux: Tux, signal_type, exit_code=None) -> NoReturn:
- """Handler when the bot shutdown
-
- It cancels all running task.
-
- Parameters
- ----------
- tux:Tux
- Object for the bot.
- signal_type:int, None
- Exiting signal code.
- exit_code:None|int
- Code to show when exiting.
- """
- if signal_type:
- log.info("%s received. Quitting...", signal_type)
- elif exit_code is None:
- log.info("Shutting down from unhandled exception")
- tux.shutdown_code = ExitCodes.CRITICAL
-
- if exit_code is not None:
- tux.shutdown_code = exit_code
-
- await tux.shutdown()
-
-
-async def run_bot(tux: Tux, cli_flags: Namespace) -> None:
- """This run the bot.
-
- Parameters
- ----------
- tux:Tux
- Object for the bot.
- cli_flags:Namespace
- All different flags passed in the console.
-
- Returns
- -------
- None
- When exiting, this function return None.
- """
- data_path = data_manager.data_path(tux.instance_name)
-
- tuxbot.logging.init_logging(10, location=data_path / "logs")
-
- log.debug("====Basic Config====")
- log.debug("Data Path: %s", data_path)
-
- if cli_flags.token:
- token = cli_flags.token
- else:
- token = tux.config("core").get("token")
-
- if not token:
- log.critical("Token must be set if you want to login.")
- sys.exit(ExitCodes.CRITICAL)
-
- try:
- await tux.load_packages()
- console.print()
- await tux.start(token=token, bot=True)
- except discord.LoginFailure:
- log.critical("This token appears to be valid.")
- console.print()
- console.print(
- "[prompt.invalid]This token appears to be valid. [i]exiting...[/i]"
- )
- sys.exit(ExitCodes.CRITICAL)
- except Exception as e:
- raise e
-
- return None
def main() -> NoReturn:
- """Main function
-
- """
- tux = None
- cli_flags = parse_cli_flags(sys.argv[1:])
-
- if cli_flags.list_instances:
- list_instances()
- elif cli_flags.debug:
- debug_info()
- elif cli_flags.version:
- print(f"Tuxbot V{version_info.major}")
- print(f"Complete Version: {__version__}")
-
- sys.exit(os.EX_OK)
-
- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(loop)
-
try:
- if not cli_flags.instance_name:
- console.print(
- "[red]No instance provided ! "
- "You can use 'tuxbot -L' to list all available instances"
- )
- sys.exit(ExitCodes.CRITICAL)
+ from .__run__ import run
- tux = Tux(
- cli_flags=cli_flags,
- description="Tuxbot, made from and for OpenSource",
- dm_help=None,
- )
-
- loop.run_until_complete(run_bot(tux, cli_flags))
- except KeyboardInterrupt:
- console.print(
- " [red]Please use <prefix>quit instead of Ctrl+C to Shutdown!"
- )
- log.warning("Please use <prefix>quit instead of Ctrl+C to Shutdown!")
- log.info("Received KeyboardInterrupt")
- console.print("[i]Trying to shutdown...")
- if tux is not None:
- loop.run_until_complete(shutdown_handler(tux, signal.SIGINT))
+ run()
except SystemExit as exc:
- log.info("Shutting down with exit code: %s", exc.code)
- if tux is not None:
- loop.run_until_complete(shutdown_handler(tux, None, exc.code))
- except Exception as exc:
- log.error("Unexpected exception (%s): ", type(exc))
+ if exc.code == ExitCodes.RESTART:
+ from .__run__ import run # reimport to load changes
+ run()
+ else:
+ raise exc
+ except Exception:
console.print_exception()
- if tux is not None:
- loop.run_until_complete(shutdown_handler(tux, None, 1))
- finally:
- loop.run_until_complete(loop.shutdown_asyncgens())
- log.info("Please wait, cleaning up a bit more")
- loop.run_until_complete(asyncio.sleep(1))
- asyncio.set_event_loop(None)
- loop.stop()
- loop.close()
- exit_code = ExitCodes.CRITICAL if tux is None else tux.shutdown_code
-
- sys.exit(exit_code)
if __name__ == "__main__":
diff --git a/tuxbot/__run__.py b/tuxbot/__run__.py
new file mode 100644
index 0000000..f8dd4c1
--- /dev/null
+++ b/tuxbot/__run__.py
@@ -0,0 +1,315 @@
+import argparse
+import asyncio
+import json
+import logging
+import signal
+import sys
+import os
+from argparse import Namespace
+from typing import NoReturn
+
+import discord
+import pip
+import tracemalloc
+from rich.columns import Columns
+from rich.console import Console
+from rich.panel import Panel
+from rich.traceback import install
+from rich.table import Table, box
+from rich.text import Text
+from rich import print
+
+import tuxbot.logging
+from tuxbot.core import data_manager
+from tuxbot.core.bot import Tux
+from . import __version__, version_info, ExitCodes
+
+log = logging.getLogger("tuxbot.main")
+
+console = Console()
+install(console=console)
+tracemalloc.start()
+
+
+def list_instances() -> NoReturn:
+ """List all available instances
+
+ """
+ with data_manager.config_file.open() as fs:
+ data = json.load(fs)
+
+ console.print(
+ Panel("[bold green]Instances", style="green"),
+ justify="center"
+ )
+ console.print()
+
+ columns = Columns(expand=True, padding=2, align="center")
+ for instance, details in data.items():
+ is_running = details.get('IS_RUNNING')
+
+ table = Table(
+ style="dim", border_style="not dim",
+ box=box.HEAVY_HEAD
+ )
+ table.add_column("Name")
+ table.add_column(("Running" if is_running else "Down") + " since")
+ table.add_row(instance, "42")
+ table.title = Text(
+ instance,
+ style="green" if is_running else "red"
+ )
+ columns.add_renderable(table)
+ console.print(columns)
+ console.print()
+
+ sys.exit(os.EX_OK)
+
+
+def debug_info() -> NoReturn:
+ """Show debug info relatives to the bot
+
+ """
+ python_version = sys.version.replace("\n", "")
+ pip_version = pip.__version__
+ tuxbot_version = __version__
+ dpy_version = discord.__version__
+
+ uptime = os.popen('uptime').read().strip().split()
+
+ console.print(
+ Panel("[bold blue]Debug Info", style="blue"),
+ justify="center"
+ )
+ console.print()
+
+ columns = Columns(expand=True, padding=2, align="center")
+
+ table = Table(
+ style="dim", border_style="not dim",
+ box=box.HEAVY_HEAD
+ )
+ table.add_column(
+ "Bot Info",
+ )
+ table.add_row(f"[u]Tuxbot version:[/u] {tuxbot_version}")
+ table.add_row(f"[u]Major:[/u] {version_info.major}")
+ table.add_row(f"[u]Minor:[/u] {version_info.minor}")
+ table.add_row(f"[u]Micro:[/u] {version_info.micro}")
+ table.add_row(f"[u]Level:[/u] {version_info.releaselevel}")
+ table.add_row(f"[u]Last change:[/u] {version_info.info}")
+ columns.add_renderable(table)
+
+ table = Table(
+ style="dim", border_style="not dim",
+ box=box.HEAVY_HEAD
+ )
+ table.add_column(
+ "Python Info",
+ )
+ table.add_row(f"[u]Python version:[/u] {python_version}")
+ table.add_row(f"[u]Python executable path:[/u] {sys.executable}")
+ table.add_row(f"[u]Pip version:[/u] {pip_version}")
+ table.add_row(f"[u]Discord.py version:[/u] {dpy_version}")
+ columns.add_renderable(table)
+
+ table = Table(
+ style="dim", border_style="not dim",
+ box=box.HEAVY_HEAD
+ )
+ table.add_column(
+ "Server Info",
+ )
+ table.add_row(f"[u]System:[/u] {os.uname().sysname}")
+ table.add_row(f"[u]System arch:[/u] {os.uname().machine}")
+ table.add_row(f"[u]Kernel:[/u] {os.uname().release}")
+ table.add_row(f"[u]User:[/u] {os.getlogin()}")
+ table.add_row(f"[u]Uptime:[/u] {uptime[2]}")
+ table.add_row(
+ f"[u]Load Average:[/u] {' '.join(map(str, os.getloadavg()))}"
+ )
+ columns.add_renderable(table)
+
+ console.print(columns)
+ console.print()
+
+ sys.exit(os.EX_OK)
+
+
+def parse_cli_flags(args: list) -> Namespace:
+ """Parser for cli values.
+
+ Parameters
+ ----------
+ args:list
+ Is a list of all passed values.
+ Returns
+ -------
+ Namespace
+ """
+ parser = argparse.ArgumentParser(
+ description="Tuxbot - OpenSource bot",
+ usage="tuxbot <instance_name> [arguments]",
+ )
+ parser.add_argument(
+ "--version", "-V", action="store_true",
+ help="Show tuxbot's used version"
+ )
+ parser.add_argument(
+ "--debug", action="store_true",
+ help="Show debug information."
+ )
+ parser.add_argument(
+ "--list-instances", "-L", action="store_true",
+ help="List all instance names"
+ )
+ parser.add_argument("--token", "-T", type=str,
+ help="Run Tuxbot with passed token")
+ parser.add_argument(
+ "instance_name",
+ nargs="?",
+ help="Name of the bot instance created during `tuxbot-setup`.",
+ )
+
+ args = parser.parse_args(args)
+
+ return args
+
+
+async def shutdown_handler(tux: Tux, signal_type, exit_code=None) -> NoReturn:
+ """Handler when the bot shutdown
+
+ It cancels all running task.
+
+ Parameters
+ ----------
+ tux:Tux
+ Object for the bot.
+ signal_type:int, None
+ Exiting signal code.
+ exit_code:None|int
+ Code to show when exiting.
+ """
+ if signal_type:
+ log.info("%s received. Quitting...", signal_type)
+ elif exit_code is None:
+ log.info("Shutting down from unhandled exception")
+ tux.shutdown_code = ExitCodes.CRITICAL
+
+ if exit_code is not None:
+ tux.shutdown_code = exit_code
+
+ await tux.shutdown()
+
+
+async def run_bot(tux: Tux, cli_flags: Namespace) -> None:
+ """This run the bot.
+
+ Parameters
+ ----------
+ tux:Tux
+ Object for the bot.
+ cli_flags:Namespace
+ All different flags passed in the console.
+
+ Returns
+ -------
+ None
+ When exiting, this function return None.
+ """
+ data_path = data_manager.data_path(tux.instance_name)
+
+ tuxbot.logging.init_logging(10, location=data_path / "logs")
+
+ log.debug("====Basic Config====")
+ log.debug("Data Path: %s", data_path)
+
+ if cli_flags.token:
+ token = cli_flags.token
+ else:
+ token = tux.config("core").get("token")
+
+ if not token:
+ log.critical("Token must be set if you want to login.")
+ sys.exit(ExitCodes.CRITICAL)
+
+ try:
+ await tux.load_packages()
+ console.print()
+ await tux.start(token=token, bot=True)
+ except discord.LoginFailure:
+ log.critical("This token appears to be valid.")
+ console.print()
+ console.print(
+ "[prompt.invalid]This token appears to be valid. [i]exiting...[/i]"
+ )
+ sys.exit(ExitCodes.CRITICAL)
+ except Exception as e:
+ raise e
+
+ return None
+
+
+def run() -> NoReturn:
+ """Main function
+
+ """
+ tux = None
+ cli_flags = parse_cli_flags(sys.argv[1:])
+
+ if cli_flags.list_instances:
+ list_instances()
+ elif cli_flags.debug:
+ debug_info()
+ elif cli_flags.version:
+ print(f"Tuxbot V{version_info.major}")
+ print(f"Complete Version: {__version__}")
+
+ sys.exit(os.EX_OK)
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+
+ try:
+ if not cli_flags.instance_name:
+ console.print(
+ "[red]No instance provided ! "
+ "You can use 'tuxbot -L' to list all available instances"
+ )
+ sys.exit(ExitCodes.CRITICAL)
+
+ tux = Tux(
+ cli_flags=cli_flags,
+ description="Tuxbot, made from and for OpenSource",
+ dm_help=None,
+ )
+
+ loop.run_until_complete(run_bot(tux, cli_flags))
+ except KeyboardInterrupt:
+ console.print(
+ " [red]Please use <prefix>quit instead of Ctrl+C to Shutdown!"
+ )
+ log.warning("Please use <prefix>quit instead of Ctrl+C to Shutdown!")
+ log.info("Received KeyboardInterrupt")
+ console.print("[i]Trying to shutdown...")
+ if tux is not None:
+ loop.run_until_complete(shutdown_handler(tux, signal.SIGINT))
+ except SystemExit as exc:
+ log.info("Shutting down with exit code: %s", exc.code)
+ if tux is not None:
+ loop.run_until_complete(shutdown_handler(tux, None, exc.code))
+ except Exception as exc:
+ log.error("Unexpected exception (%s): ", type(exc))
+ console.print_exception()
+ if tux is not None:
+ loop.run_until_complete(shutdown_handler(tux, None, 1))
+ finally:
+ loop.run_until_complete(loop.shutdown_asyncgens())
+ log.info("Please wait, cleaning up a bit more")
+ loop.run_until_complete(asyncio.sleep(1))
+ asyncio.set_event_loop(None)
+ loop.stop()
+ loop.close()
+ exit_code = ExitCodes.CRITICAL if tux is None else tux.shutdown_code
+
+ sys.exit(exit_code)
diff --git a/tuxbot/cogs/admin/__init__.py b/tuxbot/cogs/admin/__init__.py
index a1281fa..11565d1 100644
--- a/tuxbot/cogs/admin/__init__.py
+++ b/tuxbot/cogs/admin/__init__.py
@@ -1,6 +1,7 @@
from collections import namedtuple
from .admin import Admin
+from .config import AdminConfig
from ...core.bot import Tux
VersionInfo = namedtuple("VersionInfo", "major minor micro release_level")
diff --git a/tuxbot/cogs/admin/admin.py b/tuxbot/cogs/admin/admin.py
index 5957825..1da8cff 100644
--- a/tuxbot/cogs/admin/admin.py
+++ b/tuxbot/cogs/admin/admin.py
@@ -6,7 +6,10 @@ from discord.ext import commands
from tuxbot.core import checks
from tuxbot.core.bot import Tux
from tuxbot.core.i18n import Translator, find_locale, get_locale_name, available_locales
-from tuxbot.core.utils.functions.extra import group_extra, ContextPlus
+from tuxbot.core.utils.functions.extra import (
+ group_extra, command_extra,
+ ContextPlus
+)
log = logging.getLogger("tuxbot.cogs.admin")
_ = Translator("Admin", __file__)
@@ -40,10 +43,30 @@ class Admin(commands.Cog, name="Admin"):
@_lang.command(name="list", aliases=["liste", "all", "view"])
async def _lang_list(self, ctx: ContextPlus):
+ description = ''
+ for key, value in available_locales.items():
+ description += f":flag_{key[-2:].lower()}: {value[0]}\n"
+
e = discord.Embed(
title=_("List of available locales: ", ctx, self.bot.config),
- description="\n".join([i[0] for i in available_locales.values()]),
+ description=description,
color=0x36393E,
)
await ctx.send(embed=e)
+
+ # =========================================================================
+
+ @command_extra(name="quit", aliases=["shutdown"], deletable=False)
+ @commands.guild_only()
+ @checks.is_owner()
+ async def _quit(self, ctx: ContextPlus):
+ await ctx.send("*quit...*")
+ await self.bot.shutdown()
+
+ @command_extra(name="restart", deletable=False)
+ @commands.guild_only()
+ @checks.is_owner()
+ async def _restart(self, ctx: ContextPlus):
+ await ctx.send("*restart...*")
+ await self.bot.shutdown(restart=True)
diff --git a/tuxbot/cogs/admin/config.py b/tuxbot/cogs/admin/config.py
new file mode 100644
index 0000000..4ec934f
--- /dev/null
+++ b/tuxbot/cogs/admin/config.py
@@ -0,0 +1,18 @@
+from structured_config import Structure, StrField
+
+
+class AdminConfig(Structure):
+ dm: str = StrField("")
+ mentions: str = StrField("")
+ guilds: str = StrField("")
+ errors: str = StrField("")
+ gateway: str = StrField("")
+
+
+extra = {
+ 'dm': str,
+ 'mentions': str,
+ 'guilds': str,
+ 'errors': str,
+ 'gateway': str,
+}
diff --git a/tuxbot/core/bot.py b/tuxbot/core/bot.py
index 3d84267..f924a5a 100644
--- a/tuxbot/core/bot.py
+++ b/tuxbot/core/bot.py
@@ -25,7 +25,10 @@ log = logging.getLogger("tuxbot")
console = Console()
install(console=console)
-packages: List[str] = ["jishaku", "tuxbot.cogs.warnings", "tuxbot.cogs.admin"]
+packages: List[str] = [
+ "jishaku",
+ "tuxbot.cogs.admin"
+]
class Tux(commands.AutoShardedBot):
@@ -109,6 +112,7 @@ class Tux(commands.AutoShardedBot):
self._progress.get("main").remove_task(
self._progress.get("tasks")["connecting"]
)
+ self._progress.get("tasks").pop("connecting")
console.clear()
console.print(
@@ -155,8 +159,8 @@ class Tux(commands.AutoShardedBot):
console.print(columns)
console.print()
- async def is_owner(self,
- user: Union[discord.User, discord.Member]) -> bool:
+ async def is_owner(self, user: Union[discord.User, discord.Member])\
+ -> bool:
"""Determines if the user is a bot owner.
Parameters
@@ -245,7 +249,7 @@ class Tux(commands.AutoShardedBot):
for task in pending:
console.log("Canceling", task.get_name(), f"({task.get_coro()})")
task.cancel()
- await asyncio.gather(*pending, return_exceptions=True)
+ await asyncio.gather(*pending, return_exceptions=False)
await super().logout()
@@ -265,4 +269,8 @@ class Tux(commands.AutoShardedBot):
self.shutdown_code = ExitCodes.RESTART
await self.logout()
- sys.exit(self.shutdown_code)
+
+ sys_e = SystemExit()
+ sys_e.code = self.shutdown_code
+
+ raise sys_e
diff --git a/tuxbot/core/config.py b/tuxbot/core/config.py
index 5a4702c..0cd6b9c 100644
--- a/tuxbot/core/config.py
+++ b/tuxbot/core/config.py
@@ -1,8 +1,10 @@
import asyncio
-import json
import logging
-from typing import List, Dict, Union, Any
-from flatten_dict import flatten, unflatten
+from typing import List, Dict
+from structured_config import (
+ ConfigFile,
+ Structure, IntField, StrField, BoolField
+)
import discord
@@ -13,159 +15,44 @@ __all__ = ["Config"]
log = logging.getLogger("tuxbot.core.config")
-class Config:
- def __init__(self, cog_instance: str = None):
- self._cog_instance = cog_instance
+class Server(Structure):
+ prefixes: List[str] = []
+ disabled_command: List[str] = []
+ locale: str = StrField("")
- self.lock = asyncio.Lock()
- self.loop = asyncio.get_event_loop()
- self._settings_file = None
- self._datas = {}
+class User(Structure):
+ aliases: List[dict] = []
+ locale: str = StrField("")
- def __getitem__(self, item) -> Dict:
- path = data_path(self._cog_instance)
- if item != "core":
- path = path / "cogs" / item
- else:
- path /= "core"
+class Config(Structure):
+ class Servers(Structure):
+ count: int = IntField(0)
+ all: List[Server] = []
- settings_file = path / "settings.json"
+ class Users(Structure):
+ all: List[User] = []
- if not settings_file.exists():
- raise FileNotFoundError(
- f"Unable to find settings file " f"'{settings_file}'"
- )
- else:
- with settings_file.open("r") as f:
- return json.load(f)
+ class Core(Structure):
+ owners_id: List[int] = []
+ prefixes: List[str] = []
+ token: str = StrField("")
+ mentionable: bool = BoolField("")
+ locale: str = StrField("")
- def __call__(self, item):
- return self.__getitem__(item)
+ class Cogs(Structure):
+ pass
- def owners_id(self) -> List[int]:
- """Simply return the owners id saved in config file.
- Returns
- -------
- str
- Owners id.
- """
- return self.__getitem__("core").get("owners_id")
+# =============================================================================
+# Configuration of Tuxbot Application (not the bot)
+# =============================================================================
- def token(self) -> str:
- """Simply return the bot token saved in config file.
+class Instance(Structure):
+ path: str = StrField("")
+ active: bool = BoolField(False)
- Returns
- -------
- str
- Bot token.
- """
- return self.__getitem__("core").get("token")
- def get_prefixes(self, guild: discord.Guild) -> List[str]:
- """Get custom prefixes for one guild.
-
- Parameters
- ----------
- guild:discord.Guild
- The required guild prefixes.
-
- Returns
- -------
- List[str]
- List of all prefixes.
- """
- core = self.__getitem__("core")
- prefixes = core.get("guild", {}).get(guild.id, {}).get("prefixes", [])
-
- return prefixes
-
- def get_blacklist(self, key: str) -> List[Union[str, int]]:
- """Return list off all blacklisted values
-
- Parameters
- ----------
- key:str
- Which type of blacklist to choice (guilds ? channels ?,...).
-
- Returns
- -------
- List[Union[str, int]]
- List containing blacklisted values.
- """
- core = self.__getitem__("core")
- blacklist = core.get("blacklist", {}).get(key, [])
-
- return blacklist
-
- def _dump(self):
- with self._settings_file.open("w") as f:
- json.dump(self._datas, f, indent=4)
-
- async def update(self, cog_name: str, item: str, value: Any) -> dict:
- """Update values in config file.
-
- Parameters
- ----------
- cog_name:str
- Name of cog who's corresponding to the config file.
- item:str
- Key to update.
- value:Any
- New values to apply.
-
- Returns
- -------
- dict:
- Updated values.
-
- """
- datas = self.__getitem__(cog_name)
- path = data_path(self._cog_instance)
-
- flat_datas = flatten(datas)
- flat_datas[tuple(item.split("."))] = value
- datas = unflatten(flat_datas)
-
- self._datas = datas
-
- if cog_name != "core":
- path = path / "cogs" / cog_name
- else:
- path /= "core"
-
- self._settings_file = path / "settings.json"
-
- async with self.lock:
- await self.loop.run_in_executor(None, self._dump)
-
- return datas
-
- def get_value(self, cog_name: str, key: str, default: Any = None) -> Any:
- """Get value by key.
-
- Parameters
- ----------
- cog_name:str
- Name of cog who's corresponding to the config file.
- key:str
- Key to fetch.
- default:Any|Optional
- Default value.
-
- Returns
- -------
- Any:
- Recovered value.
-
- """
- datas = self.__getitem__(cog_name)
-
- flat_datas = flatten(datas)
-
- try:
- return flat_datas[tuple(key.split("."))]
- except KeyError:
- return default
+class AppConfig(Structure):
+ instances: Dict[str, Instance] = {}
diff --git a/tuxbot/core/data_manager.py b/tuxbot/core/data_manager.py
index d96661c..a6c438f 100644
--- a/tuxbot/core/data_manager.py
+++ b/tuxbot/core/data_manager.py
@@ -7,7 +7,7 @@ log = logging.getLogger("tuxbot.core.data_manager")
app_dir = appdirs.AppDirs("Tuxbot-bot")
config_dir = Path(app_dir.user_config_dir)
-config_file = config_dir / "config.json"
+config_file = config_dir / "config.yaml"
def data_path(instance_name: str) -> Path:
diff --git a/tuxbot/core/exceptions.py b/tuxbot/core/exceptions.py
new file mode 100644
index 0000000..d47a97f
--- /dev/null
+++ b/tuxbot/core/exceptions.py
@@ -0,0 +1,9 @@
+from discord.ext import commands
+
+
+class DisabledCommandByServerOwner(commands.CheckFailure):
+ pass
+
+
+class DisabledCommandByBotOwner(commands.CheckFailure):
+ pass
diff --git a/tuxbot/core/i18n.py b/tuxbot/core/i18n.py
index 4190dc0..51e0fd1 100644
--- a/tuxbot/core/i18n.py
+++ b/tuxbot/core/i18n.py
@@ -38,14 +38,14 @@ def get_locale_name(locale: str) -> str:
class Translator(Callable[[str], str]):
"""Class to load texts at init."""
- def __init__(self, name: str, file_location: Union[str, Path, os.PathLike]):
+ def __init__(self, name: str, file_location: Union[Path, os.PathLike]):
"""Initializes the Translator object.
Parameters
----------
name : str
The cog name.
- file_location:str|Path|os.PathLike
+ file_location:Path|os.PathLike
File path for the required extension.
"""
diff --git a/tuxbot/core/utils/functions/cli.py b/tuxbot/core/utils/functions/cli.py
deleted file mode 100644
index 56bc67a..0000000
--- a/tuxbot/core/utils/functions/cli.py
+++ /dev/null
@@ -1,89 +0,0 @@
-import codecs
-import itertools
-import sys
-
-
-def bordered(*columns: dict) -> str:
- """
- credits to https://github.com/Cog-Creators/Red-DiscordBot/blob/V3/develop/redbot/core/utils/chat_formatting.py
-
- Get two blocks of text in a borders.
-
- Note
- ----
- This will only work with a monospaced font.
-
- Parameters
- ----------
- *columns : `sequence` of `str`
- The columns of text, each being a list of lines in that column.
-
- Returns
- -------
- str
- The bordered text.
-
- """
- encoder = codecs.getencoder(sys.stdout.encoding)
- try:
- encoder("┌┐└┘─│") # border symbols
- except UnicodeEncodeError:
- ascii_border = True
- else:
- ascii_border = False
-
- borders = {
- "TL": "+" if ascii_border else "┌", # Top-left
- "TR": "+" if ascii_border else "┐", # Top-right
- "BL": "+" if ascii_border else "└", # Bottom-left
- "BR": "+" if ascii_border else "┘", # Bottom-right
- "HZ": "-" if ascii_border else "─", # Horizontal
- "VT": "|" if ascii_border else "│", # Vertical
- }
-
- sep = " " * 4 # Separator between boxes
- widths = tuple(
- max(len(row) for row in column.get("rows")) + 9 for column in columns
- ) # width of each col
- cols_done = [False] * len(columns) # whether or not each column is done
- lines = [""]
-
- for i, column in enumerate(columns):
- lines[0] += (
- "{TL}"
- + "{HZ}"
- + column.get("title")
- + "{HZ}" * (widths[i] - len(column.get("title")) - 1)
- + "{TR}"
- + sep
- )
-
- for line in itertools.zip_longest(*[column.get("rows") for column in columns]):
- row = []
- for colidx, column in enumerate(line):
- width = widths[colidx]
- done = cols_done[colidx]
- if column is None:
- if not done:
- # bottom border of column
- column = "{HZ}" * width
- row.append("{BL}" + column + "{BR}")
- cols_done[colidx] = True # mark column as done
- else:
- # leave empty
- row.append(" " * (width + 2))
- else:
- column += " " * (width - len(column)) # append padded spaces
- row.append("{VT}" + column + "{VT}")
-
- lines.append(sep.join(row))
-
- final_row = []
- for width, done in zip(widths, cols_done):
- if not done:
- final_row.append("{BL}" + "{HZ}" * width + "{BR}")
- else:
- final_row.append(" " * (width + 2))
- lines.append(sep.join(final_row))
-
- return "\n".join(lines).format(**borders)
diff --git a/tuxbot/core/utils/functions/extra.py b/tuxbot/core/utils/functions/extra.py
index 98c48ca..ace86f5 100644
--- a/tuxbot/core/utils/functions/extra.py
+++ b/tuxbot/core/utils/functions/extra.py
@@ -5,6 +5,11 @@ import discord
from discord import Embed
from discord.ext import commands, flags
+from rich.console import Console
+console = Console()
+
+console.clear()
+
class ContextPlus(commands.Context):
async def send(self, content=None, *args, **kwargs):
@@ -16,12 +21,11 @@ class ContextPlus(commands.Context):
e = str(kwargs.get('embed').to_dict())
e = e.replace(self.bot.config('core').get('token'), '<token>')
e = yaml.load(e, Loader=yaml.FullLoader)
-
kwargs['embed'] = Embed.from_dict(e)
if (
hasattr(self.command, "deletable") and self.command.deletable
- ) and kwargs.pop("deletable", True):
+ ) or kwargs.pop("deletable", False):
message = await super().send(content, *args, **kwargs)
await message.add_reaction("🗑")
@@ -33,7 +37,10 @@ class ContextPlus(commands.Context):
)
try:
- await self.bot.wait_for("reaction_add", timeout=45.0, check=check)
+ await self.bot.wait_for(
+ "reaction_add",
+ timeout=42.0, check=check
+ )
except asyncio.TimeoutError:
await message.remove_reaction("🗑", self.bot.user)
else:
diff --git a/tuxbot/setup.py b/tuxbot/setup.py
index 38d9b6c..5aa8858 100644
--- a/tuxbot/setup.py
+++ b/tuxbot/setup.py
@@ -9,9 +9,9 @@ from rich.prompt import Prompt, IntPrompt
from rich.console import Console
from rich.rule import Rule
from rich.traceback import install
-from rich import print
from tuxbot.core.data_manager import config_dir, app_dir
+from tuxbot.core import config
console = Console()
console.clear()
@@ -20,56 +20,15 @@ install(console=console)
try:
config_dir.mkdir(parents=True, exist_ok=True)
except PermissionError:
- print(f"mkdir: cannot create directory '{config_dir}': Permission denied")
+ console.print(f"mkdir: cannot create directory '{config_dir}': Permission denied")
sys.exit(1)
-config_file = config_dir / "config.json"
+app_config = config.ConfigFile(config_dir / "config.yaml", config.AppConfig)
-
-def load_existing_config() -> dict:
- """Loading and returning configs.
-
- Returns
- -------
- dict
- a dict containing all configurations.
-
- """
- if not config_file.exists():
- return {}
-
- with config_file.open() as fs:
- return json.load(fs)
-
-
-instances_data = load_existing_config()
-if not instances_data:
+if not app_config.config.instances:
instances_list = []
else:
- instances_list = list(instances_data.keys())
-
-
-def save_config(name: str, data: dict, delete=False) -> NoReturn:
- """save data in config file.
-
- Parameters
- ----------
- name:str
- name of instance.
- data:dict
- settings for `name` instance.
- delete:bool
- delete or no data.
- """
- _config = load_existing_config()
-
- if delete and name in _config:
- _config.pop(name)
- else:
- _config[name] = data
-
- with config_file.open("w") as fs:
- json.dump(_config, fs, indent=4)
+ instances_list = list(app_config.config.instances.keys())
def get_name() -> str:
@@ -89,8 +48,8 @@ def get_name() -> str:
console=console
)
if re.fullmatch(r"[a-zA-Z0-9_\-]*", name) is None:
- print()
- print("[prompt.invalid]ERROR: Invalid characters provided")
+ console.print()
+ console.print("[prompt.invalid]ERROR: Invalid characters provided")
name = ""
return name
@@ -111,14 +70,14 @@ def get_data_dir(instance_name: str) -> Path:
"""
data_path = Path(app_dir.user_data_dir) / "data" / instance_name
data_path_input = ""
- print()
+ console.print()
def make_data_dir(path: Path) -> Union[Path, str]:
try:
path.mkdir(parents=True, exist_ok=True)
except OSError:
- print()
- print(
+ console.print()
+ console.print(
f"mkdir: cannot create directory '{path}': Permission denied"
)
path = ""
@@ -137,8 +96,8 @@ def get_data_dir(instance_name: str) -> Path:
try:
exists = data_path_input.exists()
except OSError:
- print()
- print(
+ console.print()
+ console.print(
"[prompt.invalid]"
"Impossible to verify the validity of the path,"
" make sure it does not contain any invalid characters."
@@ -149,8 +108,8 @@ def get_data_dir(instance_name: str) -> Path:
if data_path_input and not exists:
data_path_input = make_data_dir(data_path_input)
- print()
- print(
+ console.print()
+ console.print(
f"You have chosen {data_path_input} to be your config directory for "
f"`{instance_name}` instance"
)
@@ -160,7 +119,7 @@ def get_data_dir(instance_name: str) -> Path:
choices=["y", "n"], default="y",
console=console
) != "y":
- print("Rerun the process to redo this configuration.")
+ console.print("Rerun the process to redo this configuration.")
sys.exit(0)
(data_path_input / "core").mkdir(parents=True, exist_ok=True)
@@ -191,7 +150,7 @@ def get_token() -> str:
r"|mfa\.[a-zA-Z0-9_\-]{84})",
token) \
is None:
- print("[prompt.invalid]ERROR: Invalid token provided")
+ console.print("[prompt.invalid]ERROR: Invalid token provided")
token = ""
return token
@@ -234,7 +193,7 @@ def get_multiple(
if new not in values:
values.append(new)
else:
- print(
+ console.print(
f"[prompt.invalid]"
f"ERROR: `{new}` is already present, [i]ignored[/i]"
)
@@ -250,21 +209,21 @@ def additional_config() -> dict:
dict:
Dict with cog name as key and configs as value.
"""
- p = Path(r"tuxbot/cogs").glob("**/additional_config.json")
+ p = Path("tuxbot/cogs").glob("**/config.py")
data = {}
for file in p:
- print("\n" * 4)
+ console.print("\n" * 4)
cog_name = str(file.parent).split("/")[-1]
data[cog_name] = {}
with file.open("r") as f:
data = json.load(f)
- print(Rule(f"\nConfiguration for `{cog_name}` module"))
+ console.print(Rule(f"\nConfiguration for `{cog_name}` module"))
for key, value in data.items():
- print()
+ console.print()
data[cog_name][key] = Prompt.ask(value["description"])
return data
@@ -278,79 +237,62 @@ def finish_setup(data_dir: Path) -> NoReturn:
data_dir:Path
Where to save configs.
"""
- print(
+ console.print(
Rule(
"Now, it's time to finish this setup by giving bot information"
)
)
- print()
+ console.print()
token = get_token()
- print()
+ console.print()
prefixes = get_multiple(
"Choice a (or multiple) prefix for the bot", "Add another prefix ?",
str
)
- print()
+ console.print()
mentionable = Prompt.ask(
"Does the bot answer if it's mentioned?",
choices=["y", "n"],
default="y"
) == "y"
- print()
+ console.print()
owners_id = get_multiple(
"Give the owner id of this bot", "Add another owner ?", int
)
- cogs_config = additional_config()
+ # cogs_config = additional_config()
- core_file = data_dir / "core" / "settings.json"
- core = {
- "token": token,
- "prefixes": prefixes,
- "mentionable": mentionable,
- "owners_id": owners_id,
- "locale": "en-US",
- }
+ instance_config = config.ConfigFile(
+ str(data_dir / "config.yaml"), config.Config
+ )
- with core_file.open("w") as fs:
- json.dump(core, fs, indent=4)
-
- for cog, data in cogs_config.items():
- data_cog_dir = data_dir / "cogs" / cog
- data_cog_dir.mkdir(parents=True, exist_ok=True)
-
- data_cog_file = data_cog_dir / "settings.json"
-
- with data_cog_file.open("w") as fs:
- json.dump(data, fs, indent=4)
+ instance_config.config.Core.owners_id = owners_id
+ instance_config.config.Core.prefixes = prefixes
+ instance_config.config.Core.token = token
+ instance_config.config.Core.mentionable = mentionable
+ instance_config.config.Core.locale = "en-US"
def basic_setup() -> NoReturn:
"""Configs who refer to instances.
"""
- print(
+ console.print(
Rule(
"Hi ! it's time for you to give me information about you instance"
)
)
- print()
+ console.print()
name = get_name()
data_dir = get_data_dir(name)
- configs = load_existing_config()
- instance_config = configs[name] if name in instances_list else {}
-
- instance_config["DATA_PATH"] = str(data_dir.resolve())
- instance_config["IS_RUNNING"] = False
-
if name in instances_list:
- print()
+ console.print()
console.print(
f"WARNING: An instance named `{name}` already exists "
f"Continuing will overwrite this instance configs.", style="red"
@@ -359,17 +301,21 @@ def basic_setup() -> NoReturn:
"Are you sure you want to continue?",
choices=["y", "n"], default="n"
) == "n":
- print("Abandon...")
+ console.print("Abandon...")
sys.exit(0)
- save_config(name, instance_config)
+ instance = config.Instance()
+ instance.path = str(data_dir.resolve())
+ instance.active = False
- print("\n" * 4)
+ app_config.config.instances[name] = instance
+
+ console.print("\n" * 4)
finish_setup(data_dir)
- print()
- print(
+ console.print()
+ console.print(
f"Instance successfully created! "
f"You can now run `tuxbot {name}` to launch this instance"
)
@@ -392,8 +338,8 @@ def setup() -> NoReturn:
basic_setup()
except KeyboardInterrupt:
- print("Exiting...")
- except:
+ console.print("Exiting...")
+ except Exception:
console.print_exception()