diff options
| -rw-r--r-- | backend_api.py | 29 | ||||
| -rw-r--r-- | bot.py | 26 | ||||
| -rw-r--r-- | keyboards.py | 54 | ||||
| -rw-r--r-- | states.py | 175 | ||||
| -rw-r--r-- | utils.py | 4 |
5 files changed, 257 insertions, 31 deletions
diff --git a/backend_api.py b/backend_api.py index b50faf6..951c570 100644 --- a/backend_api.py +++ b/backend_api.py @@ -3,6 +3,7 @@ import urllib from config import BACKEND_URL import logging import json +import datetime as dt from typing import Tuple, Dict @@ -41,6 +42,10 @@ def get_request(url: str, **kwargs): return make_request("get", url, **kwargs) +def patch_request(url: str, **kwargs): + return make_request("patch", url, **kwargs) + + def register_user(tg_id: int, username: str, fullname: str) -> Tuple[int, Dict]: logger.debug(f"Trying to register user with id={tg_id}; username={username}") return post_request(f"{BACKEND_URL}/profiles/", data={ @@ -55,6 +60,11 @@ def get_tasks(): return get_request(f"{BACKEND_URL}/tasks/") +def get_published_tasks(): + logger.debug(f"Trying to retrieve all published tasks") + return get_request(f"{BACKEND_URL}/api/tasks/published/") + + def get_task(title: str) -> Tuple[int, Dict]: logger.debug(f"Trying to retrieve task with title={title}") return get_request(f"{BACKEND_URL}/api/get_task/" + urllib.parse.quote(title)) @@ -92,3 +102,22 @@ def get_attempts(tg_id: int=None, task_title: str=None): data["task_title"] = task_title return get_request(f"{BACKEND_URL}/api/attempts/", data=data) + + +def get_profile(tg_id: int): + return get_request(f"{BACKEND_URL}/api/profile/get/{tg_id}") + + +def publish_task(title: str): + url_title = urllib.parse.quote(title) + return patch_request(f"{BACKEND_URL}/api/tasks/{url_title}/update/", data={ + "first_published": dt.datetime.now(), + "is_public": True + }) + + +def hide_task(title: str): + url_title = urllib.parse.quote(title) + return patch_request(f"{BACKEND_URL}/api/tasks/{url_title}/update/", data={ + "is_public": False + }) @@ -12,10 +12,10 @@ from config import TG_TOKEN, REQUEST_KWARGS import backend_api from keyboards import ( MenuKeyboard, TasksKeyboard, TaskChosenKeyboard, ContinueKeyboard, - AnsweringKeyboard, + AnsweringKeyboard, AdminKeyboard ) from utils import * -from states import States +from states import States, AdminStates logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) @@ -98,6 +98,7 @@ conversation_handler = ConversationHandler( MessageHandler(Filters.regex(MenuKeyboard.CHOOSE_TASK), States.choose_task, pass_user_data=True), MessageHandler(Filters.regex(MenuKeyboard.TOP_10), States.top_10, pass_user_data=True), MessageHandler(Filters.regex(MenuKeyboard.RULES), States.rules, pass_user_data=True), + CommandHandler('admin', AdminStates.admin_panel, pass_user_data=True) ], TASK_CHOOSING: [ MessageHandler(Filters.regex(TasksKeyboard.CANCEL), States.main_menu, pass_user_data=True), @@ -112,7 +113,26 @@ conversation_handler = ConversationHandler( MessageHandler(Filters.text, States.accept_answer, pass_user_data=True), ], ANSWER_RIGHT: [MessageHandler(Filters.text, States.main_menu, pass_user_data=True)], - ANSWER_WRONG: [MessageHandler(Filters.text, States.show_task, pass_user_data=True)] + ANSWER_WRONG: [MessageHandler(Filters.text, States.show_task, pass_user_data=True)], + + ADMIN_MENU: [ + MessageHandler(Filters.regex(AdminKeyboard.CANCEL), States.main_menu, pass_user_data=True), + MessageHandler(Filters.regex(AdminKeyboard.PUBLISH_TASK), AdminStates.choose_task_publish, pass_user_data=True), + MessageHandler(Filters.regex(AdminKeyboard.HIDE_TASK), AdminStates.choose_task_hide, pass_user_data=True), + ], + + ADMIN_TASK_CHOOSE_HIDE: [ + MessageHandler(Filters.regex(TasksKeyboard.CANCEL), AdminStates.admin_panel, pass_user_data=True), + MessageHandler(Filters.text, AdminStates.hide_task, pass_user_data=True), + ], + + ADMIN_TASK_CHOOSE_PUBLISH: [ + MessageHandler(Filters.regex(TasksKeyboard.CANCEL), AdminStates.admin_panel, pass_user_data=True), + MessageHandler(Filters.text, AdminStates.publish_task, pass_user_data=True), + ], + + ADMIN_TASK_PUBLISHED: [MessageHandler(Filters.text, AdminStates.admin_panel, pass_user_data=True)], + ADMIN_ACCESS_DENIED: [MessageHandler(Filters.text, States.main_menu, pass_user_data=True)], }, fallbacks=[CommandHandler('stop', stop)] diff --git a/keyboards.py b/keyboards.py index cbdd770..d544059 100644 --- a/keyboards.py +++ b/keyboards.py @@ -9,13 +9,22 @@ class Keyboard(ABC): class MenuKeyboard(Keyboard): - CHOOSE_TASK = "Выбрать задание" - TOP_10 = "Топ-10" - RULES = "Правила" + CHOOSE_TASK = "Выбрать задание📚" + TOP_10 = "Топ-10📊" + RULES = "Правилаℹ️" ADMIN = "/admin" @classmethod def get_keyboard(cls, telegram_id=None): + if telegram_id is not None: + status_code, data = backend_api.get_profile(telegram_id) + if status_code == 200 and data["is_admin"]: + return [ + [cls.ADMIN], + [cls.CHOOSE_TASK], + [cls.TOP_10, cls.RULES], + ] + return [ [cls.CHOOSE_TASK], [cls.TOP_10, cls.RULES], @@ -23,7 +32,7 @@ class MenuKeyboard(Keyboard): class BackToMenuKeyboard(Keyboard): - CANCEL = "Вернуться в меню" + CANCEL = "Вернуться в меню↩️" @classmethod def get_keyboard(cls, telegram_id=None): @@ -31,7 +40,20 @@ class BackToMenuKeyboard(Keyboard): class TasksKeyboard(Keyboard): - CANCEL = "Вернуться в меню" + CANCEL = "Вернуться в меню↩️" + + @classmethod + def get_keyboard(cls, telegram_id=None): + status, tasks = backend_api.get_published_tasks() + titles_keyboard = [[cls.CANCEL]] + if status == 200: + titles_keyboard.extend([task.get("title")] for task in tasks) + + return titles_keyboard + + +class PublishTasksKeyboard(Keyboard): + CANCEL = "Вернуться в меню↩️" @classmethod def get_keyboard(cls, telegram_id=None): @@ -44,8 +66,8 @@ class TasksKeyboard(Keyboard): class TaskChosenKeyboard(Keyboard): - TYPE_ANSWER = "Ввести ответ" - CANCEL = "Назад" + TYPE_ANSWER = "Ввести ответ✏️" + CANCEL = "Назад↩️" @classmethod def get_keyboard(cls, telegram_id=None): @@ -56,7 +78,7 @@ class TaskChosenKeyboard(Keyboard): class ContinueKeyboard(Keyboard): - CONTINUE = "Продолжить" + CONTINUE = "Продолжить➡️" @classmethod def get_keyboard(cls, telegram_id=None): @@ -64,8 +86,22 @@ class ContinueKeyboard(Keyboard): class AnsweringKeyboard(Keyboard): - CANCEL = "Отмена" + CANCEL = "Отмена↩️" @classmethod def get_keyboard(cls, telegram_id=None): return [[cls.CANCEL]] + + +class AdminKeyboard(Keyboard): + CANCEL = "Вернуться в меню↩️" + PUBLISH_TASK = "Опубликовать задачу" + HIDE_TASK = "Скрыть задачу" + + @classmethod + def get_keyboard(cls, telegram_id=None): + return [ + [cls.CANCEL], + [cls.PUBLISH_TASK], + [cls.HIDE_TASK], + ] @@ -2,7 +2,7 @@ from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove from keyboards import ( MenuKeyboard, TasksKeyboard, TaskChosenKeyboard, ContinueKeyboard, - AnsweringKeyboard, + AnsweringKeyboard, AdminKeyboard, PublishTasksKeyboard ) from utils import * import backend_api @@ -36,10 +36,38 @@ class States: def main_menu(bot: Bot, update: Update, user_data: dict): user_data["chosen_task"] = None - update.message.reply_text("Твой счет: 0") + status_code, response = backend_api.get_attempts( + tg_id=update.message.from_user.id, + task_title=user_data["chosen_task"] + ) + + menu_text = [ + + ] + + if status_code == 200: + if len(response) != 0: + menu_text.append("Твои решенные задачи:") + for attempt in response: + menu_text.append( + f"_{attempt['task']['title']}_ " + f"({attempt['task']['base_score']})" + ) + else: + menu_text.append("У тебя еще нет решенных задач") + else: + menu_text.append( + "К сожалению, не удалось получить данные о твоих попытках =(\n" + "Попробуй обратиться к боту чуть позже." + ) + + menu_text.append("\n*Итоговый счет*: 0\n*Место в топе*: 0") + + update.message.reply_text("\n".join(menu_text), parse_mode="Markdown") + update.message.reply_text( "Выбери следующее действие...", - reply_markup=ReplyKeyboardMarkup(MenuKeyboard.get_keyboard()) + reply_markup=ReplyKeyboardMarkup(MenuKeyboard.get_keyboard(update.message.from_user.id)) ) return MAIN_MENU @@ -49,11 +77,21 @@ class States: def choose_task(bot: Bot, update: Update, user_data: dict): user_data["chosen_task"] = None - update.message.reply_text( - "Какую задачу ты хочешь сдать?", - reply_markup=ReplyKeyboardMarkup(TasksKeyboard.get_keyboard()) - ) - return TASK_CHOOSING + status, published = backend_api.get_published_tasks() + if len(published) == 0: + update.message.reply_text( + "Пока что не опубликовано ни одной задачи =(", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) + + # okay this is epic (pile of shit) + return ANSWER_RIGHT + else: + update.message.reply_text( + "Какую задачу ты хочешь сдать?", + reply_markup=ReplyKeyboardMarkup(TasksKeyboard.get_keyboard()) + ) + return TASK_CHOOSING @staticmethod @save_state @@ -76,21 +114,35 @@ class States: task_title = update.message.text user_data["chosen_task"] = task_title - status_code, task = backend_api.get_task(task_title) + status_code, tasks_response = backend_api.get_published_tasks() if status_code != 200: - message = "Произошла ошибка в работе квиза. Мы уже работаем над её устранением!" - keyboard = ContinueKeyboard.get_keyboard() + update.message.reply_text( + "Произошла ошибка в работе квиза. Мы уже работаем над её устранением!", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) return MAIN_MENU - else: - message = '\n'.join([ - f"*{task['title']}*", - f"{task['statement']}", - "", - f"_Теги: {task['tags']}_", - ]) - keyboard = TaskChosenKeyboard.get_keyboard() + titles = {task.get("title"): task for task in tasks_response} + + if task_title not in titles.keys(): + update.message.reply_text( + "Такой задачи не найдено, попробуй ввести другое название!", + reply_markup=ReplyKeyboardMarkup(TasksKeyboard.get_keyboard()) + ) + + return TASK_CHOOSING + + # status_code, task = backend_api.get_task(task_title) + task = titles[task_title] + + message = '\n'.join([ + f"*{task['title']}*", + f"{task['statement']}", + "", + f"_Теги: {task['tags']}_", + ]) + keyboard = TaskChosenKeyboard.get_keyboard() update.message.reply_text( message, parse_mode="Markdown", @@ -151,3 +203,88 @@ class States: ) return ANSWER_RIGHT + + +class AdminStates: + @staticmethod + @save_state + def admin_panel(bot: Bot, update: Update, user_data: dict): + status_code, data = backend_api.get_profile(update.message.from_user.id) + if status_code != 200: + update.message.reply_text( + "Не удалось аутентифицировать пользователя. Доступ запрещен.", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) + + return ADMIN_ACCESS_DENIED + + if not data["is_admin"]: + update.message.reply_text( + "Вы не являетесь администратором. Доступ запрещен.", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) + + return ADMIN_ACCESS_DENIED + + update.message.reply_text( + "Выберите действие", + reply_markup=ReplyKeyboardMarkup(AdminKeyboard.get_keyboard()) + ) + + return ADMIN_MENU + + @staticmethod + @save_state + def choose_task_hide(bot: Bot, update: Update, user_data: dict): + update.message.reply_text( + "Выберите задачу, которую хотите скрыть", + reply_markup=ReplyKeyboardMarkup(PublishTasksKeyboard.get_keyboard()) + ) + + return ADMIN_TASK_CHOOSE_HIDE + + @staticmethod + @save_state + def choose_task_publish(bot: Bot, update: Update, user_data: dict): + update.message.reply_text( + "Выберите задачу, которую хотите опубликовать", + reply_markup=ReplyKeyboardMarkup(PublishTasksKeyboard.get_keyboard()) + ) + + return ADMIN_TASK_CHOOSE_PUBLISH + + @staticmethod + @save_state + def hide_task(bot: Bot, update: Update, user_data: dict): + title = update.message.text + status, data = backend_api.hide_task(title) + if status != 200: + update.message.reply_text( + "Не удалось скрыть задачу.", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) + else: + update.message.reply_text( + "Задача была скрыта.", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) + + return ADMIN_TASK_PUBLISHED + + @staticmethod + @save_state + def publish_task(bot: Bot, update: Update, user_data: dict): + title = update.message.text + status, data = backend_api.publish_task(title) + if status != 200: + update.message.reply_text( + "Не удалось опубликовать задачу.", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) + else: + update.message.reply_text( + "Задача была опубликована.", + reply_markup=ReplyKeyboardMarkup(ContinueKeyboard.get_keyboard()) + ) + + return ADMIN_TASK_PUBLISHED @@ -3,5 +3,9 @@ MAIN_MENU, TOP_10, RULES, TASK_CHOOSING, TASK_SHOWN, ANSWERING, ANSWER_RIGHT, ANSWER_WRONG, + + ADMIN_MENU, ADMIN_TASK_CHOOSE_PUBLISH, ADMIN_TASK_CHOOSE_HIDE, + ADMIN_TASK_PUBLISHED, + ADMIN_ACCESS_DENIED, *_ ) = range(100) |