Bot

class Bot(token: str, channel: int, **kwargs)

Bases: object

Bot superclass.

Parameters:

  • token (str) – The token for bot authentication.

  • channel (int) – The bot channel ID.

  • **kwargs – Additional keyword arguments.

start_time

The bot initialization timestamp.

Type:

arrow.Arrow

token

The token for bot authentication.

Type:

str

channel

The bot channel ID.

Type:

int

has_remote

Whether the repository has a remote.

Type:

bool

repo

The git repository object, if has_remote.

Type:

git.Repo

repository_api_url

The GitHub API URL for the remote repository, if has_remote.

Type:

str

repository_url

The URL of the remote repository, if has_remote.

Type:

str

webpage_url

The URL of the GitHub pages webpage for the remote repository, if has_remote.

Type:

str

process

The variable to hold a process spawned by run_query.

Type:

multiprocessing.Process

cooldown_limit

The cooldown time in seconds to wait between consecutive calls to run_query.

Type:

Tnt

class Reports(value)

Bases: Enum

An enumeration.

abort()

Aborts a running YugiQuery flow by terminating the process.

Returns:

The result message indicating whether the abortion was successful.

Return type:

str

async battle(callback: Callable, atk_weight: int = 4, def_weight: int = 1)

This function loads the list of all Monster Cards and simulates a battle between them. Each card is represented by its name, attack (ATK), and defense (DEF) stats. At the beginning of the battle, a random card is chosen as the initial contestant. Then, for each subsequent card, a random stat (ATK or DEF) is chosen to compare with the corresponding stat of the current winner. If the challenger’s stat is higher, the challenger becomes the new winner. If the challenger’s stat is lower, the current winner retains its position. If the stats are tied, the comparison is repeated with the other stat. The battle continues until there is only one card left standing.

Parameters:

  • callback – Callable: A callback function which receives a string argument.

  • atk_weight (int, optional) – The weight to use for the ATK stat when randomly choosing the monster’s stat to compare. This affects the probability that ATK will be chosen over DEF. The default value is 4.

  • def_weight (int, optional) – The weight to use for the DEF stat when randomly choosing the monster’s stat to compare. This affects the probability that DEF will be chosen over ATK. The default value is 1.

Returns:

A dictionary containing information about the battle.

Return type:

dict

benchmark()

Returns the average time each report takes to complete and the latest time for each report.

Returns:

A dictionary containing benchmark information.

Return type:

dict

data()

Sends the latest data files available in the repository as direct download links.

Returns:

A dictionary containing direct links to the latest data files.

Return type:

dict

init_reports_enum()

Initializes and returns an Enum object containing the available reports. The reports are read from the yugiquery.SCRIPT_DIR directory, where they are expected to be Jupyter notebooks. The Enum object is created using the reports’ file names, with the .ipynb extension removed and the first letter capitalized.

Returns:

An Enum object containing the available reports.

Return type:

Enum

latest()

Displays the timestamp of the latest local and live reports generated. Reads the report files from yq.PARENT_DIR and queries the GitHub API for the latest commit timestamp for each file. Returns the result as an message in the channel.

Returns:

A dictionary containing information about the latest reports.

Return type:

dict

Displays the links to the YugiQuery webpage, repository, and data.

Returns:

A dictionary containing links to YugiQuery resources.

Return type:

dict

load_repo_vars()

Load repository variables including URLs and paths.

async run_query(callback: Callable, channel_id: int, report: Reports = Reports.All, progress_bar: tqdm_asyncio | None = None)

Runs a YugiQuery flow by launching a separate thread and monitoring its progress.

Parameters:

  • callback (Callable) – A callback function which receives a string argument.

  • channel_id (int) – The channel ID to display the progress bar.

  • report (Reports, optional) – The report to run. Defaults to All.

  • progress_bar (tqdm, optional) – A tqdm progress bar. Defaults to None.

Returns:

A dictionary containing the result of the query execution.

Return type:

dict

uptime()

Returns humanized bot uptime.

class Discord(token: str, channel: str | int)

Bases: Bot, Bot

Discord bot subclass. Inherits from Bot class and discord.ext.commands.Bot.

Parameters:

  • token (str) – The token for the Discord bot.

  • channel (Union[str, int]) – The channel for the bot.

discord.ext.commands.Bot attributes
Bot attributes
async on_command_error(ctx, error: CommandError)

