self/ruz-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Hseapp

Browse Source
This commit is contained in:
Egor Matveev
2023-08-13 19:48:28 +00:00
parent 6a5a9a1a75
commit 5ec9bb4a7e
14 changed files with 272 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
helpers/__init__.py
View File

@@ -1,9 +1,5 @@
import datetime
import logging
import zoneinfo
from helpers.models import User
campus_timdelta = {
"Москва": 3,
@@ -12,14 +8,15 @@ campus_timdelta = {
"Пермь": 5
}
def now(user: User):
today = datetime.datetime.now() + datetime.timedelta(hours=campus_timdelta[user.campus])
def now(user: dict):
today = datetime.datetime.now() + datetime.timedelta(hours=campus_timdelta[user.get('campus', 'Москва')])
return today
def get_next_daily_notify_time(user: User) -> datetime.datetime:
def get_next_daily_notify_time(user: dict) -> datetime.datetime:
time_now = now(user)
hours, minutes = map(int, user.daily_notify_time.split(":"))
hours, minutes = map(int, user['daily_notify_time'].split(":"))
next_time = datetime.datetime(
year=time_now.year,
month=time_now.month,
@@ -32,4 +29,4 @@ def get_next_daily_notify_time(user: User) -> datetime.datetime:
if time_now.hour * 60 + time_now.minute > hours * 60 + minutes:
logging.info('go to next day')
next_time = next_time + datetime.timedelta(days=1)
return next_time - datetime.timedelta(hours=campus_timdelta[user.campus])
return next_time - datetime.timedelta(hours=campus_timdelta[user.get('campus', 'Москва')])

2
helpers/alice.py
View File

@@ -3,7 +3,6 @@ from typing import Optional
from daemons.bot import bot
from helpers import now
from helpers.models import UserSchema
from helpers.mongo import mongo
@@ -40,7 +39,6 @@ class Processor:
def get_lesson_for_user(self, hse_id: int):
user = mongo.users_collection.find_one({"hse_id": hse_id})
user = UserSchema().load(user)
t = now(user)
for lesson in mongo.lessons_collection.find({"hse_user_id": hse_id, "begin": {"$gte": t}}).sort([("begin", 1)]):
return lesson

432
helpers/answer.py
View File

@@ -1,351 +1,266 @@
from telebot.types import Message
import datetime
from telebot.types import Message, ReplyKeyboardRemove
from daemons.bot import bot
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, groups_keyboard, \
no_daily_notify, student_or_teacher_keyboard, campus_keyboard, daily_notify_type, notify_type, first_lesson_notify
from helpers.models import UserSchema, User
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.ruz import ruz
class BaseAnswer:
def process(self, message: Message):
class Answer:
user: dict
message: Message
message_text: str
def __init__(self, message: Message):
self.message = message
self.message = message
self.message_text = message.text or message.caption or ""
user = mongo.users_collection.find_one({"chat_id": message.chat.id})
if user is None:
user = User(chat_id=message.chat.id)
mongo.users_collection.insert_one(UserSchema().dump(user))
else:
user = UserSchema().load(user)
attr = getattr(self, "handle_state_" + user.state, None)
if attr is None:
raise NotImplementedError(f"handled state {user.state} but not implemented!")
attr(message, user)
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
}
mongo.users_collection.insert_one(user)
self.user = user
def set_state(self, user: User, state: str):
user.state = state
mongo.users_collection.update_one({"chat_id": user.chat_id}, {"$set": {"state": state}})
def process(self):
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!")
class Answer(BaseAnswer):
def send_message(self, text, reply_markup=None, remove_keyboard=True, **kwargs):
if reply_markup is None and remove_keyboard:
reply_markup = ReplyKeyboardRemove()
bot.send_message(self.user['chat_id'], text, reply_markup=reply_markup, **kwargs)
def handle_state_new(self, message: Message, user: User):
bot.send_message(
message.chat.id,
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(user, "wait_for_campus")
self.set_state("wait_for_campus")
def handle_state_wait_for_campus(self, message: Message, user: User):
if message.text not in ["Москва", "Санкт-Петербург", "Нижний Новгород", "Пермь"]:
bot.send_message(
user.chat_id,
def handle_state_wait_for_campus(self):
if self.message_text not in ["Москва", "Санкт-Петербург", "Нижний Новгород", "Пермь"]:
self.send_message(
"Ты прислал мне что-то непонятное, используй кнопки. Из какого ты кампуса?",
reply_markup=campus_keyboard()
)
return
bot.send_message(
user.chat_id,
"Принято! Ты преподаватель или студент?",
reply_markup=student_or_teacher_keyboard()
)
self.send_message("Принято! А теперь отправь мне свою корпоративную почту.")
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"$set": {"campus": message.text, "state": "wait_for_student_or_teacher"}}
{"chat_id": self.user['chat_id']},
{"$set": {"campus": self.message_text, "state": "wait_for_email"}}
)
def handle_state_wait_for_student_or_teacher(self, message: Message, user: User):
if message.text == "Студент 👨‍🎓":
bot.send_message(user.chat_id, "Принято! Теперь отправь мне свое ФИО.", reply_markup=again_keyboard())
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"$set": {"is_teacher": False, "state": "wait_for_name"}}
)
self.set_state(user, "wait_for_name")
elif message.text == "Преподаватель 👨‍🏫":
bot.send_message(user.chat_id, "Принято! Теперь отправь мне свое ФИО.", reply_markup=again_keyboard())
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"$set": {"is_teacher": True, "state": "wait_for_name"}}
)
self.set_state(user, "wait_for_name")
elif message.text == "Начать заново 🔄":
bot.send_message(
message.chat.id,
"Привет! Я буду помогать тебе выживать в вышке!\nИз какого ты кампуса?",
reply_markup=campus_keyboard()
)
self.set_state(user, "wait_for_campus")
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
else:
bot.send_message(user.chat_id, "Ты отправил мне что-то неправильное, используй кнопки. Ты преподаватель или студент?", reply_markup=student_or_teacher_keyboard())
def handle_state_wait_for_name(self, message: Message, user: User):
if message.text == "Начать заново 🔄":
bot.send_message(
message.chat.id,
"Привет! Я буду помогать тебе выживать в вышке!\nИз какого ты кампуса?",
reply_markup=campus_keyboard()
)
self.set_state(user, "wait_for_campus")
return
user.name = message.text
data = ruz.find_person(user)
if data is None:
bot.send_message(
user.chat_id,
"В РУЗе какая-то поломка, попробуй еще раз позже."
)
return
if len(data) == 0:
bot.send_message(user.chat_id, "К сожалению, в РУЗе не нашлось такого человека, попробуй еще раз.")
return
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"$set": {"name": user.name}})
if user.is_teacher:
bot.send_message(
user.chat_id,
"Отлично! Теперь выбери из списка свой департамент.",
reply_markup=groups_keyboard(data)
)
else:
bot.send_message(
user.chat_id,
"Отлично! Теперь выбери из списка свою группу.",
reply_markup=groups_keyboard(data)
)
self.set_state(user, "wait_for_group")
def handle_state_wait_for_group(self, message: Message, user: User):
if message.text == "Начать заново 🔄":
bot.send_message(
message.chat.id,
"Привет! Я буду помогать тебе выживать в вышке!\nИз какого ты кампуса?",
reply_markup=campus_keyboard()
)
self.set_state(user, "wait_for_campus")
return
group = message.text
data = ruz.find_person(user)
if data is None:
bot.send_message(
user.chat_id,
"В РУЗе какая-то поломка, попробуй еще раз позже."
)
return
for element in data:
if element['description'] == group:
user.hse_id = int(element['id'])
user.group = group
user.name = element['label']
break
if user.group is None:
bot.send_message(user.chat_id, "Ты ввел что-то неправильно, попробуй еще раз сначала. Из какого ты кампуса?", reply_markup=campus_keyboard())
self.set_state(user, "wait_for_campus")
return
mongo.users_collection.update_one({"chat_id": user.chat_id}, {"$set": {
"hse_id": user.hse_id,
"group": group,
"name": user.name
mongo.users_collection.update_one({"chat_id": self.user['chat_id']}, {"$set": {
"email": self.user['email'],
}})
bot.send_message(
user.chat_id,
self.send_message(
"Я нашел тебя в базе РУЗ. Я буду подсказывать тебе расписание, а также уведомлять о предстоящих парах.",
)
success = fetch_schedule_for_user(user)
if success:
lessons = mongo.get_today_lessons(user)
if len(lessons) == 0:
bot.send_message(
user.chat_id,
"Сегодня у тебя нет пар, отдыхай",
reply_markup=main_keyboard()
)
else:
bot.send_message(
user.chat_id,
"Твои пары сегодня:\n" + ruz.schedule_builder(lessons),
reply_markup=main_keyboard(),
parse_mode='Markdown',
)
self.set_state(user, "ready")
lessons = mongo.get_today_lessons(self.user)
if len(lessons) == 0:
self.send_message(
"Сегодня у тебя нет пар, отдыхай",
reply_markup=main_keyboard()
)
else:
self.send_message(
"Твои пары сегодня:\n" + ruz.schedule_builder(lessons),
reply_markup=main_keyboard(),
parse_mode='Markdown',
)
self.set_state("ready")
def handle_state_ready(self, message: Message, user: User):
if message.text == "Пары сегодня":
lessons = mongo.get_today_lessons(user)
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 message.text == "Пары завтра":
lessons = mongo.get_tomorrow_lessons(user)
elif self.message_text == "Пары завтра":
lessons = mongo.get_tomorrow_lessons(self.user)
if len(lessons) == 0:
text = "Завтра у тебя нет пар, отдыхай."
else:
text = ruz.schedule_builder(lessons)
elif message.text == "Расписание на неделю":
lessons = mongo.get_week_lessons(user)
elif self.message_text == "Расписание на неделю":
lessons = mongo.get_week_lessons(self.user)
if len(lessons) == 0:
text = "На этой неделе у тебя нет пар, отдыхай."
else:
text = ruz.schedule_builder(lessons)
elif message.text == "Напоминания о парах":
bot.send_message(
user.chat_id,
elif self.message_text == "Напоминания о парах":
self.send_message(
"Я умею напоминать о каждой паре и о первой паре. Что хочешь настроить?",
reply_markup=notify_type()
)
self.set_state(user, "notify_type")
self.set_state("notify_type")
return
elif message.text == "Ежедневные уведомления":
bot.send_message(
user.chat_id,
elif self.message_text == "Ежедневные уведомления":
self.send_message(
"Я могу присылать тебе расписание на текущий день или на следующий. Как ты хочешь чтобы я тебя уведомлял?",
reply_markup=daily_notify_type()
)
self.set_state(user, "wait_for_daily_notify_type")
self.set_state("wait_for_daily_notify_type")
return
elif message.text == "Сброс настроек":
bot.send_message(user.chat_id, "Ты уверен что хочешь сбросить все настройки и больше не получать уведомления?", reply_markup=yes_no_keyboard())
self.set_state(user, "reset")
elif self.message_text == "Сброс настроек":
self.send_message("Ты уверен что хочешь сбросить все настройки и больше не получать уведомления?", reply_markup=yes_no_keyboard())
self.set_state("reset")
return
elif message.text == "Подключение Алисы":
if user.yandex_id is None:
text = "Для того, чтобы подключить Яндекс.Алису, вызови навык \"Расписание вышки\" и назови этот код: " + str(user.hse_id)
elif self.message_text == "Подключение Алисы":
if self.user['yandex_id'] is None:
text = "Для того, чтобы подключить Яндекс.Алису, вызови навык \"Расписание вышки\" и назови этот код: " + str(self.user['hse_id'])
else:
text = "Янедкс.Алиса уже подключена. Чтобы узнать ближайшую пару, вызови навык \"Расписание вышки\""
else:
text = "Я не понимаю такой команды, используй кнопки."
bot.send_message(
user.chat_id,
self.send_message(
text,
reply_markup=main_keyboard(),
parse_mode='Markdown'
)
def handle_state_notify_type(self, message: Message, user: User):
text = message.text
if text == "О каждой паре":
bot.send_message(
user.chat_id,
def handle_state_notify_type(self):
if self.message_text == "О каждой паре":
self.send_message(
"Выбери за сколько минут мне нужно напомнить тебе о предстоящей паре",
reply_markup=notify_keyboard()
)
self.set_state(user, "wait_for_notify")
elif text == "О первой паре":
bot.send_message(
user.chat_id,
self.set_state("wait_for_notify")
elif self.message_text == "О первой паре":
self.send_message(
"Выбери за сколько минут мне нужно напоминать тебе о первой паре",
reply_markup=first_lesson_notify()
)
self.set_state(user, "wait_for_first_notify")
elif text == "Назад":
self.set_state(user, "ready")
bot.send_message(
user.chat_id,
text,
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()
)
else:
bot.send_message(user.chat_id, "Используй кнопки!", reply_markup=notify_type())
self.send_message("Используй кнопки!", reply_markup=notify_type())
def handle_state_wait_for_first_notify(self, message: Message, user: User):
text = message.text
if text == "30 минут":
def handle_state_wait_for_first_notify(self):
if self.message_text == "30 минут":
time_notify = 30
elif text == "1 час":
elif self.message_text == "1 час":
time_notify = 60
elif text == "4 часа":
elif self.message_text == "4 часа":
time_notify = 4 * 60
elif text == "12 часов":
elif self.message_text == "12 часов":
time_notify = 12 * 60
elif text == "Не уведомлять":
elif self.message_text == "Не уведомлять":
time_notify = None
else:
bot.send_message(user.chat_id, "Используй кнопки!", reply_markup=first_lesson_notify())
self.send_message("Используй кнопки!", reply_markup=first_lesson_notify())
return
mongo.users_collection.update_one({"chat_id": user.chat_id}, {"$set": {"first_lesson_notify": time_notify}})
self.set_state(user, "ready")
bot.send_message(user.chat_id, "Запомнил!", reply_markup=main_keyboard())
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())
def handle_state_wait_for_daily_notify_type(self, message: Message, user: User):
text = message.text
if text == "Текущий день":
bot.send_message(
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": user.chat_id},
{"chat_id": self.user['chat_id']},
{"$set": {"daily_notify_today": True, "state": "wait_for_daily_notify"}}
)
elif text == "Следующий день":
bot.send_message(
user.chat_id,
elif self.message_text == "Следующий день":
self.send_message(
"Каждый день (кроме субботы) я буду присылать тебе расписание на следующий день. Пришли мне в формате чч:мм время, в которое ты хочешь получать расписание. Например, 20:30",
reply_markup=no_daily_notify()
)
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"chat_id": self.user['chat_id']},
{"$set": {"daily_notify_today": False, "state": "wait_for_daily_notify"}}
)
elif text == "Не уведомлять":
bot.send_message(
user.chat_id,
elif self.message_text == "Не уведомлять":
self.send_message(
"Принято! Я не буду уведомлять тебя.",
reply_markup=main_keyboard()
)
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"chat_id": self.user['chat_id']},
{"$set": {"next_daily_notify_time": None, "daily_notify_time": None, "state": "ready"}}
)
elif text == "Назад":
bot.send_message(
user.chat_id,
elif self.message_text == "Назад":
self.send_message(
"Возвращаюсь!",
reply_markup=main_keyboard()
)
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"chat_id": self.user['chat_id']},
{"$set": {"next_daily_notify_time": None, "daily_notify_time": None, "state": "ready"}}
)
else:
bot.send_message(
user.chat_id,
self.send_message(
"Ты прислал мне что-то неправильное, используй кнопки. Как ты хочешь чтобы я тебя уведомлял?",
reply_markup=daily_notify_type()
)
def handle_state_wait_for_notify(self, message: Message, user: User):
text = message.text
if text == "Не уведомлять":
user.notify_minutes = None
elif text == "5 минут":
user.notify_minutes = 5
elif text == "10 минут":
user.notify_minutes = 10
elif text == "15 минут":
user.notify_minutes = 15
elif text == "20 минут":
user.notify_minutes = 20
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:
bot.send_message(
user.chat_id,
self.send_message(
"Я не понимаю такой команды, используй кнопки.",
reply_markup=notify_keyboard()
)
return
mongo.users_collection.update_one({"chat_id": user.chat_id}, {"$set": {"notify_minutes": user.notify_minutes}})
if user.notify_minutes is not None:
text = f"Принято! Буду уведомлять тебя за {text}."
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"Принято! Я не буду уведомлять тебя."
bot.send_message(user.chat_id, text, reply_markup=main_keyboard())
self.set_state(user, "ready")
self.send_message(text, reply_markup=main_keyboard())
self.set_state("ready")
def _validate_time(self, line: str) -> bool:
if len(line) != 5:
@@ -362,53 +277,44 @@ class Answer(BaseAnswer):
return False
return True
def handle_state_wait_for_daily_notify(self, message: Message, user: User):
text = message.text
if text == "Не уведомлять":
bot.send_message(
user.chat_id,
def handle_state_wait_for_daily_notify(self):
if self.message_text == "Не уведомлять":
self.send_message(
"Принято! Я не буду присылать тебе ежедневные уведомления.",
reply_markup=main_keyboard()
)
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"chat_id": self.user['chat_id']},
{"$set": {"daily_notify_time": None, "next_daily_notify_time": None, "state": "ready"}}
)
return
if not self._validate_time(text):
bot.send_message(
user.chat_id,
if not self._validate_time(self.message_text):
self.send_message(
"Ты прислал мне что-то неправильное. Пришли мне в формате чч:мм время, в которое ты хочешь получать расписание. Например, 09:30",
reply_markup=no_daily_notify()
)
return
user.daily_notify_time = text
next_time = get_next_daily_notify_time(user)
bot.send_message(
user.chat_id,
f"Принято! Буду уведомлять тебя каждый день в {text}.",
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()
)
mongo.users_collection.update_one(
{"chat_id": user.chat_id},
{"chat_id": self.user['chat_id']},
{"$set": {
"daily_notify_time": text,
"daily_notify_time": self.message_text,
"next_daily_notify_time": next_time,
"state": "ready"
}}
)
def handle_state_reset(self, message: Message, user: User):
if message.text == "Да":
mongo.users_collection.delete_one({"chat_id": user.chat_id})
bot.send_message(user.chat_id, "Настройки сброшены, ждем твоего возвращения", reply_markup=again_keyboard())
elif message.text == "Нет":
bot.send_message(user.chat_id, "Возращаюсь к прежнему режиму", reply_markup=main_keyboard())
self.set_state(user, "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.set_state("ready")
else:
bot.send_message(user.chat_id,
"Я не понимаю, используй кнопки",
reply_markup=yes_no_keyboard())
answer = Answer()
self.send_message("Я не понимаю, используй кнопки", reply_markup=yes_no_keyboard())

16
helpers/keyboards.py
View File

@@ -24,14 +24,6 @@ def campus_keyboard():
return kb
def student_or_teacher_keyboard():
kb = telebot.types.ReplyKeyboardMarkup(True, False)
kb.row("Преподаватель 👨‍🏫")
kb.row("Студент 👨‍🎓")
kb.row("Начать заново 🔄")
return kb
def notify_keyboard():
kb = telebot.types.ReplyKeyboardMarkup(True, False)
kb.row("Не уведомлять")
@@ -55,14 +47,6 @@ def again_keyboard():
return kb
def groups_keyboard(data):
kb = telebot.types.ReplyKeyboardMarkup(True, False)
for entity in data:
kb.row(entity['description'])
kb.row("Начать заново 🔄")
return kb
def no_daily_notify():
kb = telebot.types.ReplyKeyboardMarkup(True, False)
kb.row("Не уведомлять")

42
helpers/models.py
View File

@@ -1,42 +0,0 @@
import datetime
from dataclasses import dataclass
from typing import Optional
import marshmallow_dataclass
from marshmallow import EXCLUDE
@dataclass
class User:
chat_id: int
name: Optional[str] = None
group: Optional[str] = None
hse_id: Optional[int] = None
state: str = "new"
notify_minutes: Optional[int] = 10
daily_notify_time: Optional[str] = None
next_daily_notify_time: Optional[datetime.datetime] = None
is_teacher: bool = False
campus: str = "Москва"
daily_notify_today: bool = True
first_lesson_notify: Optional[float] = None
yandex_id: Optional[str] = None
class Meta:
unknown = EXCLUDE
@dataclass
class Lesson:
hse_id: int
USchema = marshmallow_dataclass.class_schema(User)
class UserSchema(USchema):
def load(self, entity, *args, **kwargs):
if entity.get('next_daily_notify_time', None) is not None:
entity['next_daily_notify_time'] = entity['next_daily_notify_time'].strftime("%Y-%m-%dT%H:%M:00.000+00:00")
return super().load(entity, *args, **kwargs)

21
helpers/mongo.py
View File

@@ -5,7 +5,6 @@ import pymongo
import settings
from helpers import now
from helpers.models import UserSchema, User
class Mongo:
@@ -26,21 +25,19 @@ class Mongo:
("discipline", 1),
("auditorium", 1),
("begin", 1),
("hse_user_id", 1),
("user_email", 1),
("link", 1)
])
self.lessons_collection.create_index([
("hse_user_id", 1),
("user_email", 1),
("begin", 1)
])
self.lessons_collection.create_index([
("hse_user_id", 1),
("user_email", 1),
("begin", 1),
("notified", 1)
])
def get_user(self, user_id: int) -> User:
return UserSchema().loads(self.users_collection.find_one({"id": user_id}))
def __getitem__(self, item):
return self.database.get_collection(item)
@@ -53,32 +50,32 @@ class Mongo:
def lessons_collection(self):
return self["lessons"]
def _get_lessons_on_date(self, user: User, date: datetime.datetime):
def _get_lessons_on_date(self, user: dict, date: datetime.datetime):
date = datetime.datetime(year=date.year, month=date.month, day=date.day)
next_date = date + datetime.timedelta(days=1)
lessons = []
for lesson in self.lessons_collection.find({
"hse_user_id": user.hse_id,
"user_email": user['email'],
"begin": {"$gte": date, "$lte": next_date}}
):
lessons.append(lesson)
lessons.sort(key=lambda les: les["begin"])
return lessons
def get_today_lessons(self, user: User):
def get_today_lessons(self, user: dict):
return self._get_lessons_on_date(user, now(user))
def get_tomorrow_lessons(self, user: User):
def get_tomorrow_lessons(self, user: dict):
return self._get_lessons_on_date(user, now(user) + datetime.timedelta(days=1))
def get_week_lessons(self, user: User):
def get_week_lessons(self, user: dict):
date = now(user)
date = datetime.datetime(year=date.year, month=date.month, day=date.day)
weekday = date.weekday()
next_date = date + datetime.timedelta(days=(6 - weekday))
lessons = []
for lesson in self.lessons_collection.find({
"hse_user_id": user.hse_id,
"user_email": user['email'],
"begin": {"$gte": date, "$lte": next_date}}
):
lessons.append(lesson)

57
helpers/ruz.py
View File

@@ -4,47 +4,43 @@ import logging
from requests import get
import settings
from helpers import User
fields = [
'discipline',
'building',
'auditorium',
'date',
'beginLesson',
'endLesson',
'lecturer',
'url1'
'date_start',
'date_end',
'lecturer_profiles',
'stream_links'
]
class RUZ:
def find_person(self, user: User) -> dict | None:
if user.is_teacher:
person_type = "person"
else:
person_type = "student"
search_str = settings.RUZ_API + f"search?term={user.name}&type={person_type}"
try:
data = get(search_str)
except:
return None
if data.status_code == 200:
data = data.json()
for index, value in enumerate(data):
data[index]['description'] = value['description'].capitalize()
return data
return None
def reformat_data(self, col):
for data in col:
date_start = datetime.datetime.strptime(data['date_start'], "%Y-%m-%dT%H:%M:%SZ")
date_end = datetime.datetime.strptime(data['date_end'], "%Y-%m-%dT%H:%M:%SZ")
data['date'] = date_start.strftime("%Y.%m.%d")
data['beginLesson'] = date_start.strftime("%H:%M")
data['endLesson'] = date_end.strftime("%H:%M")
try:
data['lecturer'] = data['lecturer_profiles'][0]["full_name"]
except:
data['lecturer'] = 'Неизвестный преподаватель'
try:
data['url1'] = data['stream_links'][0]['link']
except:
data['url1'] = None
del data['date_start']
del data['date_end']
del data['lecturer_profiles']
del data['stream_links']
def get_schedule(self, user: User, begin_date: datetime.datetime, end_date: datetime.datetime):
def get_schedule(self, user: dict, begin_date: datetime.datetime):
start_date_str = begin_date.strftime("%Y.%m.%d")
end_date_str = end_date.strftime("%Y.%m.%d")
if user.is_teacher:
person_type = "person"
else:
person_type = "student"
search_str = settings.RUZ_API + f"schedule/{person_type}/{user.hse_id}?start={start_date_str}&finish={end_date_str}&lng=1"
search_str = settings.RUZ_API + f"v3/ruz/lessons?start={start_date_str}&offset=30&email={user['email']}"
try:
data = get(search_str)
except:
@@ -59,11 +55,12 @@ class RUZ:
return None
formatted_data = [
{
field: element[field]
field: element.get(field)
for field in fields
}
for element in data
]
self.reformat_data(formatted_data)
return formatted_data
def _name_weekday(self, weekday: int) -> str: