codekingpro/portable-devtools

1"""2Sends updates to a Telegram bot.3 4Usage:5>>> from tqdm.contrib.telegram import tqdm, trange6>>> for i in trange(10, token='{token}', chat_id='{chat_id}'):7...     ...8 9![screenshot](https://tqdm.github.io/img/screenshot-telegram.gif)10"""11from os import getenv12from warnings import warn13 14from requests import Session15 16from ..auto import tqdm as tqdm_auto17from ..std import TqdmWarning18from .utils_worker import MonoWorker19 20__author__ = {"github.com/": ["casperdcl"]}21__all__ = ['TelegramIO', 'tqdm_telegram', 'ttgrange', 'tqdm', 'trange']22 23 24class TelegramIO(MonoWorker):25    """Non-blocking file-like IO using a Telegram Bot."""26    API = 'https://api.telegram.org/bot'27 28    def __init__(self, token, chat_id):29        """Creates a new message in the given `chat_id`."""30        super().__init__()31        self.token = token32        self.chat_id = chat_id33        self.session = Session()34        self.text = self.__class__.__name__35        self.message_id36 37    @property38    def message_id(self):39        if hasattr(self, '_message_id'):40            return self._message_id41        try:42            res = self.session.post(43                self.API + '%s/sendMessage' % self.token,44                data={'text': '`' + self.text + '`', 'chat_id': self.chat_id,45                      'parse_mode': 'MarkdownV2'}).json()46        except Exception as e:47            tqdm_auto.write(str(e))48        else:49            if res.get('error_code') == 429:50                warn("Creation rate limit: try increasing `mininterval`.",51                     TqdmWarning, stacklevel=2)52            else:53                self._message_id = res['result']['message_id']54                return self._message_id55 56    def write(self, s):57        """Replaces internal `message_id`'s text with `s`."""58        if not s:59            s = "..."60        s = s.replace('\r', '').strip()61        if s == self.text:62            return  # avoid duplicate message Bot error63        message_id = self.message_id64        if message_id is None:65            return66        self.text = s67        try:68            future = self.submit(69                self.session.post, self.API + '%s/editMessageText' % self.token,70                data={'text': '`' + s + '`', 'chat_id': self.chat_id,71                      'message_id': message_id, 'parse_mode': 'MarkdownV2'})72        except Exception as e:73            tqdm_auto.write(str(e))74        else:75            return future76 77    def delete(self):78        """Deletes internal `message_id`."""79        try:80            future = self.submit(81                self.session.post, self.API + '%s/deleteMessage' % self.token,82                data={'chat_id': self.chat_id, 'message_id': self.message_id})83        except Exception as e:84            tqdm_auto.write(str(e))85        else:86            return future87 88 89class tqdm_telegram(tqdm_auto):90    """91    Standard `tqdm.auto.tqdm` but also sends updates to a Telegram Bot.92    May take a few seconds to create (`__init__`).93 94    - create a bot <https://core.telegram.org/bots#6-botfather>95    - copy its `{token}`96    - add the bot to a chat and send it a message such as `/start`97    - go to <https://api.telegram.org/bot`{token}`/getUpdates> to find out98      the `{chat_id}`99    - paste the `{token}` & `{chat_id}` below100 101    >>> from tqdm.contrib.telegram import tqdm, trange102    >>> for i in tqdm(iterable, token='{token}', chat_id='{chat_id}'):103    ...     ...104    """105    def __init__(self, *args, **kwargs):106        """107        Parameters108        ----------109        token  : str, required. Telegram token110            [default: ${TQDM_TELEGRAM_TOKEN}].111        chat_id  : str, required. Telegram chat ID112            [default: ${TQDM_TELEGRAM_CHAT_ID}].113 114        See `tqdm.auto.tqdm.__init__` for other parameters.115        """116        if not kwargs.get('disable'):117            kwargs = kwargs.copy()118            self.tgio = TelegramIO(119                kwargs.pop('token', getenv('TQDM_TELEGRAM_TOKEN')),120                kwargs.pop('chat_id', getenv('TQDM_TELEGRAM_CHAT_ID')))121        super().__init__(*args, **kwargs)122 123    def display(self, **kwargs):124        super().display(**kwargs)125        fmt = self.format_dict126        if fmt.get('bar_format', None):127            fmt['bar_format'] = fmt['bar_format'].replace(128                '<bar/>', '{bar:10u}').replace('{bar}', '{bar:10u}')129        else:130            fmt['bar_format'] = '{l_bar}{bar:10u}{r_bar}'131        self.tgio.write(self.format_meter(**fmt))132 133    def clear(self, *args, **kwargs):134        super().clear(*args, **kwargs)135        if not self.disable:136            self.tgio.write("")137 138    def close(self):139        if self.disable:140            return141        super().close()142        if not (self.leave or (self.leave is None and self.pos == 0)):143            self.tgio.delete()144 145 146def ttgrange(*args, **kwargs):147    """Shortcut for `tqdm.contrib.telegram.tqdm(range(*args), **kwargs)`."""148    return tqdm_telegram(range(*args), **kwargs)149 150 151# Aliases152tqdm = tqdm_telegram153trange = ttgrange154