Event callback that runs whenever a command invoked by the user results in an error. Sends a message to the channel indicating the type of error that occurred.

Parameters:

  • ctx (commands.Context) – The context of the error.

  • error (commands.CommandError) – The error received.

async on_message(message: Message)

Event callback that runs whenever a message is sent in a server where the bot is present. Responds with a greeting to any message starting with ‘hi’.

Parameters:

  • ctx (commands.Context) – The context of the message.

  • message (discord.Message) – The message received.

async on_ready()

Event callback that runs when the bot is ready to start receiving events and commands. Prints out the bot’s username and the guilds it’s connected to.

register_commands()

Register command handlers for the Discord bot.

Command descriptions:

abort - Aborts a running YugiQuery flow by terminating the process.

battle - Simulate a battle of all monster cards.

benchmark - Show average time each report takes to complete.

data - Send latest data files.

shutdown - Shutdown bot

latest - Show latest time each report was generated.

links - Show YugiQuery links.

ping - Test the bot connection latency.

run - Run full YugiQuery flow.

status - Display bot status and system information.

run()

Starts running the discord Bot.

class ProgressHandler(queue: Queue, progress_bar: tqdm_asyncio | None = None, pbar_kwargs: Dict[str, Any] = {})

Bases: object

Progress handler class.

Parameters:

  • queue (multiprocessing.Queue) – The multiprocessing queue to communicate progress status.

  • progress_bar (tqdm, optional) – The tqdm progress bar implementation. Defaults to None.

  • pbar_kwargs (Dict[str, Any], optional) – Keyword arguments to customize the progress bar. Defaults to None.

queue

The multiprocessing queue to communicate progress status.

Type:

multiprocessing.Queue

progress_bar

The tqdm progress bar implementation.

Type:

tqdm, optional

pbar_kwargs

Keyword arguments to customize the progress bar.

Type:

Dict[str, Any], optional

exit(API_status: bool = True)

Puts the API status in the queue.

Parameters:

API_status (bool, optional) – The status to put in the queue. Defaults to True.

pbar(iterable, **kwargs)

Initializes and returns a progress bar instance if progress_bar is not None.

Parameters:

  • iterable (iterable) – The iterable to track progress.

  • **kwargs – Additional keyword arguments for the progress bar.

Returns:

The initialized progress bar instance or None if progress_bar is None.

Return type:

Progress bar instance or None

class Telegram(token: str, channel: str | int)

Bases: Bot

Telegram bot subclass. Inherits from Bot class.

Parameters:

  • token (str) – The token for the Telegram bot.

  • channel (Union[str, int]) – The Telegram channel ID.

application

The Telegram Application bot instance.

Type:

telegram.Application

Bot attributes
register_commands()

Register command handlers for the Telegram bot.

Command descriptions to pass to BotFather:

abort - Aborts a running YugiQuery flow by terminating the process.

battle - Simulate a battle of all monster cards.

benchmark - Show average time each report takes to complete.

data - Send latest data files.

shutdown - Shutdown bot.

latest - Show latest time each report was generated.

links - Show YugiQuery links.

ping - Test the bot connection latency.

run - Run full YugiQuery flow.

status - Display bot status and system information.

register_events()

Register event handlers for the Telegram bot.

run()

Start running the Telegram bot.

escape_chars(string: str, chars: List[str] = ['_', '.', '-', '+', '#', '@', '='])

Escapes specified characters in a given string by adding a backslash before each occurrence.

Parameters:

  • string (str) – The input string to be processed.

  • chars (list, optional) – A list of characters to be escaped. Default is [“_”, “.”, “-”, “+”, “#”, “@”, “=”].

Returns:

The input string with the specified characters escaped.

Return type:

str

get_humanize_granularity(seconds: int)

Humanizes a time interval given in seconds.

Parameters:

seconds (int) – The time interval in seconds.

Returns:

A list of human-readable granularities for the time interval.

Return type:

list

load_secrets_with_args(args: Dict[str, Any])

Load secrets from command-line arguments, and update them with values from environment variables or a .env file if necessary.

Parameters:

dict[str – Any]: A dictionary of secrets.

Returns:

A dictionary containing the loaded secrets.

Return type:

Dict[str, str]

Raises:

KeyError – If a required secret is not found in the loaded secrets.