145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
import _thread
|
|
import os
|
|
import logging
|
|
import telebot
|
|
from telebot import types
|
|
from pymongo import MongoClient
|
|
|
|
telebot.logger.setLevel(logging.DEBUG)
|
|
|
|
# Bot & config
|
|
bot = telebot.TeleBot(os.environ['TOKEN'])
|
|
admin = int(os.environ['ADMIN'])
|
|
|
|
# Acceso a bd
|
|
client = MongoClient('database:27017')
|
|
db = client.bot
|
|
tests = db.tests
|
|
users = db.users
|
|
|
|
# Funciones de utilidad
|
|
def find_user(cid):
|
|
return users.find_one(str(cid))
|
|
|
|
def is_user(cid):
|
|
return find_user(cid) is not None
|
|
|
|
def add_user(cid):
|
|
users.insert_one({
|
|
"_id": str(cid),
|
|
"taken_tests": [],
|
|
"current_tests": []
|
|
})
|
|
|
|
def add_test(cid, test_number):
|
|
user = find_user(cid)
|
|
if user and test_number not in [x['test'] for x in user['current_tests']]:
|
|
test = tests.find_one({'test': int(test_number)})
|
|
users.update_one({'_id':str(cid)},{'$push': {'current_tests': test}})
|
|
|
|
|
|
def restart_process():
|
|
global client
|
|
client.close()
|
|
_thread.interrupt_main()
|
|
|
|
def is_admin(cid):
|
|
return str(cid) == str(admin)
|
|
|
|
def split(a, n):
|
|
k, m = divmod(len(a), n)
|
|
return (a[i*k+min(i, m):(i+1)*k+min(i+1, m)] for i in range(n))
|
|
|
|
def find_question(test, question):
|
|
questions = tests.find_one({'test': int(test)}, {'preguntas':{'$elemMatch': {'pregunta': int(question)}}, '_id':0})
|
|
if questions:
|
|
return questions['preguntas'][0]
|
|
return {}
|
|
|
|
def check_answer(question, answer):
|
|
return '✅ Correcto' if question['respuestas'][int(answer)]['correcta'] else '❌ Error'
|
|
|
|
def generate_tests_keyboard():
|
|
keyboard = types.InlineKeyboardMarkup(row_width=4)
|
|
buttons = []
|
|
for i in range(1,33):
|
|
buttons.append(types.InlineKeyboardButton('Test {}'.format(i),callback_data='test {}'.format(100+i)))
|
|
for chunk in split(buttons, 4):
|
|
keyboard.add(*chunk)
|
|
return keyboard
|
|
|
|
def generate_test_keyboard(question, test):
|
|
keyboard = types.InlineKeyboardMarkup()
|
|
for i,answer in enumerate(question.get('respuestas')):
|
|
keyboard.add(types.InlineKeyboardButton('{}'.format(answer['texto']), callback_data='pregunta {} {} {}'.format(test, question['pregunta'], i)))
|
|
next_prev_buttons = []
|
|
if question.get('pregunta') == 1:
|
|
next_prev_buttons.append(types.InlineKeyboardButton('⏩', callback_data='pregunta {} {} +'.format(test, question['pregunta'])))
|
|
elif (question.get('pregunta') == 20 and str(test) != '132') or (question.get('pregunta') == 17 and str(test) == '132'):
|
|
next_prev_buttons.append(types.InlineKeyboardButton('⏪', callback_data='pregunta {} {} -'.format(test, question['pregunta'])))
|
|
else:
|
|
next_prev_buttons.append(types.InlineKeyboardButton('⏪', callback_data='pregunta {} {} -'.format(test, question['pregunta'])))
|
|
next_prev_buttons.append(types.InlineKeyboardButton('⏩', callback_data='pregunta {} {} +'.format(test, question['pregunta'])))
|
|
keyboard.add(*next_prev_buttons)
|
|
return keyboard
|
|
|
|
# Manejadores de comandos
|
|
@bot.message_handler(commands=['start'])
|
|
def start_handler(m):
|
|
cid = m.chat.id
|
|
if not is_user(cid):
|
|
add_user(cid)
|
|
bot.send_message(cid, "Bienvenid@")
|
|
|
|
@bot.message_handler(commands=['reload'])
|
|
def reload_handler(m):
|
|
cid = m.chat.id
|
|
if is_admin(cid):
|
|
restart_process()
|
|
|
|
@bot.message_handler(commands=['tests'])
|
|
def tests_handler(m):
|
|
cid = m.chat.id
|
|
keyboard = generate_tests_keyboard()
|
|
bot.send_message(cid, "Haz click en un test para empezar a resolverlo", reply_markup=keyboard)
|
|
|
|
@bot.callback_query_handler(func=lambda call: call.data.startswith('test'))
|
|
def test_callback_handler(call):
|
|
cid = call.message.chat.id
|
|
test = call.data.split()[-1]
|
|
question = find_question(test, 1)
|
|
keyboard = generate_test_keyboard(question, test)
|
|
txt = "Test: {}\nPregunta: {}\n\n{}"
|
|
bot.send_photo(cid, question.get('img'), caption=txt.format(test, 1, question['encabezado']), reply_markup=keyboard)
|
|
add_test(cid, test)
|
|
bot.answer_callback_query(call.id)
|
|
|
|
|
|
@bot.callback_query_handler(func=lambda call: call.data.startswith('pregunta') and call.data[-1] in [str(x) for x in range(0,4)])
|
|
def select_option_callback_handler(call):
|
|
cid = call.message.chat.id
|
|
mid = call.message.id
|
|
_, test, q, answer = call.data.split()
|
|
question = find_question(test, q)
|
|
answer_reply = check_answer(question, answer)
|
|
bot.answer_callback_query(call.id, answer_reply)
|
|
|
|
@bot.callback_query_handler(func=lambda call: call.data.startswith('pregunta') and (call.data.endswith('+') or call.data.endswith('-')))
|
|
def question_pager_callback_handler(call):
|
|
cid = call.message.chat.id
|
|
mid = call.message.id
|
|
_, test, q, action = call.data.split()
|
|
question = find_question(test, eval(q+action+'1'))
|
|
keyboard = generate_test_keyboard(question, test)
|
|
txt = "Test: {}\nPregunta: {}\n\n{}"
|
|
try:
|
|
bot.edit_message_media(types.InputMediaPhoto(question.get('img'), caption=txt.format(test, question['pregunta'], question['encabezado'])), cid, mid, reply_markup=keyboard)
|
|
bot.answer_callback_query(call.id)
|
|
except:
|
|
bot.answer_callback_query(call.id, "No se pudo cargar la pregunta")
|
|
|
|
|
|
# Iniciar bot
|
|
bot.send_message(admin, "Reiniciado")
|
|
bot.polling(skip_pending=True)
|