Bot Subpackage

This subpackage contains the bot related modules used to run a YugiQuery bot.

Base Module

class yugiquery.bot.base.Base

Bases: object

Bot superclass.

Parameters:

**kwargs – Additional keyword arguments.

start_time

The bot initialization timestamp.

Type:

arrow.Arrow

repo

The git repository object, if there is a repository, else None.

Type:

git.Repo

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

property DataFlows: dict[str, str]

Dynamically returns the current available data flows as a dictionary.

property Reports: dict[str, Path | str]

Dynamically returns the current available reports as a dictionary.

property URLS: SimpleNamespace

Property to get the URLs of the remote repository and webpage. Cached after first access.

Returns:

Object with api, repo, and webpage attributes (empty strings if no repo).

Return type:

SimpleNamespace

__init__(**kwargs)

Initializes the Bot base class.

Parameters:

**kwargs – Additional keyword arguments.

abort() → str

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[[str], Awaitable[None]], atk_weight: int = 4, def_weight: int = 1) → dict

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[[str], Awaitable[None]]) – 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() → Dict[str, str]

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() → Dict[str, str]

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

git_cmd(command: GitCommands, passphrase: str = '') → str

Executes a git command on the repository.

Parameters:

  • command (Git_command) – The git command to execute.

  • passphrase (str, optional) – The passphrase to use for encryption/decryption. Defaults to empty.

Returns:

The result message of the command.

Return type:

str

latest() → Dict[str, str]

Displays the timestamp of the latest local and live reports generated. Reads the report files from dirs.REPORTS and queries the GitHub API for the latest commit timestamp for each file.

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

async query_fetch(callback: Callable[[str], Awaitable[None]], data: str | list[str] = [], progress_bar: Type[tqdm_asyncio] | None = None, **pbar_kwargs) → Dict[str, str]

Shortcut function to run the fetch operation on YugiQuery.

Parameters:

  • callback (Callable[[str], None]) – A callback function which receives a string argument.

  • data (str | list[str], optional) – The data to fetch. Defaults to DataFlows.All.

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

Returns:

A dictionary containing the result of the data fetching.

Return type:

dict

async query_report(callback: Callable[[str], Awaitable[None]], report: str | list[str] | list[Path] = [], progress_bar: Type[tqdm_asyncio] | None = None, **pbar_kwargs) → Dict[str, str]

Shortcut function to run the report operation on YugiQuery.

Parameters:

  • callback (Callable[[str], None]) – A callback function which receives a string argument.

  • report (Reports | None, optional) – The report to generate. Defaults to Reports.All.

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

Returns:

A dictionary containing the result of the report fetching.

Return type:

dict

async query_run(callback: Callable[[str], Awaitable[None]], data: str | list[str] = [], report: str | list[str] | list[Path] = [], progress_bar: Type[tqdm_asyncio] | None = None, **pbar_kwargs) → Dict[str, str]

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

Parameters:

  • callback (Callable[[str], None]) – A callback function which receives a string argument.

  • data (str | list[str], optional) – The data flow to run. Defaults to DataFlows.All.

  • report (str | list[str] | list[Path], optional) – The report to run. Defaults to Reports.All.

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

  • **pbar_kwargs – Additional Keyword arguments to customize the progress bar..

Returns:

A dictionary containing the result of the query execution.

Return type:

dict

uptime()

Returns humanized bot uptime.

class yugiquery.bot.base.GitCommands

Bases: StrEnum

Enum class to represent the available git commands.

__new__(value)

Discord Module

class yugiquery.bot.discord.Discord

Bases: Base, Bot

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

Parameters:

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

  • channel_id (str | int) – The channel ID for the bot.

discord.ext.commands.Bot attributes
Bot attributes
property ReportsOrData: dict[str, dict[str, str | Path | None]]

Combines the current Reports and DataFlows dictionaries into a single mapping.

__init__(token: str, channel_id: str | int)

Initializes the Discord bot subclass.

Parameters:

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

  • channel_id (str | int) – The channel ID for the Discord bot.

async on_command_error(ctx, error: CommandError) → None

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) → None

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() → None

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() → None

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. fetch - Run the YugiQuery data fetch operation. git - Run a Git command. latest - Show latest time each report was generated. links - Show YugiQuery links. ping - Test the bot connection latency. report - Run the YugiQuery report generation operation. run - Run full YugiQuery flow. info - Display bot and system information. shutdown - Shutdown bot.

run() → None

Starts running the discord Bot.

async send_long_message(ctx, content: str, filename: str = 'message.txt', **kwargs)

Sends a message, ensuring it does not exceed Discord’s message limit. If the message is too long, sends it as a text file attachment.

Parameters:

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

  • content (str) – The message content.

  • filename (str) – The name of the file if the message is sent as an attachment.

  • **kwargs – Additional keyword arguments to pass to the send method.

Telegram Module

class yugiquery.bot.telegram.Telegram

Bases: Base

Telegram bot subclass. Inherits from Bot class.

Parameters:

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

  • chat_id (str | int) – The Telegram chat ID.

application

The Telegram Application bot instance.

Type:

telegram.Application

Bot attributes
__init__(token: str, chat_id: str | int)

Initialize the Telegram Bot subclass.

Parameters:

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

  • chat_id (str | int) – The chat ID for the Telegram bot.

command_handler(command, **kwargs)

Decorator to register a command handler for the Telegram bot.

Parameters:

  • command (str) – The command to register.

  • **kwargs – Additional keyword arguments to pass to the CommandHandler.

Returns:

The decorated function.

Return type:

function

async post_init_handler(application: Application) → None

Post-initialization handler for the Telegram bot.

Parameters:

application (telegram.ext.Application) – The Telegram Application bot instance.

register_commands() → None

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. fetch - Run the YugiQuery data fetch operation. git - Run a Git command. latest - Show latest time each report was generated. links - Show YugiQuery links. ping - Test the bot connection latency. run - Run full YugiQuery flow. report - Run the YugiQuery report generation operation. info - Display bot and system information. shutdown - Shutdown bot.

register_events() → None

Register event handlers for the Telegram bot.

run() → None

Start running the Telegram bot.