Utilities

All the utilities used to manage plugins, client and events can be found here.

Client

class userbot.utils.client.Command(func: <built-in function callable>, handlers: list, info: str, usage: str, builtin: bool)

Bases: object

class userbot.utils.client.UserBotClient(session: typing.Union[str, Session], api_id: int, api_hash: str, *, connection: typing.Type[Connection] = <class 'telethon.network.connection.tcpfull.ConnectionTcpFull'>, use_ipv6: bool = False, proxy: Union[tuple, dict] = None, local_addr=None, timeout: int = 10, request_retries: int = 5, connection_retries: int = 5, retry_delay: int = 1, auto_reconnect: bool = True, sequential_updates: bool = False, flood_sleep_threshold: int = 60, raise_last_call_error: bool = False, device_model: str = None, system_version: str = None, app_version: str = None, lang_code: str = 'en', system_lang_code: str = 'en', loop: asyncio.events.AbstractEventLoop = None, base_logger: Union[str, logging.Logger] = None)

Bases: telethon.client.telegramclient.TelegramClient

UserBot client with additional attributes inheriting TelegramClient

answer(entity, message: str = '', *, reply_to: Union[int, telethon.tl.patched.Message] = None, parse_mode: Optional[str] = 'markdown', link_preview: bool = False, file: Union[str, bytes, BinaryIO, telethon.tl.types.MessageMediaEmpty, telethon.tl.types.MessageMediaPhoto, telethon.tl.types.MessageMediaGeo, telethon.tl.types.MessageMediaContact, telethon.tl.types.MessageMediaUnsupported, telethon.tl.types.MessageMediaDocument, telethon.tl.types.MessageMediaWebPage, telethon.tl.types.MessageMediaVenue, telethon.tl.types.MessageMediaGame, telethon.tl.types.MessageMediaInvoice, telethon.tl.types.MessageMediaGeoLive, telethon.tl.types.MessageMediaPoll, telethon.tl.types.MessageMediaDice, telethon.tl.types.InputFile, telethon.tl.types.InputFileBig, telethon.tl.types.InputFileLocation, telethon.tl.types.InputEncryptedFileLocation, telethon.tl.types.InputDocumentFileLocation, telethon.tl.types.InputSecureFileLocation, telethon.tl.types.InputTakeoutFileLocation, telethon.tl.types.InputPhotoFileLocation, telethon.tl.types.InputPhotoLegacyFileLocation, telethon.tl.types.InputPeerPhotoFileLocation, telethon.tl.types.InputStickerSetThumb, Sequence[Union[str, bytes, BinaryIO, telethon.tl.types.MessageMediaEmpty, telethon.tl.types.MessageMediaPhoto, telethon.tl.types.MessageMediaGeo, telethon.tl.types.MessageMediaContact, telethon.tl.types.MessageMediaUnsupported, telethon.tl.types.MessageMediaDocument, telethon.tl.types.MessageMediaWebPage, telethon.tl.types.MessageMediaVenue, telethon.tl.types.MessageMediaGame, telethon.tl.types.MessageMediaInvoice, telethon.tl.types.MessageMediaGeoLive, telethon.tl.types.MessageMediaPoll, telethon.tl.types.MessageMediaDice, telethon.tl.types.InputFile, telethon.tl.types.InputFileBig, telethon.tl.types.InputFileLocation, telethon.tl.types.InputEncryptedFileLocation, telethon.tl.types.InputDocumentFileLocation, telethon.tl.types.InputSecureFileLocation, telethon.tl.types.InputTakeoutFileLocation, telethon.tl.types.InputPhotoFileLocation, telethon.tl.types.InputPhotoLegacyFileLocation, telethon.tl.types.InputPeerPhotoFileLocation, telethon.tl.types.InputStickerSetThumb]]] = None, force_document: bool = False, clear_draft: bool = False, buttons: Union[telethon.tl.types.ReplyKeyboardHide, telethon.tl.types.ReplyKeyboardForceReply, telethon.tl.types.ReplyKeyboardMarkup, telethon.tl.types.ReplyInlineMarkup, telethon.tl.types.KeyboardButton, telethon.tl.types.KeyboardButtonUrl, telethon.tl.types.KeyboardButtonCallback, telethon.tl.types.KeyboardButtonRequestPhone, telethon.tl.types.KeyboardButtonRequestGeoLocation, telethon.tl.types.KeyboardButtonSwitchInline, telethon.tl.types.KeyboardButtonGame, telethon.tl.types.KeyboardButtonBuy, telethon.tl.types.KeyboardButtonUrlAuth, telethon.tl.types.InputKeyboardButtonUrlAuth, telethon.tl.types.KeyboardButtonRequestPoll, telethon.tl.custom.button.Button, Sequence[Union[telethon.tl.types.KeyboardButton, telethon.tl.types.KeyboardButtonUrl, telethon.tl.types.KeyboardButtonCallback, telethon.tl.types.KeyboardButtonRequestPhone, telethon.tl.types.KeyboardButtonRequestGeoLocation, telethon.tl.types.KeyboardButtonSwitchInline, telethon.tl.types.KeyboardButtonGame, telethon.tl.types.KeyboardButtonBuy, telethon.tl.types.KeyboardButtonUrlAuth, telethon.tl.types.InputKeyboardButtonUrlAuth, telethon.tl.types.KeyboardButtonRequestPoll, telethon.tl.custom.button.Button]], Sequence[Sequence[Union[telethon.tl.types.KeyboardButton, telethon.tl.types.KeyboardButtonUrl, telethon.tl.types.KeyboardButtonCallback, telethon.tl.types.KeyboardButtonRequestPhone, telethon.tl.types.KeyboardButtonRequestGeoLocation, telethon.tl.types.KeyboardButtonSwitchInline, telethon.tl.types.KeyboardButtonGame, telethon.tl.types.KeyboardButtonBuy, telethon.tl.types.KeyboardButtonUrlAuth, telethon.tl.types.InputKeyboardButtonUrlAuth, telethon.tl.types.KeyboardButtonRequestPoll, telethon.tl.custom.button.Button]]]] = None, silent: bool = True, schedule: Union[float, datetime.datetime, datetime.date, datetime.timedelta, None] = None, log: str = None, reply: bool = False, self_destruct: int = None, event: telethon.tl.custom.message.Message = None) → Union[None, telethon.tl.custom.message.Message, Sequence[telethon.tl.custom.message.Message]]

