Skip to content

module/telegram_bot.py

module.telegram_bot

TelegramBot

Telegram

Bases: Session

Telegram

Parameters:

Name Type Description Default
token str

API token.

required
Source code in module/telegram_bot.py
class Telegram(Session):
    ''' Telegram

    Args:
        token (str): API token.

    '''

    def __init__(self, token: str) -> None:
        super().__init__()
        self.url = f'https://api.telegram.org/bot{token}'

    def get_me(self) -> Response:
        ''' Get me '''
        return self.post(f'{self.url}/getMe')

    def send_message(self, chat_id: str, text: str,
                     parse_mode: str = 'Markdown',
                     reply_markup: Union[dict[str, Any], None] = None) -> Response:
        ''' Send message

        Args:
            chat_id (str): Chat id.
            text (str): Text.
            parse_mode (str): `Markdown`, `MarkdownV2`, `HTML`.
            reply_markup (dict): Reply markup.

        References:
            https://core.telegram.org/bots/api#sendmessage

        '''
        data = {
            'chat_id': chat_id,
            'text': text,
            'parse_mode': parse_mode}  # type: dict[str, Union[str, dict[str, Any]]]

        if reply_markup is not None:
            data['reply_markup'] = reply_markup

        return self.post(f'{self.url}/sendMessage', json=data)

    def set_webhook(self, url: str) -> Response:
        ''' Set webhook

        Args:
            url (str): URL.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.url}/setWebhook', json={'url': url})

    def get_webhook_info(self, url: str) -> Response:
        ''' Get webhook info

        Args:
            url (str): URL.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.url}/getWebhookInfo', json={'url': url})

    def delete_webhook(self) -> Response:
        ''' delete webhook

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.url}/deleteWebhook')

    @staticmethod
    def is_command_start(data: dict[str, Any]) -> bool:
        ''' command start

        Args:
            data (dict): The data from telegram return.

        Returns:
            Check the `message.from.is_bot` ot `message.text`.

        '''
        if data['message']['from']['is_bot']:
            return False

        if data['message']['text'].strip() == '/start':
            return True

        return False

    @staticmethod
    def is_command_start_linkme(data: dict[str, Any]) -> bool:
        ''' command start

        Args:
            data (dict): The data from telegram return.

        Returns:
            Check the `message.from.is_bot` ot `message.text`.

        '''
        if 'message' not in data:
            return False

        if data['message']['from']['is_bot']:
            return False

        if data['message']['text'].strip() in ('/start linkme', '/linkme'):
            return True

        return False

delete_webhook

delete_webhook() -> Response

delete webhook

Returns:

Type Description
Response

Return the requests.Response object.

Source code in module/telegram_bot.py
def delete_webhook(self) -> Response:
    ''' delete webhook

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.url}/deleteWebhook')

get_me

get_me() -> Response

Get me

Source code in module/telegram_bot.py
def get_me(self) -> Response:
    ''' Get me '''
    return self.post(f'{self.url}/getMe')

get_webhook_info

get_webhook_info(url: str) -> Response

Get webhook info

Parameters:

Name Type Description Default
url str

URL.

required

Returns:

Type Description
Response

Return the requests.Response object.

Source code in module/telegram_bot.py
def get_webhook_info(self, url: str) -> Response:
    ''' Get webhook info

    Args:
        url (str): URL.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.url}/getWebhookInfo', json={'url': url})

is_command_start staticmethod

is_command_start(data: dict[str, Any]) -> bool

command start

Parameters:

Name Type Description Default
data dict

The data from telegram return.

required

Returns:

Type Description
bool

Check the message.from.is_bot ot message.text.

Source code in module/telegram_bot.py
@staticmethod
def is_command_start(data: dict[str, Any]) -> bool:
    ''' command start

    Args:
        data (dict): The data from telegram return.

    Returns:
        Check the `message.from.is_bot` ot `message.text`.

    '''
    if data['message']['from']['is_bot']:
        return False

    if data['message']['text'].strip() == '/start':
        return True

    return False

is_command_start_linkme staticmethod

is_command_start_linkme(data: dict[str, Any]) -> bool

command start

Parameters:

Name Type Description Default
data dict

The data from telegram return.

required

Returns:

Type Description
bool

Check the message.from.is_bot ot message.text.

Source code in module/telegram_bot.py
@staticmethod
def is_command_start_linkme(data: dict[str, Any]) -> bool:
    ''' command start

    Args:
        data (dict): The data from telegram return.

    Returns:
        Check the `message.from.is_bot` ot `message.text`.

    '''
    if 'message' not in data:
        return False

    if data['message']['from']['is_bot']:
        return False

    if data['message']['text'].strip() in ('/start linkme', '/linkme'):
        return True

    return False

send_message

send_message(
    chat_id: str,
    text: str,
    parse_mode: str = "Markdown",
    reply_markup: Union[dict[str, Any], None] = None,
) -> Response

Send message

Parameters:

Name Type Description Default
chat_id str

Chat id.

required
text str

Text.

required
parse_mode str

Markdown, MarkdownV2, HTML.

'Markdown'
reply_markup dict

Reply markup.

None
References

https://core.telegram.org/bots/api#sendmessage

Source code in module/telegram_bot.py
def send_message(self, chat_id: str, text: str,
                 parse_mode: str = 'Markdown',
                 reply_markup: Union[dict[str, Any], None] = None) -> Response:
    ''' Send message

    Args:
        chat_id (str): Chat id.
        text (str): Text.
        parse_mode (str): `Markdown`, `MarkdownV2`, `HTML`.
        reply_markup (dict): Reply markup.

    References:
        https://core.telegram.org/bots/api#sendmessage

    '''
    data = {
        'chat_id': chat_id,
        'text': text,
        'parse_mode': parse_mode}  # type: dict[str, Union[str, dict[str, Any]]]

    if reply_markup is not None:
        data['reply_markup'] = reply_markup

    return self.post(f'{self.url}/sendMessage', json=data)

set_webhook

set_webhook(url: str) -> Response

Set webhook

Parameters:

Name Type Description Default
url str

URL.

required

Returns:

Type Description
Response

Return the requests.Response object.

Source code in module/telegram_bot.py
def set_webhook(self, url: str) -> Response:
    ''' Set webhook

    Args:
        url (str): URL.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.url}/setWebhook', json={'url': url})

