Бот не реагирует на нажатие кнопок на Inline - клавиатурах
Изначально я писал бота используя Reply - клавиатуры. Попробовал перенести на Inline и он не реагировал на нажатия. Подскажите пожалуйста в чём ошибка. Заранее благодарю.
import sqlite3
import requests
from random import shuffle
import datetime
import os
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Updater, MessageHandler, Filters, ConversationHandler
from telegram.ext import CommandHandler, CallbackQueryHandler
def start(update, context):
# Формируем список клавиш, которые должны быть отображены
context.user_data['user_in_test_mode'], context.user_data['user_in_admin_mode'] = False, False
buttons = [
[
InlineKeyboardButton('Начать тестирование', callback_data='start_test'),
InlineKeyboardButton('Режим администратора', callback_data='start_admin')
]
]
# Создаём объект клавиатуры и потом передаём её в сообщение
markup = InlineKeyboardMarkup(buttons)
update.message.reply_text(f'Здравствуйте, {update.message.chat.first_name}. '
f'Данный бот предназначен для подготовки к ЕГЭ по Русскому языку. '
f'Для того, чтобы начать тестирование напишите /start_test.', reply_markup=markup)
return 'start_of_bot'
def start_inline_query_handler(update, context):
query, query_data = update.callback_query, update.callback_query.data
query.answer()
if query_data == 'start_test':
context.user_data['user_in_test_mode'], context.user_data['user_in_admin_mode'] = True, False
buttons = [
[
InlineKeyboardButton(f'{i} задание', callback_data=f'{i} задание') for i in [4, 7, 9]
],
[
InlineKeyboardButton(f'{i} задание', callback_data=f'{i} задание') for i in [10, 11, 12]
],
[
InlineKeyboardButton(f'{i} задание', callback_data=f'{i} задание') for i in [13, 14, 15]
]
]
markup = InlineKeyboardMarkup(buttons)
query.edit_message_text(text=f'Выберете, пожалуйста, '
f'задание, по которому вы хотите пройти тестирование.', reply_markup=markup)
context.user_data['cnt'], context.user_data['cnt_right_answers'] = 0, 0
context.user_data['flag_allow_answer_from_user'] = False
return 'tasks_numbers_showed'
if query_data == 'start_admin':
query.edit_message_text(text='Для получения доступа к администрированию, введите пароль')
return 'admin_mode_started' # -> Следующий шаг
def num_of_task_response(update, context):
# Проверяем пользовательский ввод и в случае его валидности начинаем готовить задания
query, query_data = update.callback_query, update.callback_query.data
query.answer()
if query_data in [f'{i} задание' for i in [4, 7, 9, 10, 11, 12, 13, 14, 15]]:
num_of_task = update.message.text.split()[0]
# Закидываем в контекст номер задания, выбранного пользователем, чтобы затем использовать его
context.user_data['num_of_task'] = num_of_task
# Осуществляем соединение с БД и создаём курсор
connection = sqlite3.connect('db/database_for_pythonanywhere.db')
cursor = connection.cursor()
# Вытаскиваем из БД подходящие нам задания
list_of_variants = cursor.execute(f'select var, explanation, '
f'right_var, id from task_{num_of_task}').fetchall()
# Перемешиваем список с заданиями и берём первые 10 заданий
shuffle(list_of_variants)
# Закидываем этот список в контекст, чтобы потом его использовать
list_of_variants = list_of_variants[0:10]
context.user_data['list_of_variants'] = list_of_variants
# Закрываем соединение с БД
connection.close()
# Отсылаем пользователю пояснительные сообщения
query.edit_message_text("Давайте начнём тестирование по" + " " + num_of_task + " " + "заданию.")
update.message.reply_text("Для того, чтобы остановить тестирование напишите /stop_test.")
# Узнаём путь до фотографии с заданием
path_to_photo_of_var = context.user_data['list_of_variants'][context.user_data['cnt']][0]
# Отсылаем пользователю пояснительное сообщение
update.message.reply_text(
f"№{context.user_data['cnt'] + 1}, идентификационный номер варианта:"
f" {context.user_data['list_of_variants'][context.user_data['cnt']][3]}")
# Отсылаем пользователю фотографию с заданием
context.bot.send_photo(update.message.chat_id, photo=open(path_to_photo_of_var, 'rb'))
# Позволяем пользователю отвечать на вопрос
context.user_data['flag_allow_answer_from_user'] = True
return 'test_started' # -> Следующий шаг
def test_alg(update, context):
if context.user_data['flag_allow_answer_from_user']:
# Узнаём путь до фотографии с пояснением к заданию
path_to_photo_of_expl = context.user_data['list_of_variants'][context.user_data['cnt']][1]
# Узнаём правильный ответ к заданию
right_answer = context.user_data['list_of_variants'][context.user_data['cnt']][2]
# Сравниваем пользовательский ответ с правильным
if update.message.text == right_answer.lower():
update.message.reply_text('Правильно!')
context.user_data['cnt_right_answers'] += 1
else:
update.message.reply_text('Похоже вы ошиблись(')
# Отсылаем пользователю фотографию с пояснением к заданию
context.bot.send_photo(update.message.chat_id, photo=open(path_to_photo_of_expl, 'rb'))
context.user_data['cnt'] += 1
context.user_data['flag_allow_answer_from_user'] = False
if context.user_data['cnt'] == 10:
update.message.reply_text('Тестирование завершено. Заносим ваши результаты в базу данных.')
# Осуществляем соединение с БД и создаём курсор
connection = sqlite3.connect('db/database_for_pythonanywhere.db')
cursor = connection.cursor()
list_of_user_id = cursor.execute('''select user_id from statistics''').fetchall()
flag_user_in_db = False
current_user_id = update.message.from_user['id']
# Проверяем, проходил ли пользователь тестирование раньше
for id_of_user in list_of_user_id:
if current_user_id in id_of_user:
flag_user_in_db = True
break
# Создаём или обновляем запись о пользователе в базе данных
if flag_user_in_db:
information_about_user = cursor.execute(f'''select user_id, all_task_cnt,
all_task_right_answers_cnt, task_{context.user_data["num_of_task"]}_cnt,
task_{context.user_data["num_of_task"]}_right_answers_cnt from statistics
where user_id = {current_user_id}''').fetchall()
cursor.execute(f'''update statistics set all_task_cnt =
{information_about_user[0][1] + context.user_data["cnt"]}, all_task_right_answers_cnt =
{information_about_user[0][2] + context.user_data["cnt_right_answers"]},
task_{context.user_data["num_of_task"]}_cnt = {information_about_user[0][3] + context.user_data["cnt"]},
task_{context.user_data["num_of_task"]}_right_answers_cnt =
{information_about_user[0][4] + context.user_data["cnt_right_answers"]} where user_id =
{information_about_user[0][0]}''')
connection.commit()
else:
cursor.execute(f'''insert into statistics(user_id, first_name, last_name, all_task_cnt,
all_task_right_answers_cnt, task_{context.user_data["num_of_task"]}_cnt,
task_{context.user_data["num_of_task"]}_right_answers_cnt) values({current_user_id},
"{update.message.chat.first_name}", "{update.message.chat.last_name}", {context.user_data["cnt"]},
{context.user_data["cnt_right_answers"]}, {context.user_data["cnt"]},
{context.user_data["cnt_right_answers"]})''')
connection.commit()
connection.close()
update.message.reply_text(f'Количество правильных ответов: {context.user_data["cnt_right_answers"]}\n'
f'Общее количество ответов: {context.user_data["cnt"]}.')
update.message.reply_text('Благодарим вас за участие в тестировании!')
update.message.reply_text('Для того, чтобы продолжить работу с ботом, напишите /start.')
context.user_data['user_in_test_mode'], context.user_data['user_in_admin_mode'] = False, False
return ConversationHandler.END # -> Конец диалога
if not context.user_data['flag_allow_answer_from_user']:
path_to_photo_of_var = context.user_data['list_of_variants'][context.user_data['cnt']][0]
update.message.reply_text(
f"№{context.user_data['cnt'] + 1}, идентификационный номер варианта:"
f" {context.user_data['list_of_variants'][context.user_data['cnt']][3]}")
context.bot.send_photo(update.message.chat_id, photo=open(path_to_photo_of_var, 'rb'))
context.user_data['flag_allow_answer_from_user'] = True
def stop_test(update, context):
if context.user_data['user_in_test_mode'] and not context.user_data['user_in_admin_mode']:
update.message.reply_text('Тестирование остановлено. Чтобы возобновить тестирование напишите /start_test.')
update.message.reply_text(f'Количество правильных ответов: {context.user_data["cnt_right_answers"]}\n'
f'Общее количество ответов: {context.user_data["cnt"]}.')
update.message.reply_text('Благодарим вас за участие в тестировании!')
return ConversationHandler.END # -> Конец диалога
def check_admin_password(update, context):
# Читаем все пароли из файла
with open("passwords.txt", encoding='utf-8') as file:
list_of_passwords = file.read().strip()
# Проверяем пользовательский ввод на вхождение в список паролей
if update.message.text in list_of_passwords:
context.user_data['user_in_test_mode'], context.user_data['user_in_admin_mode'] = False, True
update.message.reply_text('Доступ к администрированию получен. Для того, чтобы выйти из режима '
'администрирования напишите /stop_admin.')
buttons = [
[
InlineKeyboardButton('Добавить вариант задания', callback_data='admin_want_add_task'),
InlineKeyboardButton('Удалить вариант задания', callback_data='admin_want_delete_task')
],
[
InlineKeyboardButton('Выгрузить статистику из базы данных', callback_data='admin_want_stat')
]
]
actions_keyboard = InlineKeyboardMarkup(buttons)
update.message.reply_text('Выберете при помощи клавиатуры действия, '
'которые вы хотите совершить с ботом', reply_markup=actions_keyboard)
return 'actions_showed' # -> Следующий шаг
else:
update.message.reply_text('Пароль введён неверно, вам отказано в доступе к администрированию.')
return ConversationHandler.END # -> Конец диалога
def check_admin_actions(update, context):
# Узнаём, что админ захотел сделать
query, query_data = update.callback_query, update.callback_query.data
query.answer()
if query_data == 'admin_want_add_task':
buttons = [
[
InlineKeyboardButton(f'{i} задание', callback_data=f'{i} задание') for i in [4, 7, 9]
],
[
InlineKeyboardButton(f'{i} задание', callback_data=f'{i} задание') for i in [10, 11, 12]
],
[
InlineKeyboardButton(f'{i} задание', callback_data=f'{i} задание') for i in [13, 14, 15]
]
]
markup = InlineKeyboardMarkup(buttons)
query.edit_message_text('Введите номер задания, вариант которого вы хотите добавить.', reply_markup=markup)
return 'admin_want_to_add_task' # -> Следующий шаг
if query_data == 'admin_want_delete_task':
update.message.reply_text('Для того, чтобы удалить вариант задания, напишите через / '
'номер задания, вариант которого вы хотите удалить, и идентификационный '
'номер варианта данного задания.')
return 'admin_want_to_delete_task' # -> Следующий шаг
if query_data == 'admin_want_stat':
query.edit_message_text('Произвожу выгрузку статистики из базы данных')
# Осуществляем соединение с БД и создаём курсор
connection = sqlite3.connect('db/database_for_pythonanywhere.db')
cursor = connection.cursor()
statistic = cursor.execute('''select * from statistics''').fetchall()
# Открываем текстовый файл для дальнейшей записи в него
file = open('statistics.txt', encoding='utf-8', mode='w')
file.write(f'Время выгрузки статистики: {datetime.datetime.now()}\n')
file.write(f'\n')
cnt_of_user = 1
# Записываем информацию о каждом пользователе в файл
for user in statistic:
file.write(f'Пользователь №{cnt_of_user}\n')
file.write(f'\n')
file.write(f'ID пользователя: {user[0]}\n')
file.write(f'Имя пользователя: {user[1]}\n')
file.write(f'Фамилия пользователя: {user[2]}\n')
file.write(f'Общее количество решённых вариантов заданий: {user[3]}\n')
file.write(f'Общее количество правильно решённых вариантов заданий: {user[4]}\n')
cnt = 5
for i in [4, 7, 9, 10, 11, 12, 13, 14, 15]:
file.write(f'Общее количество решённых вариантов задания №{i}: {user[cnt]}\n')
cnt += 1
file.write(f'Общее количество правильно решённых вариантов задания №{i}: {user[cnt]}\n')
cnt += 1
file.write(f'\n')
# Закрываем файл
file.close()
update.message.reply_text('Выполнено.')
# Отправляем файл в чат
context.bot.send_document(update.message.chat_id, document=open('statistics.txt', mode='rb'))
# Удаляем файл со статистикой на сервере
os.remove('statistics.txt')
connection.close()
context.user_data['user_in_test_mode'], context.user_data['user_in_admin_mode'] = False, False
return ConversationHandler.END # -> Конец диалога
else:
update.message.reply_text('Команда не распознана, повторите ещё раз.')
def admin_action_add_task_get_num(update, context):
# Проверяем валидность введённого админом номера задания
query, query_data = update.callback_query, update.callback_query.data.split()[0]
query.answer()
if query_data in [f'{i}' for i in [4, 7, 9, 10, 11, 12, 13, 14, 15]]:
# Заносим в контекст номер задания, которое админ хочет добавить
context.user_data['num_of_task_to_add'] = query_data
update.message.reply_text('Отправьте фотографию, содержащую вариант '
'задания, который вы хотите создать.')
return 'num_of_task_for_add_ok' # -> Следующий шаг
else:
update.message.reply_text('Ошибка в номере задания, повторите ещё раз.')
def admin_action_add_task_get_var(update, context):
# Заносим в контекст ID фотографии, содержащей текст задания
context.user_data['ID_of_var'] = update.message.photo[3]['file_id']
update.message.reply_text('Отправьте правильный ответ к варианту задания, который вы хотите создать.')
return 'ID_var_ok' # -> Следующий шаг
def admin_action_add_task_get_right_answer(update, context):
# Заносим в контекст правильный ответ к заданию
context.user_data['right_answer'] = update.message.text
update.message.reply_text('Отправьте фотографию, содержащую пояснение к варианту '
'задания, который вы хотите создать.')
return 'right_answer_ok' # -> Следующий шаг
def admin_action_add_task_get_expl(update, context):
# Заносим в контекст ID фотографии, содержащей пояснение к заданию
context.user_data['ID_of_expl'] = update.message.photo[3]['file_id']
update.message.reply_text('Добавляю новый вариант задания')
# Осуществляем соединение с БД и создаём курсор
connection = sqlite3.connect('db/database_for_pythonanywhere.db')
cursor = connection.cursor()
# Узнаём последний ID в БД и создаём новый
id_of_new_var = cursor.execute(f'''select id from
task_{context.user_data['num_of_task_to_add']}''').fetchall()[-1][0] + 1
# Считываем токен к боту
with open("token.txt", encoding='utf-8') as file:
token = file.read().strip()
# Вытаскиваем фотографии, полученные от пользователя, с сервера Telegram и закидываем их на наш сервер
response = requests.get(f'https://api.telegram.org/'
f'bot{token}/getFile?file_id={context.user_data["ID_of_var"]}').json()
photo_of_var = requests.get(f'https://api.telegram.org/file/bot'
f'{token}/{response["result"]["file_path"]}').content
path_to_var = f'data/task_{context.user_data["num_of_task_to_add"]}/' \
f'task_{context.user_data["num_of_task_to_add"]}_n_{id_of_new_var}.png'
with open(path_to_var, "wb") as file:
file.write(photo_of_var)
response = requests.get(f'https://api.telegram.org/'
f'bot{token}/getFile?file_id={context.user_data["ID_of_expl"]}').json()
photo_of_expl = requests.get(f'https://api.telegram.org/file/bot'
f'{token}/{response["result"]["file_path"]}').content
path_to_expl = f'data/explanation/task_{context.user_data["num_of_task_to_add"]}/' \
f'task_{context.user_data["num_of_task_to_add"]}_n_{id_of_new_var}_expl.png'
with open(path_to_expl, "wb") as file:
file.write(photo_of_expl)
# Делаем запись в БД
cursor.execute(f'''insert into task_{context.user_data["num_of_task_to_add"]}(id, right_var,
var, explanation) values ({id_of_new_var}, "{context.user_data["right_answer"]}",
"{path_to_var}", "{path_to_expl}")''')
connection.commit()
connection.close()
update.message.reply_text('Добавление произведено.')
context.user_data['user_in_test_mode'], context.user_data['user_in_admin_mode'] = False, False
return ConversationHandler.END # -> Конец диалога
def admin_action_delete_task(update, context):
list_of_args = update.message.text.split('/')
num_of_task, id_of_var = list_of_args[0], list_of_args[1]
# Проверяем валидность пользовательского ввода
if num_of_task not in [f'{i}' for i in [4, 7, 9, 10, 11, 12, 13, 14, 15]]:
update.message.reply_text('Неверно введён номер задания')
else:
# Осуществляем соединение с БД и создаём курсор
connection = sqlite3.connect('db/database_for_pythonanywhere.db')
cursor = connection.cursor()
# Узнаём пути до фотографий с заданием и пояснением и удаляем их с сервера
list_of_delete_task = cursor.execute(
f'select var, explanation from task_{num_of_task} where id = {id_of_var}').fetchall()
path_to_var, path_to_expl = list_of_delete_task[0][0], list_of_delete_task[0][1]
os.remove(path_to_var)
os.remove(path_to_expl)
# Удаляем запись из БД
cursor.execute(f'''delete from task_{num_of_task} where id = {id_of_var}''')
connection.commit()
connection.close()
update.message.reply_text('Удаление произведено.')
context.user_data['user_in_test_mode'], context.user_data['user_in_admin_mode'] = False, False
return ConversationHandler.END # -> Конец диалога
def stop_admin(update, context):
if not context.user_data['user_in_test_mode'] and context.user_data['user_in_admin_mode']:
update.message.reply_text('Произведён выход из режима администратора')
return ConversationHandler.END # -> Конец диалога
def main():
# Считываем токен к боту
with open("token.txt", encoding='utf-8') as file:
token = file.read().strip()
updater = Updater(token, use_context=True)
dp = updater.dispatcher
dp.add_handler(CommandHandler('start', start, pass_user_data=True))
bot_conv_handler = ConversationHandler(
entry_points=[
CommandHandler('start', start, pass_user_data=True)
],
states={
'start_of_bot': [
CallbackQueryHandler(start_inline_query_handler, pass_user_data=True)
],
'tasks_numbers_showed': [
CallbackQueryHandler(num_of_task_response, pass_user_data=True)
],
'test_started': [
MessageHandler(Filters.text & (~ Filters.command), test_alg,
pass_user_data=True)
],
'admin_mode_started': [
MessageHandler(Filters.text & (~ Filters.command),
check_admin_password, pass_user_data=True)
],
'actions_showed': [
CallbackQueryHandler(check_admin_actions, pass_user_data=True)
],
'admin_want_to_add_task': [
CallbackQueryHandler(admin_action_add_task_get_num,
pass_user_data=True)
],
'num_of_task_for_add_ok': [
MessageHandler(Filters.photo & (~ Filters.command),
admin_action_add_task_get_var,
pass_user_data=True)
],
'ID_var_ok': [
MessageHandler(Filters.text & (~ Filters.command),
admin_action_add_task_get_right_answer,
pass_user_data=True)
],
'right_answer_ok': [
MessageHandler(Filters.photo & (~ Filters.command),
admin_action_add_task_get_expl,
pass_user_data=True)
],
'admin_want_to_delete_task': [
MessageHandler(Filters.text & (~ Filters.command), admin_action_delete_task, pass_user_data=True)
]
},
fallbacks=[
CommandHandler('stop_test', stop_test, pass_user_data=True),
CommandHandler('stop_admin', stop_admin, pass_user_data=True)
],
per_message=True
)
dp.add_handler(bot_conv_handler)
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()