-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbot.py
More file actions
114 lines (93 loc) · 3.71 KB
/
Copy pathbot.py
File metadata and controls
114 lines (93 loc) · 3.71 KB
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
import logging
import os
import sys
from threading import Thread
from redis import Redis
from telegram import Update
from telegram.ext import (
Updater, CommandHandler, CallbackContext, CallbackQueryHandler,
Dispatcher, Filters, MessageHandler,
)
from sentry_sdk import init as init_setry, capture_exception, configure_scope
from learn_python_bot import __version__
from learn_python_bot.config import (TELEGRAM_PROXY_SETTINGS, TELEGRAM_BOT_TOKEN, REDIS_URL,
SENTRY_URL, LOG_LEVEL)
from learn_python_bot.api.airtable import AirtableAPI
from learn_python_bot.decorators import for_admins_only
from learn_python_bot.handlers.admin import admin_keyboard, admin_show_students, get_admin_announce_command_handler
from learn_python_bot.handlers.start import start
from learn_python_bot.handlers.student_feedback_command import get_student_feedback_command_handler
from learn_python_bot.handlers.student_weekly_feedback import process_feedback
from learn_python_bot.scheduler import init_schedulers
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=getattr(logging, LOG_LEVEL),
)
logger = logging.getLogger(__name__)
if SENTRY_URL:
init_setry(SENTRY_URL, release=__version__)
def error(update: Update, context: CallbackContext) -> None:
logger.warning('Update "%s" caused error "%s"', update, context.error)
if SENTRY_URL:
with configure_scope() as scope:
if update and update.effective_chat:
scope.user = {
'username': update.effective_chat.username,
'chat_id': update.effective_chat.id,
}
capture_exception(
context.error,
)
else:
logger.error(context.error)
def mutate_bot_to_be_restartable(updater: Updater):
def stop_and_restart() -> None:
updater.stop()
os.execl(sys.executable, sys.executable, *sys.argv)
@for_admins_only
def restart(update: Update, context: CallbackContext) -> None:
update.message.reply_text('Bot is restarting...')
Thread(target=stop_and_restart).start()
dp = updater.dispatcher
dp.add_handler(MessageHandler(Filters.regex('^(Restart bot)$'), restart))
def set_initial_bot_data(dispatcher: Dispatcher) -> None:
airtable_api = AirtableAPI.get_default_api()
students = airtable_api.extract_students(
airtable_api.fetch_students_data(),
)
dispatcher.bot_data.update({
'airtable_api': airtable_api,
'students': students,
'redis': Redis.from_url(REDIS_URL),
})
@for_admins_only
def set_initial_bot_data_command(update: Update, context: CallbackContext) -> None:
set_initial_bot_data(context.dispatcher)
students_count = len(context.dispatcher.bot_data['students'])
update.message.reply_text(f'Loaded {students_count} students')
def main() -> None:
handlers = [
CommandHandler('start', start),
CommandHandler('admin', admin_keyboard),
MessageHandler(Filters.regex('^(Show students)$'), admin_show_students),
MessageHandler(Filters.regex('^(Reload students)$'), set_initial_bot_data_command),
get_student_feedback_command_handler(),
CallbackQueryHandler(process_feedback),
get_admin_announce_command_handler(),
]
updater = Updater(
TELEGRAM_BOT_TOKEN,
use_context=True,
request_kwargs=TELEGRAM_PROXY_SETTINGS,
)
dp = updater.dispatcher
mutate_bot_to_be_restartable(updater)
init_schedulers(updater)
set_initial_bot_data(dp)
for handler in handlers:
dp.add_handler(handler)
dp.add_error_handler(error)
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()