Custom bound method for the Message object

commandcategories = {}
commands = {}
config = None
database = True
disabled_commands = {}
failed_imports = []
fast_download_file(location: Union[telethon.tl.types.Document, telethon.tl.types.InputDocumentFileLocation, telethon.tl.types.InputPeerPhotoFileLocation, telethon.tl.types.InputFileLocation, telethon.tl.types.InputPhotoFileLocation], out: BinaryIO, progress_callback: callable = None) → BinaryIO
fast_upload_file(file: BinaryIO, progress_callback: callable = None) → Union[telethon.tl.types.InputFile, telethon.tl.types.InputFileBig]
get_traceback(exc: Exception) → str
logger = False
onMessage(builtin: bool = False, command: str = None, edited: bool = True, info: str = None, doc_args: dict = {}, **kwargs) → callable

Method to register a function without the client

static parse_arguments(arguments: str) → Tuple[List[Union[int, str, float, list]], Dict[str, Union[int, str, float, list, range, List[Union[int, str, float, list]]]]]
pluginManager = None
plugins = []
prefix = None
reconnect = True
register_commands = False
resanswer(entity, message: str, plugin: str = None, name: str = None, formats: dict = {}, **kwargs) → telethon.tl.custom.message.Message
running_processes = {}
version = 0
userbot.utils.client.parse_arguments

staticmethod(function) -> method

Convert a function to be a static method.

A static method does not receive an implicit first argument. To declare a static method, use this idiom:

class C:

@staticmethod def f(arg1, arg2, …):

…

It can be called either on the class (e.g. C.f()) or on an instance (e.g. C().f()). The instance is ignored except for its class.

Static methods in Python are similar to those found in Java or C++. For a more advanced concept, see the classmethod builtin.

