AndroidRemoteController/python/py/Lib/site-packages/tqdm/contrib/discord.py

"""
Sends updates to a Discord bot.

Usage:
>>> from tqdm.contrib.discord import tqdm, trange
>>> for i in trange(10, token='{token}', channel_id='{channel_id}'):
...     ...

![screenshot](https://tqdm.github.io/img/screenshot-discord.png)
"""
from os import getenv
from warnings import warn

from requests import Session
from requests.utils import default_user_agent

from ..auto import tqdm as tqdm_auto
from ..std import TqdmWarning
from ..version import __version__
from .utils_worker import MonoWorker

__author__ = {"github.com/": ["casperdcl", "guigoruiz1"]}
__all__ = ['DiscordIO', 'tqdm_discord', 'tdrange', 'tqdm', 'trange']


class DiscordIO(MonoWorker):
    """Non-blocking file-like IO using a Discord Bot."""
    API = "https://discord.com/api/v10"
    UA = f"tqdm (https://tqdm.github.io, {__version__}) {default_user_agent()}"

    def __init__(self, token, channel_id):
        """Creates a new message in the given `channel_id`."""
        super().__init__()
        self.token = token
        self.channel_id = channel_id
        self.session = Session()
        self.text = self.__class__.__name__
        self.message_id

    @property
    def message_id(self):
        if hasattr(self, '_message_id'):
            return self._message_id
        try:
            res = self.session.post(
                f'{self.API}/channels/{self.channel_id}/messages',
                headers={'Authorization': f'Bot {self.token}', 'User-Agent': self.UA},
                json={'content': f"`{self.text}`"}).json()
        except Exception as e:
            tqdm_auto.write(str(e))
        else:
            if res.get('error_code') == 429:
                warn("Creation rate limit: try increasing `mininterval`.",
                     TqdmWarning, stacklevel=2)
            else:
                self._message_id = res['id']
                return self._message_id

    def write(self, s):
        """Replaces internal `message_id`'s text with `s`."""
        if not s:
            s = "..."
        s = s.replace('\r', '').strip()
        if s == self.text:
            return  # avoid duplicate message Bot error
        message_id = self.message_id
        if message_id is None:
            return
        self.text = s
        try:
            future = self.submit(
                self.session.patch,
                f'{self.API}/channels/{self.channel_id}/messages/{message_id}',
                headers={'Authorization': f'Bot {self.token}', 'User-Agent': self.UA},
                json={'content': f"`{self.text}`"})
        except Exception as e:
            tqdm_auto.write(str(e))
        else:
            return future

    def delete(self):
        """Deletes internal `message_id`."""
        try:
            future = self.submit(
                self.session.delete,
                f'{self.API}/channels/{self.channel_id}/messages/{self.message_id}',
                headers={'Authorization': f'Bot {self.token}', 'User-Agent': self.UA})
        except Exception as e:
            tqdm_auto.write(str(e))
        else:
            return future


class tqdm_discord(tqdm_auto):
    """
    Standard `tqdm.auto.tqdm` but also sends updates to a Discord Bot.
    May take a few seconds to create (`__init__`).

    - create a discord bot (not public, no requirement of OAuth2 code
      grant, only send message permissions) & invite it to a channel:
      <https://discordpy.readthedocs.io/en/latest/discord.html>
    - copy the bot `{token}` & `{channel_id}` and paste below

    >>> from tqdm.contrib.discord import tqdm, trange
    >>> for i in tqdm(iterable, token='{token}', channel_id='{channel_id}'):
    ...     ...
    """
    def __init__(self, *args, **kwargs):
        """
        Parameters
        ----------
        token  : str, required. Discord bot token
            [default: ${TQDM_DISCORD_TOKEN}].
        channel_id  : int, required. Discord channel ID
            [default: ${TQDM_DISCORD_CHANNEL_ID}].

        See `tqdm.auto.tqdm.__init__` for other parameters.
        """
        if not kwargs.get('disable'):
            kwargs = kwargs.copy()
            self.dio = DiscordIO(
                kwargs.pop('token', getenv('TQDM_DISCORD_TOKEN')),
                kwargs.pop('channel_id', getenv('TQDM_DISCORD_CHANNEL_ID')))
        super().__init__(*args, **kwargs)

    def display(self, **kwargs):
        super().display(**kwargs)
        fmt = self.format_dict
        if fmt.get('bar_format', None):
            fmt['bar_format'] = fmt['bar_format'].replace(
                '<bar/>', '{bar:10u}').replace('{bar}', '{bar:10u}')
        else:
            fmt['bar_format'] = '{l_bar}{bar:10u}{r_bar}'
        self.dio.write(self.format_meter(**fmt))

    def clear(self, *args, **kwargs):
        super().clear(*args, **kwargs)
        if not self.disable:
            self.dio.write("")

    def close(self):
        if self.disable:
            return
        super().close()
        if not (self.leave or (self.leave is None and self.pos == 0)):
            self.dio.delete()


def tdrange(*args, **kwargs):
    """Shortcut for `tqdm.contrib.discord.tqdm(range(*args), **kwargs)`."""
    return tqdm_discord(range(*args), **kwargs)


# Aliases
tqdm = tqdm_discord
trange = tdrange