369 lines
18 KiB
Python
369 lines
18 KiB
Python
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']}
|
||
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())
|