264 lines
10 KiB
Python
264 lines
10 KiB
Python
import time
|
|
import json
|
|
import sys
|
|
import traceback
|
|
import discord
|
|
import inspect
|
|
|
|
# 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()
|
|
|
|
###############################
|
|
# Simple Logging System
|
|
###############################
|
|
|
|
def log(message: str, level="INFO", exec_info=False, linebreaks=False):
|
|
"""
|
|
Log a message with the specified log level.
|
|
|
|
Capable of logging individual levels to the terminal and/or logfile separately.
|
|
Can also append traceback information if needed, and is capable of preserving/removing
|
|
linebreaks from log messages as needed. Also prepends the calling function name.
|
|
|
|
Args:
|
|
message (str): The message to log.
|
|
level (str, optional): Log level of the message. Defaults to "INFO".
|
|
exec_info (bool, optional): If True, append traceback information. Defaults to False.
|
|
linebreaks (bool, optional): If True, preserve line breaks in the log. Defaults to False.
|
|
|
|
Available levels:
|
|
DEBUG - Information useful for debugging.
|
|
INFO - Informational messages.
|
|
WARNING - Something happened that may lead to issues.
|
|
ERROR - A non-critical error has occurred.
|
|
CRITICAL - A critical, but non-fatal, error occurred.
|
|
FATAL - Fatal error; program exits after logging this.
|
|
RESTART - Graceful restart.
|
|
SHUTDOWN - Graceful exit.
|
|
|
|
See:
|
|
config.json for further configuration options under "logging".
|
|
|
|
Example:
|
|
log("An error occured during processing", "ERROR", exec_info=True, linebreaks=False)
|
|
"""
|
|
|
|
# Hard-coded options/settings (can be expanded as needed)
|
|
default_level = "INFO" # Fallback log level
|
|
allowed_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "FATAL", "RESTART", "SHUTDOWN"}
|
|
|
|
# Ensure valid level
|
|
if level not in allowed_levels:
|
|
level = default_level
|
|
|
|
# Capture the calling function's name using inspect
|
|
try:
|
|
caller_frame = inspect.stack()[1]
|
|
caller_name = caller_frame.function
|
|
except Exception:
|
|
caller_name = "Unknown"
|
|
|
|
# Optionally remove linebreaks if not desired
|
|
if not linebreaks:
|
|
message = message.replace("\n", " ")
|
|
|
|
# Get current timestamp and uptime
|
|
elapsed = time.time() - get_bot_start_time() # Assuming this function is defined elsewhere
|
|
from modules import utility
|
|
uptime_str, _ = utility.format_uptime(elapsed)
|
|
timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
# Prepend dynamic details including the caller name
|
|
log_message = f"[{timestamp} - {uptime_str}] [{level}] [Func: {caller_name}] {message}"
|
|
|
|
# Append traceback if required or for error levels
|
|
if exec_info or level in {"ERROR", "CRITICAL", "FATAL"}:
|
|
log_message += f"\n{traceback.format_exc()}"
|
|
|
|
# Read logging settings from the configuration
|
|
lfp = config_data["logging"]["logfile_path"] # Log File Path
|
|
clfp = f"cur_{lfp}" # Current Log File Path
|
|
|
|
if not (config_data["logging"]["terminal"]["log_to_terminal"] or
|
|
config_data["logging"]["file"]["log_to_file"]):
|
|
print("!!! WARNING !!! CONSOLE AND LOGFILE OUTPUT DISABLED !!!\n"
|
|
"!!! NO LOGS WILL BE PROVIDED !!!")
|
|
|
|
# Check if this log level is enabled (or if it's a FATAL message which always prints)
|
|
if level in config_data["logging"]["log_levels"] or level == "FATAL":
|
|
# Terminal output
|
|
if config_data["logging"]["terminal"]["log_to_terminal"] or level == "FATAL":
|
|
config_level_format = f"log_{level.lower()}"
|
|
if config_data["logging"]["terminal"].get(config_level_format, False) or level == "FATAL":
|
|
print(log_message)
|
|
|
|
# File output
|
|
if config_data["logging"]["file"]["log_to_file"] or level == "FATAL":
|
|
config_level_format = f"log_{level.lower()}"
|
|
if config_data["logging"]["file"].get(config_level_format, False) or level == "FATAL":
|
|
try:
|
|
with open(lfp, "a", encoding="utf-8") as logfile:
|
|
logfile.write(f"{log_message}\n")
|
|
logfile.flush()
|
|
with open(clfp, "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}")
|
|
|
|
# Handle fatal errors with 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 reset_curlogfile():
|
|
"""
|
|
Clear the current log file.
|
|
|
|
This function constructs the current log file path by prepending 'cur_'
|
|
to the log file path specified in the configuration data under the "logging"
|
|
section. It then opens the file in write mode, effectively truncating and
|
|
clearing its contents.
|
|
|
|
If an exception occurs while attempting to clear the log file, the error is
|
|
silently ignored.
|
|
"""
|
|
# Initiate logfile
|
|
lfp = config_data["logging"]["logfile_path"] # Log File Path
|
|
clfp = f"cur_{lfp}" # Current Log File Path
|
|
|
|
try:
|
|
open(clfp, "w")
|
|
# log(f"Current-run logfile cleared", "DEBUG")
|
|
except Exception as e:
|
|
# log(f"Failed to clear current-run logfile: {e}")
|
|
pass
|
|
|
|
|
|
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.
|
|
log("Terminating bot due to no DB connection.", "FATAL")
|
|
sys.exit(1)
|
|
return db_conn
|
|
except Exception as e:
|
|
log(f"Unable to initialize database!: {e}", "FATAL")
|
|
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:
|
|
log("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() |