-
-
Notifications
You must be signed in to change notification settings - Fork 254
/
duckbot.py
126 lines (101 loc) · 5.14 KB
/
duckbot.py
1
2
3
4
5
6
7
8
9
10
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
125
126
import logging
import sys, time
import traceback
import telegram.error
import nuconfig
log = logging.getLogger(__name__)
def factory(cfg: nuconfig.NuConfig):
"""Construct a DuckBot type based on the passed config."""
def catch_telegram_errors(func):
"""Decorator, can be applied to any function to retry in case of Telegram errors."""
def result_func(*args, **kwargs):
while True:
try:
return func(*args, **kwargs)
# Bot was blocked by the user
except telegram.error.Unauthorized:
log.debug(f"Unauthorized to call {func.__name__}(), skipping.")
return None
# Telegram API didn't answer in time
except telegram.error.TimedOut:
log.warning(f"Timed out while calling {func.__name__}(),"
f" retrying in {cfg['Telegram']['timed_out_pause']} secs...")
time.sleep(cfg["Telegram"]["timed_out_pause"])
continue
# Telegram is not reachable
except telegram.error.NetworkError as error:
log.error(f"Network error while calling {func.__name__}(),"
f" retrying in {cfg['Telegram']['error_pause']} secs...\n"
f"Full error: {error.message}")
time.sleep(cfg["Telegram"]["error_pause"])
continue
# Unknown error
except telegram.error.TelegramError as error:
if error.message.lower() in ["bad gateway", "invalid server response"]:
log.warning(f"Bad Gateway while calling {func.__name__}(),"
f" retrying in {cfg['Telegram']['error_pause']} secs...")
time.sleep(cfg["Telegram"]["error_pause"])
continue
elif error.message.lower() == "timed out":
log.warning(f"Timed out while calling {func.__name__}(),"
f" retrying in {cfg['Telegram']['timed_out_pause']} secs...")
time.sleep(cfg["Telegram"]["timed_out_pause"])
continue
else:
log.error(f"Telegram error while calling {func.__name__}(),"
f" retrying in {cfg['Telegram']['error_pause']} secs...\n"
f"Full error: {error.message}")
traceback.print_exception(*sys.exc_info())
time.sleep(cfg["Telegram"]["error_pause"])
continue
return result_func
class DuckBot:
def __init__(self, *args, **kwargs):
self.bot = telegram.Bot(token=cfg["Telegram"]["token"], *args, **kwargs)
@catch_telegram_errors
def send_message(self, *args, **kwargs):
# All messages are sent in HTML parse mode
return self.bot.send_message(parse_mode="HTML", *args, **kwargs)
@catch_telegram_errors
def send_photo(self, *args, **kwargs):
return self.bot.send_photo(parse_mode="HTML", *args, **kwargs)
@catch_telegram_errors
def edit_message_text(self, *args, **kwargs):
# All messages are sent in HTML parse mode
return self.bot.edit_message_text(parse_mode="HTML", *args, **kwargs)
@catch_telegram_errors
def edit_message_caption(self, *args, **kwargs):
# All messages are sent in HTML parse mode
return self.bot.edit_message_caption(parse_mode="HTML", *args, **kwargs)
@catch_telegram_errors
def edit_message_reply_markup(self, *args, **kwargs):
return self.bot.edit_message_reply_markup(*args, **kwargs)
@catch_telegram_errors
def get_updates(self, *args, **kwargs):
return self.bot.get_updates(*args, **kwargs)
@catch_telegram_errors
def get_me(self, *args, **kwargs):
return self.bot.get_me(*args, **kwargs)
@catch_telegram_errors
def answer_callback_query(self, *args, **kwargs):
return self.bot.answer_callback_query(*args, **kwargs)
@catch_telegram_errors
def answer_pre_checkout_query(self, *args, **kwargs):
return self.bot.answer_pre_checkout_query(*args, **kwargs)
@catch_telegram_errors
def send_invoice(self, *args, **kwargs):
return self.bot.send_invoice(*args, **kwargs)
@catch_telegram_errors
def get_file(self, *args, **kwargs):
return self.bot.get_file(*args, **kwargs)
@catch_telegram_errors
def send_chat_action(self, *args, **kwargs):
return self.bot.send_chat_action(*args, **kwargs)
@catch_telegram_errors
def delete_message(self, *args, **kwargs):
return self.bot.delete_message(*args, **kwargs)
@catch_telegram_errors
def send_document(self, *args, **kwargs):
return self.bot.send_document(*args, **kwargs)
# More methods can be added here
return DuckBot