Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f3bfe0fa8c | |||
| 14ad9f2b28 |
+47
-7
@@ -8,7 +8,14 @@ from DB import UseDB
|
||||
from get import fetch_timetable_data, get_timetable_for_day
|
||||
|
||||
|
||||
def get_group_id(user_id):
|
||||
def get_group_id(user_id: int) -> int | None:
|
||||
"""
|
||||
Функция для получения id группы по id пользователя
|
||||
|
||||
:param user_id: int - id пользователя
|
||||
|
||||
:return: int | None - либо id группы полученный по id пользователя или ничего
|
||||
"""
|
||||
db = UseDB("users")
|
||||
if db.find_document({"user_id": user_id}):
|
||||
return db.find_document({"user_id": user_id})[0]["group_id"]
|
||||
@@ -16,7 +23,14 @@ def get_group_id(user_id):
|
||||
return None
|
||||
|
||||
|
||||
async def on_notification(msg: Message):
|
||||
async def on_notification(msg: Message) -> None:
|
||||
"""
|
||||
Функция для отправки уведомлений пользователя по времени
|
||||
|
||||
:param msg: Message - объект сообщения от пользователя
|
||||
|
||||
:return: None
|
||||
"""
|
||||
time_on_notification = msg.text.split()[1]
|
||||
pattern = r'^([01]?[0-9]|2[0-3]):([0-5]?[0-9])$'
|
||||
if re.match(pattern, time_on_notification):
|
||||
@@ -34,14 +48,22 @@ async def on_notification(msg: Message):
|
||||
)
|
||||
|
||||
|
||||
async def set_group(msg: Message):
|
||||
async def set_group(msg: Message) -> None:
|
||||
"""
|
||||
Функция для установки группы пользователя
|
||||
|
||||
:param msg: Message - объект сообщения от пользователя
|
||||
|
||||
:return: None
|
||||
"""
|
||||
try:
|
||||
group_id = int(msg.text.split()[1])
|
||||
except ValueError:
|
||||
return await msg.answer(
|
||||
await msg.answer(
|
||||
"Указанная вами группа не найдена. "
|
||||
"Пожалуйста, проверьте корректность номера группы."
|
||||
)
|
||||
return
|
||||
response = fetch_timetable_data(group_id)
|
||||
if "404" in str(response) or "500" in str(response):
|
||||
await msg.answer(
|
||||
@@ -72,18 +94,36 @@ async def set_group(msg: Message):
|
||||
await msg.answer("Я сохранил группу для вас.", reply_markup=keyboard)
|
||||
|
||||
|
||||
async def timetable_for_day(msg: Message, shift: int = 0):
|
||||
async def timetable_for_day(msg: Message, shift: int = 0) -> None:
|
||||
"""
|
||||
Функция для отправки сообщения пользователю со сдвигом на некоторое количество дней
|
||||
|
||||
:param msg: Message - объект сообщения от пользователя
|
||||
:param shift: int - на сколько дней вперёд пытается пользователь получить расписание
|
||||
|
||||
:return: None
|
||||
"""
|
||||
group_id = get_group_id(msg.from_user.id)
|
||||
if group_id is None:
|
||||
return await msg.answer(
|
||||
await msg.answer(
|
||||
"Кажется, вашего пользователя нет в нашей базе данных."
|
||||
"Чтобы начать работу, пожалуйста, укажите вашу группу с "
|
||||
"помощью команды: /set_group номер_группы."
|
||||
)
|
||||
return
|
||||
for text in get_timetable_for_day(group_id, shift):
|
||||
await msg.answer(text)
|
||||
|
||||
|
||||
async def timetable_day(user_id: int, group_id: int, bot: Bot):
|
||||
async def timetable_day(user_id: int, group_id: int, bot: Bot) -> None:
|
||||
"""
|
||||
Функция для отправки сообщения пользователю по уведомлению
|
||||
|
||||
:param user_id: int - id пользователя
|
||||
:param group_id: int - id группы пользователя
|
||||
:param bot: Bot - бот, с помощью которого можно отправлять сообщения
|
||||
|
||||
:return: None
|
||||
"""
|
||||
for text in get_timetable_for_day(group_id, 1):
|
||||
await bot.send_message(chat_id=user_id, text=text)
|
||||
|
||||
@@ -7,6 +7,13 @@ from DB import UseDB
|
||||
|
||||
|
||||
def fetch_timetable_data(group_id: int) -> BeautifulSoup | None:
|
||||
"""
|
||||
Функция для парсинга сайта по id группы пользователя
|
||||
|
||||
:param group_id: int - id группы пользователя
|
||||
|
||||
:return: BeautifulSoup | None - объект парсиного сайта или ничего если сайт не доступен
|
||||
"""
|
||||
base_url = "https://timetable.pallada.sibsau.ru/timetable/group/"
|
||||
response = requests.get(f"{base_url}{group_id}")
|
||||
if response.status_code == 200:
|
||||
@@ -16,6 +23,14 @@ def fetch_timetable_data(group_id: int) -> BeautifulSoup | None:
|
||||
|
||||
|
||||
def extract_week_data(soup: BeautifulSoup, shift: int) -> dict:
|
||||
"""
|
||||
Функция для получения нужной недели из сайта
|
||||
|
||||
:param soup: BeautifulSoup - объект парсиного сайта
|
||||
:param shift: int - на сколько дней вперёд пытается пользователь получить расписание
|
||||
|
||||
:return: dict - готовый текст дня
|
||||
"""
|
||||
current_week_number = int(datetime.datetime.utcnow().isocalendar()[1])
|
||||
current_day_of_week = datetime.datetime.now().weekday()
|
||||
if current_day_of_week + shift > 6:
|
||||
@@ -29,6 +44,13 @@ def extract_week_data(soup: BeautifulSoup, shift: int) -> dict:
|
||||
|
||||
|
||||
def parse_day_data(day_data: BeautifulSoup) -> dict:
|
||||
"""
|
||||
Функция для получения данных из объекта дня
|
||||
|
||||
:param day_data: BeautifulSoup - объект дня для получения данных
|
||||
|
||||
:return: dict - готовый текст дня
|
||||
"""
|
||||
schedule = {}
|
||||
lines = day_data.find_all('div', {"class": "line"})
|
||||
for line in lines:
|
||||
@@ -41,6 +63,13 @@ def parse_day_data(day_data: BeautifulSoup) -> dict:
|
||||
|
||||
|
||||
def format_lesson_details(lesson_text: str) -> str:
|
||||
"""
|
||||
Функция для удобного преобразования урока
|
||||
|
||||
:param lesson_text: str - текст урока
|
||||
|
||||
:return: str - готовый текст урока
|
||||
"""
|
||||
if "1 подгруппа" in lesson_text or "* подгруппа" in lesson_text:
|
||||
lesson_text = f"⭐ {lesson_text} ⭐"
|
||||
elif "2 подгруппа" in lesson_text:
|
||||
@@ -49,6 +78,14 @@ def format_lesson_details(lesson_text: str) -> str:
|
||||
|
||||
|
||||
def get_timetable_for_day(group_id: int, shift: int = 0) -> list[str]:
|
||||
"""
|
||||
Функция для получения всех предметов в виде строк
|
||||
|
||||
:param group_id: int - id группы пользователя
|
||||
:param shift: int - на сколько дней вперёд пытается пользователь получить расписание
|
||||
|
||||
:return: list[str] - готовый объект со строками уроков дня
|
||||
"""
|
||||
try:
|
||||
soup = fetch_timetable_data(group_id)
|
||||
if soup:
|
||||
@@ -65,6 +102,14 @@ def get_timetable_for_day(group_id: int, shift: int = 0) -> list[str]:
|
||||
|
||||
|
||||
def get_data_in_database(group_id: int, shift: int = 0) -> dict:
|
||||
"""
|
||||
Функция для получения данных из базы данных
|
||||
|
||||
:param group_id: int - id группы пользователя
|
||||
:param shift: int - на сколько дней вперёд пытается пользователь получить расписание
|
||||
|
||||
:return: dict - готовый текст дня
|
||||
"""
|
||||
current_week_number = int(datetime.datetime.utcnow().isocalendar()[1])
|
||||
current_day_of_week = datetime.datetime.now().weekday()
|
||||
if current_day_of_week + shift > 6:
|
||||
@@ -83,5 +128,9 @@ def get_data_in_database(group_id: int, shift: int = 0) -> dict:
|
||||
)
|
||||
response = {}
|
||||
for i in response_data:
|
||||
response[i["lesson_time"]] = [i["lesson_name"], i["lesson_author"], i["lesson_nav"]]
|
||||
response[i["lesson_time"]] = [
|
||||
i["lesson_name"],
|
||||
i["lesson_author"],
|
||||
i["lesson_nav"]
|
||||
]
|
||||
return response
|
||||
|
||||
@@ -78,7 +78,14 @@ async def command_the_day_after_tomorrow(msg: Message):
|
||||
await timetable_for_day(msg, 2)
|
||||
|
||||
|
||||
async def send_notification(bot):
|
||||
async def send_notification(bot: Bot) -> None:
|
||||
"""
|
||||
Функция отправляющая уведомления по прошествию времени
|
||||
|
||||
:param bot: Bot - бот, с помощью которого можно отправлять сообщения
|
||||
|
||||
:return: None
|
||||
"""
|
||||
while True:
|
||||
db = UseDB("users")
|
||||
responses = db.find_document({})
|
||||
@@ -90,7 +97,12 @@ async def send_notification(bot):
|
||||
await asyncio.sleep(59)
|
||||
|
||||
|
||||
async def update_data():
|
||||
async def update_data() -> None:
|
||||
"""
|
||||
Функция базу каждый день в полночь
|
||||
|
||||
:return: None
|
||||
"""
|
||||
while True:
|
||||
now = datetime.now()
|
||||
next_midnight = (
|
||||
|
||||
@@ -5,6 +5,13 @@ from get import fetch_timetable_data, parse_day_data
|
||||
|
||||
|
||||
def update_database(group_id: int) -> None:
|
||||
"""
|
||||
Функция обновление расписание в базе
|
||||
|
||||
:param group_id: int - id группы
|
||||
|
||||
:return: None
|
||||
"""
|
||||
day_mass = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday"]
|
||||
db = UseDB("saved_data")
|
||||
soup: BeautifulSoup = fetch_timetable_data(group_id)
|
||||
@@ -58,6 +65,11 @@ def update_database(group_id: int) -> None:
|
||||
|
||||
|
||||
def detect_group_to_update() -> None:
|
||||
"""
|
||||
Функция для перебора всех group_id из базы пользователей
|
||||
|
||||
:return: None
|
||||
"""
|
||||
db = UseDB("users")
|
||||
all_users = db.find_document({})
|
||||
users_ids = []
|
||||
|
||||
Reference in New Issue
Block a user