349 lines
12 KiB
Python
349 lines
12 KiB
Python
import time
|
|
import json
|
|
import sys
|
|
import traceback
|
|
import discord
|
|
import inspect
|
|
import os
|
|
|
|
# Store the start time globally.
|
|
_bot_start_time = time.time()
|
|
|
|
def get_bot_start_time():
|
|
"""
|
|
Retrieve the bot's start time.
|
|
|
|
This function returns the Unix timestamp (in seconds) when the bot was started.
|
|
The timestamp is stored in the global variable `_bot_start_time`, which is set
|
|
when the module is first imported.
|
|
|
|
Returns:
|
|
float: The Unix timestamp representing the bot's start time.
|
|
"""
|
|
return _bot_start_time
|
|
|
|
def load_config_file():
|
|
"""
|
|
Load the configuration file.
|
|
|
|
This function attempts to read the JSON configuration from 'config.json'
|
|
in the current directory and return its contents as a dictionary. If the
|
|
file is not found or if the file contains invalid JSON, an error message
|
|
is printed and the program terminates with a non-zero exit code.
|
|
|
|
Returns:
|
|
dict: The configuration data loaded from 'config.json'.
|
|
|
|
Raises:
|
|
SystemExit: If 'config.json' is missing or cannot be parsed.
|
|
"""
|
|
CONFIG_PATH = "config.json"
|
|
try:
|
|
with open(CONFIG_PATH, "r") as f:
|
|
config_data = json.load(f)
|
|
return config_data
|
|
except FileNotFoundError:
|
|
print("Error: config.json not found.")
|
|
sys.exit(1)
|
|
except json.JSONDecodeError as e:
|
|
print(f"Error parsing config.json: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
# Load configuration file
|
|
config_data = load_config_file()
|
|
|
|
def load_settings_file(file: str):
|
|
"""
|
|
Load a settings file from the settings directory.
|
|
|
|
Args:
|
|
file (str): The name of the settings file, with or without the .json extension.
|
|
|
|
Returns:
|
|
dict: The configuration data loaded from the specified settings file.
|
|
"""
|
|
SETTINGS_PATH = "settings"
|
|
|
|
# Ensure the file has a .json extension
|
|
if not file.endswith(".json"):
|
|
file += ".json"
|
|
|
|
file_path = os.path.join(SETTINGS_PATH, file)
|
|
|
|
if not os.path.exists(file_path):
|
|
logger.fatal(f"Unable to read the settings file {file}!")
|
|
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
config_data = json.load(f)
|
|
return config_data
|
|
except json.JSONDecodeError as e:
|
|
logger.fatal(f"Error parsing {file}: {e}")
|
|
|
|
|
|
###############################
|
|
# Simple Logging System
|
|
###############################
|
|
class Logger:
|
|
"""
|
|
Custom logger class to handle different log levels, terminal & file output,
|
|
and system events (FATAL, RESTART, SHUTDOWN).
|
|
"""
|
|
|
|
allowed_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "FATAL", "RESTART", "SHUTDOWN"}
|
|
|
|
def __init__(self):
|
|
"""
|
|
Initializes the Logger with configurations from `config_data`.
|
|
"""
|
|
self.default_level = "INFO"
|
|
self.log_file_path = config_data["logging"]["logfile_path"]
|
|
self.current_log_file_path = f"cur_{self.log_file_path}"
|
|
self.log_to_terminal = config_data["logging"]["terminal"]["log_to_terminal"]
|
|
self.log_to_file = config_data["logging"]["file"]["log_to_file"]
|
|
self.enabled_log_levels = set(config_data["logging"]["log_levels"])
|
|
|
|
# Check if both logging outputs are disabled
|
|
if not self.log_to_terminal and not self.log_to_file:
|
|
print("!!! WARNING !!! LOGGING DISABLED !!! NO LOGS WILL BE PROVIDED !!!")
|
|
|
|
def log(self, message: str, level="INFO", exec_info=False, linebreaks=False):
|
|
"""
|
|
Logs a message at the specified log level.
|
|
|
|
Args:
|
|
message (str): The message to log.
|
|
level (str, optional): Log level. Defaults to "INFO".
|
|
exec_info (bool, optional): Append traceback information if True. Defaults to False.
|
|
linebreaks (bool, optional): Preserve line breaks if True. Defaults to False.
|
|
"""
|
|
from modules.utility import format_uptime
|
|
|
|
level = level.upper()
|
|
if level not in self.allowed_levels:
|
|
level = self.default_level
|
|
|
|
# Capture calling function name
|
|
caller_name = self.get_caller_function()
|
|
|
|
# Remove line breaks if required
|
|
if not linebreaks:
|
|
message = message.replace("\n", " ")
|
|
|
|
# Timestamp & uptime
|
|
elapsed = time.time() - get_bot_start_time() # Assuming this function exists
|
|
uptime_str, _ = format_uptime(elapsed)
|
|
timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
# Format log message
|
|
log_message = f"[{timestamp} - {uptime_str}] [{level}] [Func: {caller_name}] {message}"
|
|
|
|
# Append traceback if needed
|
|
if exec_info or level in {"ERROR", "CRITICAL", "FATAL"}:
|
|
log_message += f"\n{traceback.format_exc()}"
|
|
|
|
# Log to terminal
|
|
self._log_to_terminal(log_message, level)
|
|
|
|
# Log to file
|
|
self._log_to_file(log_message, level)
|
|
|
|
# Handle special log levels
|
|
self._handle_special_levels(level)
|
|
|
|
def _log_to_terminal(self, log_message: str, level: str):
|
|
"""
|
|
Outputs log messages to the terminal if enabled.
|
|
"""
|
|
if self.log_to_terminal or level == "FATAL":
|
|
if config_data["logging"]["terminal"].get(f"log_{level.lower()}", False) or level == "FATAL":
|
|
print(log_message)
|
|
|
|
def _log_to_file(self, log_message: str, level: str):
|
|
"""
|
|
Writes log messages to a file if enabled.
|
|
"""
|
|
if self.log_to_file or level == "FATAL":
|
|
if config_data["logging"]["file"].get(f"log_{level.lower()}", False) or level == "FATAL":
|
|
try:
|
|
with open(self.log_file_path, "a", encoding="utf-8") as logfile:
|
|
logfile.write(f"{log_message}\n")
|
|
logfile.flush()
|
|
with open(self.current_log_file_path, "a", encoding="utf-8") as c_logfile:
|
|
c_logfile.write(f"{log_message}\n")
|
|
c_logfile.flush()
|
|
except Exception as e:
|
|
print(f"[WARNING] Failed to write to logfile: {e}")
|
|
|
|
def _handle_special_levels(self, level: str):
|
|
"""
|
|
Handles special log levels like FATAL, RESTART, and SHUTDOWN.
|
|
"""
|
|
if level == "FATAL":
|
|
print("!!! FATAL ERROR LOGGED, SHUTTING DOWN !!!")
|
|
sys.exit(1)
|
|
|
|
if level == "RESTART":
|
|
print("!!! RESTART LOG LEVEL TRIGGERED, EXITING !!!")
|
|
sys.exit(0)
|
|
|
|
if level == "SHUTDOWN":
|
|
print("!!! SHUTDOWN LOG LEVEL TRIGGERED, EXITING !!!")
|
|
sys.exit(0)
|
|
|
|
def get_caller_function(self):
|
|
"""
|
|
Retrieves the calling function's name using `inspect`.
|
|
"""
|
|
try:
|
|
caller_frame = inspect.stack()[3]
|
|
return caller_frame.function
|
|
except Exception:
|
|
return "Unknown"
|
|
|
|
def reset_curlogfile(self):
|
|
"""
|
|
Clears the current log file.
|
|
"""
|
|
try:
|
|
open(self.current_log_file_path, "w").close()
|
|
except Exception as e:
|
|
print(f"[WARNING] Failed to clear current-run logfile: {e}")
|
|
|
|
## Shorter, cleaner methods for each log level
|
|
def debug(self, message: str, exec_info=False, linebreaks=False):
|
|
"""
|
|
Debug message for development troubleshooting.
|
|
"""
|
|
self.log(message, "DEBUG", exec_info, linebreaks)
|
|
|
|
def info(self, message: str, exec_info=False, linebreaks=False):
|
|
"""
|
|
General informational messages.
|
|
"""
|
|
self.log(message, "INFO", exec_info, linebreaks)
|
|
|
|
def warning(self, message: str, exec_info=False, linebreaks=False):
|
|
"""
|
|
Warning messages.
|
|
Something unusal happened, but shouldn't affect the program overall.
|
|
"""
|
|
self.log(message, "WARNING", exec_info, linebreaks)
|
|
|
|
def error(self, message: str, exec_info=True, linebreaks=False):
|
|
"""
|
|
Error messages.
|
|
Something didn't execute properly, but the bot overall should manage.
|
|
"""
|
|
self.log(message, "ERROR", exec_info, linebreaks)
|
|
|
|
def critical(self, message: str, exec_info=True, linebreaks=False):
|
|
"""
|
|
Critical error messages.
|
|
Something happened that is very likely to cause unintended behaviour and potential crashes.
|
|
"""
|
|
self.log(message, "CRITICAL", exec_info, linebreaks)
|
|
|
|
def fatal(self, message: str, exec_info=True, linebreaks=False):
|
|
"""
|
|
Fatal error messages.
|
|
Something happened that requires the bot to shut down somewhat nicely.
|
|
"""
|
|
self.log(message, "FATAL", exec_info, linebreaks)
|
|
|
|
def restart(self, message: str):
|
|
"""
|
|
Indicate bot restart log event.
|
|
"""
|
|
self.log(message, "RESTART")
|
|
|
|
def shutdown(self, message: str):
|
|
"""
|
|
Indicate bot shutdown log event.
|
|
"""
|
|
self.log(message, "SHUTDOWN")
|
|
|
|
|
|
# Instantiate Logger globally
|
|
logger = Logger()
|
|
|
|
#
|
|
#
|
|
#
|
|
|
|
def init_db_conn():
|
|
"""
|
|
Initialize and return a database connection.
|
|
|
|
This function reads the configuration settings and attempts to establish a
|
|
connection to the database by invoking `modules.db.init_db_connection()`. If
|
|
no valid connection is obtained (i.e. if the connection is None), it logs a
|
|
fatal error and terminates the program using sys.exit(1). If an exception is
|
|
raised during the initialization process, the error is logged and the function
|
|
returns None.
|
|
|
|
Returns:
|
|
DatabaseConnection or None: A valid database connection object if
|
|
successfully established; otherwise, None (or the program may exit if the
|
|
connection is missing).
|
|
"""
|
|
try:
|
|
import modules.db
|
|
db_conn = modules.db.init_db_connection(config_data)
|
|
if not db_conn:
|
|
# If we get None, it means a fatal error occurred.
|
|
logger.fatal("Terminating bot due to no DB connection.")
|
|
sys.exit(1)
|
|
return db_conn
|
|
except Exception as e:
|
|
logger.fatal(f"Unable to initialize database!: {e}")
|
|
return None
|
|
|
|
class Constants:
|
|
@property
|
|
def config_data(self) -> dict:
|
|
"""Returns a dictionary of the contents of the config.json config file"""
|
|
return load_config_file()
|
|
|
|
def bot_start_time(self) -> float:
|
|
"""Returns the bot epoch start time"""
|
|
return _bot_start_time
|
|
|
|
def primary_discord_guild(self) -> object | None:
|
|
"""
|
|
Retrieve the primary Discord guild from the configuration.
|
|
|
|
This function attempts to obtain the primary Discord guild based on the
|
|
configuration data stored in `config_data["discord_guilds"]`. It converts the first
|
|
guild ID in the list to an integer and then creates a `discord.Object` from it. If the
|
|
configuration defines more than one (or fewer than the expected number of) guilds, the function
|
|
returns `None` for the guild ID.
|
|
|
|
Returns:
|
|
dict: A dictionary with the following keys:
|
|
- "object": A `discord.Object` representing the primary Discord guild if exactly one
|
|
guild is defined; otherwise, `None`.
|
|
- "id": The integer ID of the primary guild if available; otherwise, `None`.
|
|
"""
|
|
# Checks for a True/False value in the config file to determine if commands sync should be global or single-guild
|
|
# If this is 'true' in config, it will
|
|
sync_commands_globally = config_data["sync_commands_globally"]
|
|
primary_guild_object = None
|
|
primary_guild_int = None
|
|
|
|
if not sync_commands_globally:
|
|
logger.info("Discord commands sync set to single-guild in config")
|
|
primary_guild_int = int(config_data["discord_guilds"][0]) if len(config_data["discord_guilds"]) > 0 else None
|
|
if primary_guild_int:
|
|
primary_guild_object = discord.Object(id=primary_guild_int)
|
|
|
|
return_dict = {"object": primary_guild_object, "id": primary_guild_int}
|
|
return return_dict
|
|
|
|
def twitch_channels_config(self):
|
|
with open("settings/twitch_channels_config.json", "r") as f:
|
|
CHANNEL_CONFIG = json.load(f)
|
|
return CHANNEL_CONFIG
|
|
|
|
constants = Constants() |