Events

class userbot.utils.events.MessageEdited(disable_prefix: bool = None, regex: Tuple[str, int] = None, require_admin: bool = None, inline: bool = False, **kwargs)

Bases: userbot.utils.events.NewMessage

Custom MessageEdited event inheriting the custom NewMessage event

class Event(message)

Bases: telethon.events.newmessage.Event

Overriding the default Event which inherits Telethon’s NewMessage

classmethod build(update, others=None, self_id=None)

Required to check if message is edited, double events. Note: Don’t handle UpdateEditChannelMessage from channels since the

update doesn’t show which user edited the message
class userbot.utils.events.NewMessage(disable_prefix: bool = None, regex: Tuple[str, int] = None, require_admin: bool = None, inline: bool = False, **kwargs)

Bases: telethon.events.newmessage.NewMessage

Custom NewMessage event inheriting the default Telethon event

filter(event)

Overriding the default filter to check additional values

userbot.utils.events.answer(self, *args, **kwargs)
userbot.utils.events.resanswer(self, *args, **kwargs)

Helpers

class userbot.utils.helpers.ProgressCallback(event, start=None, filen='unamed', update=5)

Bases: object

Custom class to handle upload and download progress.

dl_progress(current, total)

Handle the download progress only.

resolve_prog(current, total)

Calculate the necessary info and make a dict from it.

up_progress(current, total)

Handle the upload progress only.

userbot.utils.helpers.calc_eta(elp: float, speed: int, current: int, total: int) → int
userbot.utils.helpers.disable_commands(client: userbot.utils.client.UserBotClient, commands: str) → None
userbot.utils.helpers.dl_prog(d: dict, cb: userbot.utils.helpers.ProgressCallback) → Tuple[Union[str, bool], bool]

Logs the download progress

userbot.utils.helpers.format_speed(speed_per_second, unit)
userbot.utils.helpers.isRestart(client: userbot.utils.client.UserBotClient) → None

Check if the script restarted itself and edit the last message

userbot.utils.helpers.is_ffmpeg_there()
userbot.utils.helpers.printUser(entity: telethon.tl.types.User) → None

Print the user’s first name + last name upon start

userbot.utils.helpers.printVersion(version: int, prefix: str) → None

Print the version of the bot with the default prefix

userbot.utils.helpers.restart(event: telethon.events.newmessage.NewMessage.Event) → None
userbot.utils.helpers.restarter(client: userbot.utils.client.UserBotClient) → None
userbot.utils.helpers.ul_prog(d: dict, cb: userbot.utils.helpers.ProgressCallback) → Tuple[Union[str, bool], bool]

Logs the upload progress

Plugin Manager

class userbot.utils.pluginManager.Callback(name: str, callback: <built-in function callable>)

Bases: object

class userbot.utils.pluginManager.Plugin(name: str, callbacks: List[userbot.utils.pluginManager.Callback], path: str, module: module)

Bases: object

class userbot.utils.pluginManager.PluginManager(client: telethon.client.telegramclient.TelegramClient)

Bases: object

active_plugins = []
add_handlers() → None

Apply event handlers to all the found callbacks.

import_all() → None

Import all the (enabled) plugins and skip the rest.

inactive_plugins = []
remove_handlers() → None

Remove event handlers to all the found callbacks.

class userbot.utils.pluginManager.SourcelessPluginLoader(name, data, path: str = '<string>')

Bases: importlib.abc.SourceLoader

Loader for (byte) strings which don’t have a source.

get_code(path)

Return the code object if it exists.

get_data(path)

The data isn’t stored locally, return the (bytes) string.

get_filename(fullname)

Return the origin (GitHub’s raw URL).

userbot.utils.pluginManager.get_pip_packages(requirements: str = None) → list

Get a list of all the pacakage’s names.

userbot.utils.pluginManager.install_pip_packages(packages: List[str]) → bool

Install pip packages.

userbot.utils.pluginManager.restart_script() → None

Restart the current script.

userbot.utils.pluginManager.run_async(func: callable)

Run async functions with the right event loop.