import datetime import logging from random import choice from telebot.types import Message, ReplyKeyboardRemove from daemons.fetch import fetch_schedule_for_user from helpers import get_next_daily_notify_time from helpers.keyboards import main_keyboard, notify_keyboard, yes_no_keyboard, again_keyboard, no_daily_notify, \ campus_keyboard, daily_notify_type, notify_type, first_lesson_notify from helpers.mongo import mongo from helpers.sprint_platform import platform from helpers.ruz import ruz from utils import queues class User: def __init__(self, telegram_id): self.telegram_id = telegram_id @property def platform_staff(self): return platform.is_staff(telegram_id=self.telegram_id) class Answer: user: dict message: Message message_text: str def __init__(self, message: Message): self.message = message self.message_text = message.text or message.caption or "" if self.message_text.startswith('/start'): mongo.users_collection.delete_many({"chat_id": message.chat.id}) user = mongo.users_collection.find_one({"chat_id": message.chat.id}) if user is None: user = { "chat_id": message.chat.id, 'email': None, 'state': 'new', 'notify_minutes': 10, 'daily_notify_time': None, 'next_daily_notify_time': None, 'campus': "Москва", 'daily_notify_today': True, 'first_lesson_notify': None, 'last_schedule_fetch': datetime.datetime.now(), 'yandex_id': None, 'yandex_code': None, } mongo.users_collection.insert_one(user) self.user = user def process(self): user = User(self.user['chat_id']) try: bot_enabled_exp = platform.get_experiment('bot_enabled') if not bot_enabled_exp['enabled'] or not eval(bot_enabled_exp['condition']): return except Exception as exc: logging.info(exc) return getattr( self, "handle_state_" + self.user['state'], self.handle_state_default )() def handle_state_default(self): raise NotImplementedError(f"handled state {self.user['state']} but not implemented!") def send_message(self, text, reply_markup=None, remove_keyboard=True, **kwargs): if reply_markup is None and remove_keyboard: reply_markup = ReplyKeyboardRemove() body = {'text': text, 'chat_id': self.user['chat_id'], 'parse_mode': 'Markdown'} if reply_markup: body['reply_markup'] = reply_markup.to_json() queues.set_task('botalka_mailbox', {'project': 'ruz-bot', 'name': 'telegram-bot', 'body': body}, 1) def set_state(self, state: str): self.user['state'] = state mongo.users_collection.update_one({"chat_id": self.user['chat_id']}, {"$set": {"state": state}}) def handle_state_new(self): self.send_message( "Привет! Я буду помогать тебе выживать в вышке!\nИз какого ты кампуса?", reply_markup=campus_keyboard() ) self.set_state("wait_for_campus") def handle_state_wait_for_campus(self): if self.message_text not in ["Москва", "Санкт-Петербург", "Нижний Новгород", "Пермь"]: self.send_message( "Ты прислал мне что-то непонятное, используй кнопки. Из какого ты кампуса?", reply_markup=campus_keyboard() ) return self.send_message("Принято! А теперь отправь мне свою корпоративную почту.") mongo.users_collection.update_one( {"chat_id": self.user['chat_id']}, {"$set": {"campus": self.message_text, "state": "wait_for_email"}} ) def handle_state_wait_for_email(self): self.user['email'] = self.message_text schedule = fetch_schedule_for_user(self.user) if schedule is None: self.send_message("Возможно, в РУЗе какие-то сбои, либо твой email неправильный. Попробуй ввести email еще раз.") return mongo.users_collection.update_one({"chat_id": self.user['chat_id']}, {"$set": { "email": self.user['email'], }}) self.send_message( "Я нашел тебя в базе РУЗ. Я буду подсказывать тебе расписание, а также уведомлять о предстоящих парах.", ) lessons = mongo.get_today_lessons(self.user) if len(lessons) == 0: self.send_message( "Сегодня у тебя нет пар, отдыхай", reply_markup=main_keyboard(self.user['chat_id']) ) else: self.send_message( "Твои пары сегодня:\n" + ruz.schedule_builder(lessons), reply_markup=main_keyboard(self.user['chat_id']), parse_mode='Markdown', ) self.set_state("ready") def handle_state_ready(self): if self.message_text == "Пары сегодня": lessons = mongo.get_today_lessons(self.user) if len(lessons) == 0: text = "Сегодня у тебя нет пар, отдыхай." else: text = ruz.schedule_builder(lessons) elif self.message_text == "Пары завтра": lessons = mongo.get_tomorrow_lessons(self.user) if len(lessons) == 0: text = "Завтра у тебя нет пар, отдыхай." else: text = ruz.schedule_builder(lessons) elif self.message_text == "Расписание на неделю": lessons = mongo.get_week_lessons(self.user) if len(lessons) == 0: text = "На этой неделе у тебя нет пар, отдыхай." else: text = ruz.schedule_builder(lessons) elif self.message_text == "Напоминания о парах": self.send_message( "Я умею напоминать о каждой паре и о первой паре. Что хочешь настроить?", reply_markup=notify_type() ) self.set_state("notify_type") return elif self.message_text == "Ежедневные уведомления": self.send_message( "Я могу присылать тебе расписание на текущий день или на следующий. Как ты хочешь чтобы я тебя уведомлял?", reply_markup=daily_notify_type() ) self.set_state("wait_for_daily_notify_type") return elif self.message_text == "Сброс настроек": self.send_message("Ты уверен что хочешь сбросить все настройки и больше не получать уведомления?", reply_markup=yes_no_keyboard()) self.set_state("reset") return elif self.message_text == "Подключение Алисы": alice_exp = platform.get_experiment('alice') telegram_id = self.user['chat_id'] try: user = User(telegram_id) alice_exp_enabled = alice_exp['enabled'] and eval(alice_exp['condition']) except Exception as exc: logging.info(exc) alice_exp_enabled = False if not alice_exp_enabled: self.send_message( 'Эта функция еще не работает!', reply_markup=main_keyboard(self.user['chat_id']), parse_mode='Markdown' ) return if self.user.get('yandex_id', None) is None: words = platform.get_config('words') while True: random_words = [choice(words) for _ in range(4)] code = ' '.join(random_words) u = mongo.users_collection.find_one({"yandex_code": code}) if u is None: break mongo.users_collection.update_one({'chat_id': self.user['chat_id']}, {"$set": {"yandex_code": code}}) text = "Для того, чтобы подключить Яндекс.Алису, вызови навык \"Расписание вышки\" и назови этот код: " + '"' + code + '"' else: text = "Янедкс.Алиса уже подключена. Чтобы узнать ближайшую пару, вызови навык \"Расписание вышки\"" else: text = "Я не понимаю такой команды, используй кнопки." self.send_message( text, reply_markup=main_keyboard(self.user['chat_id']), parse_mode='Markdown' ) def handle_state_notify_type(self): if self.message_text == "О каждой паре": self.send_message( "Выбери за сколько минут мне нужно напомнить тебе о предстоящей паре", reply_markup=notify_keyboard() ) self.set_state("wait_for_notify") elif self.message_text == "О первой паре": self.send_message( "Выбери за сколько минут мне нужно напоминать тебе о первой паре", reply_markup=first_lesson_notify() ) self.set_state("wait_for_first_notify") elif self.message_text == "Назад": self.set_state("ready") self.send_message( self.message_text, reply_markup=main_keyboard(self.user['chat_id']) ) else: self.send_message("Используй кнопки!", reply_markup=notify_type()) def handle_state_wait_for_first_notify(self): if self.message_text == "30 минут": time_notify = 30 elif self.message_text == "1 час": time_notify = 60 elif self.message_text == "4 часа": time_notify = 4 * 60 elif self.message_text == "12 часов": time_notify = 12 * 60 elif self.message_text == "Не уведомлять": time_notify = None else: self.send_message("Используй кнопки!", reply_markup=first_lesson_notify()) return mongo.users_collection.update_one({"chat_id": self.user['chat_id']}, {"$set": {"first_lesson_notify": time_notify}}) self.set_state("ready") self.send_message("Запомнил!", reply_markup=main_keyboard(self.user['chat_id'])) def handle_state_wait_for_daily_notify_type(self): if self.message_text == "Текущий день": self.send_message( "Каждый день (кроме воскресенья) я буду присылать тебе расписание на текущий день. Пришли мне в формате чч:мм время, в которое ты хочешь получать расписание. Например, 09:30", reply_markup=no_daily_notify() ) mongo.users_collection.update_one( {"chat_id": self.user['chat_id']}, {"$set": {"daily_notify_today": True, "state": "wait_for_daily_notify"}} ) elif self.message_text == "Следующий день": self.send_message( "Каждый день (кроме субботы) я буду присылать тебе расписание на следующий день. Пришли мне в формате чч:мм время, в которое ты хочешь получать расписание. Например, 20:30", reply_markup=no_daily_notify() ) mongo.users_collection.update_one( {"chat_id": self.user['chat_id']}, {"$set": {"daily_notify_today": False, "state": "wait_for_daily_notify"}} ) elif self.message_text == "Не уведомлять": self.send_message( "Принято! Я не буду уведомлять тебя.", reply_markup=main_keyboard(self.user['chat_id']) ) mongo.users_collection.update_one( {"chat_id": self.user['chat_id']}, {"$set": {"next_daily_notify_time": None, "daily_notify_time": None, "state": "ready"}} ) elif self.message_text == "Назад": self.send_message( "Возвращаюсь!", reply_markup=main_keyboard(self.user['chat_id']) ) mongo.users_collection.update_one( {"chat_id": self.user['chat_id']}, {"$set": {"next_daily_notify_time": None, "daily_notify_time": None, "state": "ready"}} ) else: self.send_message( "Ты прислал мне что-то неправильное, используй кнопки. Как ты хочешь чтобы я тебя уведомлял?", reply_markup=daily_notify_type() ) def handle_state_wait_for_notify(self): if self.message_text == "Не уведомлять": self.user['notify_minutes'] = None elif self.message_text == "5 минут": self.user['notify_minutes'] = 5 elif self.message_text == "10 минут": self.user['notify_minutes'] = 10 elif self.message_text == "15 минут": self.user['notify_minutes'] = 15 elif self.message_text == "20 минут": self.user['notify_minutes'] = 20 else: self.send_message( "Я не понимаю такой команды, используй кнопки.", reply_markup=notify_keyboard() ) return mongo.users_collection.update_one({"chat_id": self.user['chat_id']}, {"$set": {"notify_minutes": self.user['notify_minutes']}}) if self.user['notify_minutes'] is not None: text = f"Принято! Буду уведомлять тебя за {self.message_text}." else: text = f"Принято! Я не буду уведомлять тебя." self.send_message(text, reply_markup=main_keyboard(self.user['chat_id'])) self.set_state("ready") def _validate_time(self, line: str) -> bool: if len(line) != 5: return False if line.count(":") != 1: return False hours, minutes = line.split(":") try: hours = int(hours) minutes = int(minutes) except ValueError: return False if hours < 0 or hours > 23 or minutes < 0 or minutes > 59: return False return True def handle_state_wait_for_daily_notify(self): if self.message_text == "Не уведомлять": self.send_message( "Принято! Я не буду присылать тебе ежедневные уведомления.", reply_markup=main_keyboard(self.user['chat_id']) ) mongo.users_collection.update_one( {"chat_id": self.user['chat_id']}, {"$set": {"daily_notify_time": None, "next_daily_notify_time": None, "state": "ready"}} ) return if not self._validate_time(self.message_text): self.send_message( "Ты прислал мне что-то неправильное. Пришли мне в формате чч:мм время, в которое ты хочешь получать расписание. Например, 09:30", reply_markup=no_daily_notify() ) return self.user['daily_notify_time'] = self.message_text next_time = get_next_daily_notify_time(self.user) self.send_message( f"Принято! Буду уведомлять тебя каждый день в {self.message_text}.", reply_markup=main_keyboard(self.user['chat_id']) ) mongo.users_collection.update_one( {"chat_id": self.user['chat_id']}, {"$set": { "daily_notify_time": self.message_text, "next_daily_notify_time": next_time, "state": "ready" }} ) def handle_state_reset(self): if self.message_text == "Да": mongo.users_collection.delete_one({"email": self.user['email']}) self.send_message("Настройки сброшены, ждем твоего возвращения", reply_markup=again_keyboard()) elif self.message_text == "Нет": self.send_message("Возращаюсь к прежнему режиму", reply_markup=main_keyboard(self.user['chat_id'])) self.set_state("ready") else: self.send_message("Я не понимаю, используй кнопки", reply_markup=yes_no_keyboard())