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 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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 
76
77
78
79
80
81
82
83
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 
23
24
25
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 
64
65
66
67
68
69
70
71
72
73
74
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 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@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 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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 
52
53
54
55
56
57
58
59
60
61
62
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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
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 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@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 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@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 
162
163
164
165
166
167
168
169
170
171
172
173
@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)
Last update: June 12, 2022