TelegramBot

Bases: Telegram

TelegramBot

Parameters:

Name Type Description Default
token str

API token.

required
Source code in module/telegram_bot.py
class TelegramBot(Telegram):
    ''' TelegramBot

    Args:
        token (str): API token.

    '''

    def __init__(self, token: str) -> None:
        super().__init__(token=token)

    @staticmethod
    def gen_uuid(chat_id: str, expired_time: int = 300) -> dict[str, Any]:
        ''' Gen uuid for verify

        Args:
            chat_id (str): Chat id.
            expired_time (int): Expired time.

        Returns:
            Return the data.

        '''
        data = {
            'uuid': str(uuid4()),
            'chat_id': chat_id,
            'code': f'{uuid4().fields[0]:08x}',
            'expired': int(time()) + expired_time,
        }

        mem_cache = MC.get_client()
        mem_cache.set(f"tg:{data['uuid']}", data, expired_time)

        return dict(mem_cache.get(f"tg:{data['uuid']}"))

    @staticmethod
    def temp_fetch_user_data(data: dict[str, Any], expired_time: int = 400) -> None:
        ''' temp fetch user data

        Args:
            data (dict): The data to cache.
            expired_time (int): Expired time.

        '''
        mem_cache = MC.get_client()
        mem_cache.set(f"tgu:{data['message']['from']['id']}",
                      data['message']['from'], expired_time)

    @staticmethod
    def get_temp_user_dta(chat_id: str) -> dict[str, Any]:
        ''' Get temp user data

        Args:
            chat_id (str): Chat id.

        Returns:
            Return the data.

        '''
        mem_cache = MC.get_client()

        data = mem_cache.get(f'tgu:{chat_id}')

        if data:
            return dict(data)

        return {}

gen_uuid staticmethod

gen_uuid(
    chat_id: str, expired_time: int = 300
) -> dict[str, Any]

Gen uuid for verify

Parameters:

Name Type Description Default
chat_id str

Chat id.

required
expired_time int

Expired time.

300

Returns:

Type Description
dict[str, Any]

Return the data.

Source code in module/telegram_bot.py
@staticmethod
def gen_uuid(chat_id: str, expired_time: int = 300) -> dict[str, Any]:
    ''' Gen uuid for verify

    Args:
        chat_id (str): Chat id.
        expired_time (int): Expired time.

    Returns:
        Return the data.

    '''
    data = {
        'uuid': str(uuid4()),
        'chat_id': chat_id,
        'code': f'{uuid4().fields[0]:08x}',
        'expired': int(time()) + expired_time,
    }

    mem_cache = MC.get_client()
    mem_cache.set(f"tg:{data['uuid']}", data, expired_time)

    return dict(mem_cache.get(f"tg:{data['uuid']}"))

get_temp_user_dta staticmethod

get_temp_user_dta(chat_id: str) -> dict[str, Any]

Get temp user data

Parameters:

Name Type Description Default
chat_id str

Chat id.

required

Returns:

Type Description
dict[str, Any]

Return the data.

Source code in module/telegram_bot.py
@staticmethod
def get_temp_user_dta(chat_id: str) -> dict[str, Any]:
    ''' Get temp user data

    Args:
        chat_id (str): Chat id.

    Returns:
        Return the data.

    '''
    mem_cache = MC.get_client()

    data = mem_cache.get(f'tgu:{chat_id}')

    if data:
        return dict(data)

    return {}

temp_fetch_user_data staticmethod

temp_fetch_user_data(
    data: dict[str, Any], expired_time: int = 400
) -> None

temp fetch user data

Parameters:

Name Type Description Default
data dict

The data to cache.

required
expired_time int

Expired time.

400
Source code in module/telegram_bot.py
@staticmethod
def temp_fetch_user_data(data: dict[str, Any], expired_time: int = 400) -> None:
    ''' temp fetch user data

    Args:
        data (dict): The data to cache.
        expired_time (int): Expired time.

    '''
    mem_cache = MC.get_client()
    mem_cache.set(f"tgu:{data['message']['from']['id']}",
                  data['message']['from'], expired_time)