244 lines
6.8 KiB
Python
244 lines
6.8 KiB
Python
import argparse
|
|
import asyncio
|
|
import logging
|
|
import signal
|
|
import sys
|
|
import os
|
|
from argparse import Namespace
|
|
|
|
import discord
|
|
import pip
|
|
from rich.columns import Columns
|
|
from rich.panel import Panel
|
|
from rich.table import Table, box
|
|
from rich import print as rprint
|
|
|
|
import tuxbot.logging
|
|
from tuxbot.core.bot import Tux
|
|
from tuxbot.core.utils import data_manager
|
|
from tuxbot.core.utils.console import console
|
|
from . import __version__, version_info, ExitCodes
|
|
|
|
log = logging.getLogger("tuxbot.main")
|
|
|
|
BORDER_STYLE = "not dim"
|
|
|
|
|
|
def debug_info() -> None:
|
|
"""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("/usr/bin/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=BORDER_STYLE, 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=BORDER_STYLE, 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=BORDER_STYLE, 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][:-1]}")
|
|
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 [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(
|
|
"--token", "-T", type=str, help="Run Tuxbot with passed token"
|
|
)
|
|
|
|
return parser.parse_args(args)
|
|
|
|
|
|
async def shutdown_handler(tux: Tux, signal_type, exit_code=None) -> None:
|
|
"""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
|
|
|
|
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.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)
|
|
except discord.LoginFailure:
|
|
log.critical("This token appears to be invalid.")
|
|
console.print()
|
|
console.print(
|
|
"[prompt.invalid]This token appears to be valid. [i]exiting...[/i]"
|
|
)
|
|
sys.exit(ExitCodes.CRITICAL)
|
|
except Exception as e:
|
|
log.critical(e)
|
|
raise e
|
|
|
|
return None
|
|
|
|
|
|
def run() -> None:
|
|
"""Main function"""
|
|
tux = None
|
|
cli_flags = parse_cli_flags(sys.argv[1:])
|
|
|
|
if cli_flags.debug:
|
|
debug_info()
|
|
elif cli_flags.version:
|
|
rprint(f"Tuxbot V{version_info.major}")
|
|
rprint(f"Complete Version: {__version__}")
|
|
|
|
sys.exit(os.EX_OK)
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
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))
|
|
raise
|
|
except Exception as exc:
|
|
log.error("Unexpected exception (%s): ", type(exc))
|
|
console.print_exception(show_locals=True)
|
|
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)
|