zatups/bot
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add schedule search and viewer

Browse source
This commit is contained in:
Server 2026-08-30 13:04:57 +00:00
parent f65ad275bc
commit f5febefa92
10 changed files with 559 additions and 226 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
bot/__init__.py
View file

@ -24,6 +24,7 @@ from .analytics import *
from .iternal import *
from .inline import *
from .main.home import *
from .schedule.search import *
from .schedule.other_days import *
from .settings.notifications import *
from .schedule.saved_groups import *
@ -40,4 +41,4 @@ def run():
logger.error('Bot', 'Stopping...')
return
except Exception as e:
logger.error('Bot', f'{e}')
logger.error('Bot', f'{e}')

48
bot/main/home.py
View file

@ -2,7 +2,7 @@ import json
from telebot.types import Message, CallbackQuery, InputFile, ReplyKeyboardRemove
from telebot.util import quick_markup
from models.bot import bot, Markup, send, Receive, Default
from models.bot import bot, Markup, Strings, send, Receive, Default
from models.user import validate
from models.schedule import generateDaySchedule, dayStrings
@ -15,9 +15,13 @@ from telebot.handler_backends import ContinueHandling
def sendBrokenMessage(message: Message):
send(message, 'Твоя группа не была найдена в боте, пожалуйста перезапусти его через команду /start', ReplyKeyboardRemove())
@bot.message_handler(content_types = ['text'], func = lambda m: m.text == 'На главную 🏠')
@bot.message_handler(content_types = ['text'], func = lambda m: m.text == Strings.back_to_home)
def returnToDefaultState(message: Message):
u = validate(message)
if not u.active_target:
from bot.schedule.search import open_search
return open_search(message, u)
u.setState('default')
return send(message, 'Главное меню 🏠', reply_markup = Markup.default(u))
@ -32,11 +36,25 @@ def defaultState(message: Message):
u = validate(message)
if message.text[0] == '/': return ContinueHandling() # type: ignore
if message.text == f'{u.group} 🔄':
u.setState('savedGroups')
return send(u.id, f'<b>Сохраненные группы</b>\nЗдесь список групп, которые ранее были использованы, чтобы быстро между ними переключаться. \n\n<i>Активная группа: {u.group}</i>', Markup.saved_groups(u))
if message.text == Strings.search:
from bot.schedule.search import open_search
schedule_actions = {'Сегодня 🔥', 'Завтра 🍀', 'На неделю 📅', 'Другие дни 📁'}
return open_search(message, u)
if message.text == Strings.current_schedule_name_F.format(u.group):
u.setState('savedGroups')
return send(
u.id,
f'<b>Сохраненные расписания</b>\n\n<i>Активное расписание: {u.group}</i>',
Markup.saved_schedules(u),
)
schedule_actions = {
Strings.today,
Strings.tomorrow,
Strings.schedule_another_days,
'На неделю 📅',
}
s = u.schedule() if message.text in schedule_actions else None
if message.text in schedule_actions and not s:
return send(
@ -46,10 +64,10 @@ def defaultState(message: Message):
)
match message.text:
case 'Сегодня 🔥':
case Strings.today:
return send(u.id, generateDaySchedule(s, isToday = True), reply_markup = Markup.default(u))
case 'Завтра 🍀':
case Strings.tomorrow:
if s.weekDay in [4, 5]:
return send(u.id, f'<b>{dayStrings[s.weekDay+1]}</b>\nЗавтра пар нет 🥰')
elif s.weekDay == 6:
@ -73,11 +91,17 @@ def defaultState(message: Message):
return send(u.id, resultString, reply_markup = Markup.default(u))
case 'Другие дни 📁':
case Strings.schedule_another_days:
u.setState('otherDays')
return send(u.id, f'<b>Сегодня {dayStrings[s.weekDay].lower()}, {"четная" if s.weekNumber%2==0 else "нечетная"} неделя</b>\n\nВыбери нужный день/пункт', reply_markup = Markup.other())
u.view_target = u.active_target
u.save()
return send(
u.id,
f'<b>Сегодня {dayStrings[s.weekDay].lower()}, {"четная" if s.weekNumber%2==0 else "нечетная"} неделя</b>\n\nВыбери нужный день/пункт',
reply_markup=Markup.schedule(u.is_saved(u.active_target)),
)
case 'Уведомления ⏰':
case Strings.notifications:
u.setState('notifications')
return send(u.id, f'''\
<b>Перед парой ⌛</b> — <i>{"включено" if 'lesson' in u.notifications else "выключено"}</i>
@ -92,7 +116,7 @@ def defaultState(message: Message):
<i>Выбери нужную кнопку, чтобы включить/выключить уведомления</i>\
''', reply_markup = Markup.notifications())
case 'Поддержать 💞':
case Strings.support:
supportJSON = json.load(open('support.json', 'r', encoding = 'utf-8'))
return send(u.id, supportJSON['text'], reply_markup = quick_markup(supportJSON['quick_markup'], row_width = 1))

9
bot/main/register_user.py
View file

@ -5,7 +5,7 @@ from telebot.util import quick_markup
from models import logger
from models.bot import Markup_v1 as Markup, bot, send
from models.schedule import ScheduleAPIError, find_main_groups, group_exists
from models.schedule import ScheduleAPIError, find_main_groups, group_exists, group_target
from models.user import getAllUsers, validate
@ -23,7 +23,10 @@ def start(message: Message):
{
"state": "selectGroup",
"group": None,
"active_schedule": None,
"saved_groups": user.saved_groups,
"saved_schedules": user.saved_schedules,
"view_schedule": None,
"notifications": [],
"valid": True,
"user": user.user,
@ -67,9 +70,7 @@ def notFoundHandler(message: Message):
def _finish_registration(message: Message, group_name: str) -> None:
user = validate(message)
user.group = group_name
if group_name not in user.saved_groups:
user.saved_groups.append(group_name)
user.save_schedule(group_target(group_name))
user.valid = True
user.setState("default")
send(

57
bot/schedule/other_days.py
View file

@ -1,27 +1,40 @@
from telebot.types import Message
from models.bot import bot, Markup, send
from models.bot import Markup, Strings, bot, send
from models.user import validate
from models.schedule import generateDaySchedule
@bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'otherDays')
def otherState(message: Message):
u = validate(message)
possibleResponces = ['ПН', 'ВТ', 'СР', 'ЧТ', 'ПТ']
if message.text not in possibleResponces and message.text not in ['Четная 📅', 'Нечетная 📅']: return
result: str = ''
s = u.schedule()
if not s:
return send(message, 'Не удалось получить расписание. Попробуй ещё раз позже.')
if message.text in possibleResponces:
s.weekDay = possibleResponces.index(message.text)
result = generateDaySchedule(s)
else:
s.weekNumber = 2 if message.text == 'Четная 📅' else 1
result = f'<i>{message.text.split(" ")[0]} неделя</i>\n'
for index, _ in enumerate(s.days):
s.weekDay = index
result += generateDaySchedule(s, False) +'\n\n'
from .search import render_schedule_action
return send(u.id, result, reply_markup = Markup.other())
@bot.message_handler(
content_types=["text"], func=lambda message: validate(message).state == "otherDays"
)
def otherState(message: Message):
user = validate(message)
target = user.view_target or user.active_target
if not target:
user.setState("default")
return send(message, "Сначала выбери расписание через поиск", Markup.default(user))
if message.text == Strings.save:
user.save_schedule(target)
return send(
message,
f"Расписание <b>{target['name']}</b> сохранено",
reply_markup=Markup.schedule(True),
)
if message.text == Strings.delete:
user.remove_schedule(target)
return send(
message,
f"Расписание <b>{target['name']}</b> удалено из сохраненных",
reply_markup=Markup.schedule(False),
)
result = render_schedule_action(user, target, message.text or "")
if result:
return send(
message,
result,
reply_markup=Markup.schedule(user.is_saved(target)),
)

77
bot/schedule/saved_groups.py
View file

@ -1,33 +1,50 @@
from telebot.types import Message, CallbackQuery
from telebot.util import quick_markup
from models.bot import bot, Markup, send
from telebot.types import Message
from models.bot import Markup, Strings, bot, send
from models.user import validate
from models.schedule import generateDaySchedule
@bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'savedGroups')
def savedGroups(message: Message):
u = validate(message)
if message.text == 'Добавить группу 🔆':
return send(message, 'Чтобы добавить группу используй команду /start')
elif message.text == 'Удалить группы 🗑':
return send(message, 'Это действие нельзя отменить, точно удалить все сохраненные группы?', quick_markup({
'Да, удалить все': {'callback_data': '$delsaved'}
}))
elif message.text in u.saved_groups:
changed = message.text != u.group
if changed: u.group = message.text
u.setState('default')
send(message, 'Группа изменена ✨' if changed else 'Ничего не поменялось ❌', Markup.default(u))
@bot.callback_query_handler(func=lambda call: '$delsaved' == call.data)
def confirmDeleteAll(call: CallbackQuery):
u = validate(call)
u.setState('default')
try:
u.saved_groups = [u.group]
u.save()
send(u.id, 'Сохраненные группы были удалены 🗑', Markup.default(u))
except:
send(u.id, 'Не удалось удалить группы', Markup.default(u))
@bot.message_handler(
content_types=["text"], func=lambda message: validate(message).state == "savedGroups"
)
def savedGroups(message: Message):
user = validate(message)
if message.text == Strings.add_schedule:
from .search import open_search
return open_search(message, user)
if user.active_target and message.text == Strings.delete_active_schedule_F.format(user.group):
removed_name = user.group
user.remove_schedule(user.active_target)
if not user.active_target:
from .search import open_search
send(message, f'Расписание <b>{removed_name}</b> удалено. Выбери новое расписание.')
return open_search(message, user)
user.setState("savedGroups")
return send(
message,
f'Расписание <b>{removed_name}</b> удалено. Активно: <b>{user.group}</b>',
Markup.saved_schedules(user),
)
if user.active_target and message.text == Strings.delete_other_schedules_F.format(user.group):
user.keep_only_active()
return send(
message,
f'Оставлено только расписание <b>{user.group}</b>',
Markup.saved_schedules(user),
)
selected = user.find_saved(message.text or "")
if selected:
changed = selected != user.active_target
user.active_target = selected
user.setState("default")
return send(
message,
"Расписание изменено ✨" if changed else "Ничего не поменялось",
Markup.default(user),
)

158
bot/schedule/search.py Normal file
View file

@ -0,0 +1,158 @@
from telebot.types import Message
from models import logger
from models.bot import Markup, Strings, bot, send
from models.schedule import ScheduleAPIError, ScheduleTarget, generateDaySchedule, search_targets
from models.user import User, validate
SEARCH_KINDS = {
Strings.groups: "group",
Strings.rooms: "room",
Strings.teachers: "teacher",
}
SEARCH_PROMPTS = {
"group": "Введи название группы",
"room": "Введи номер или название аудитории",
"teacher": "Введи фамилию или имя преподавателя",
}
def open_search(message: Message, user: User | None = None) -> None:
user = user or validate(message)
user.search_kind = None
user.search_results = []
user.view_target = None
user.setState("searchType")
send(
message,
"Выберите, для кого или чего вы хотите найти расписание",
reply_markup=Markup.search_types(),
)
def open_schedule_viewer(message: Message, target: ScheduleTarget) -> None:
user = validate(message)
user.view_target = target
user.setState("scheduleSearch")
send(
message,
f"<b>{target['name']}</b>\nВыбери нужный день или неделю",
reply_markup=Markup.schedule(user.is_saved(target), from_search=True),
)
def render_schedule_action(user: User, target: ScheduleTarget, text: str) -> str | None:
schedule = user.schedule(target)
if not schedule:
return None
action = text.replace("→", "").strip()
if action == Strings.today:
return generateDaySchedule(schedule, isToday=True)
if action == Strings.tomorrow:
if schedule.weekDay == 6:
schedule.weekDay = 0
schedule.weekNumber += 1
else:
schedule.weekDay += 1
return generateDaySchedule(schedule)
if action in Strings.weekdays:
schedule.weekDay = Strings.weekdays.index(action)
return generateDaySchedule(schedule)
if action in (Strings.odd, Strings.even):
schedule.weekNumber = 2 if action == Strings.even else 1
result = f"<i>{action.split()[0]} неделя</i>\n"
for weekday in range(len(schedule.days)):
schedule.weekDay = weekday
result += generateDaySchedule(schedule, False) + "\n\n"
return result.rstrip()
return None
@bot.message_handler(content_types=["text"], func=lambda message: message.text == Strings.back_to_search)
def back_to_search(message: Message):
open_search(message)
@bot.message_handler(content_types=["text"], func=lambda message: validate(message).state == "searchType")
def select_search_type(message: Message):
kind = SEARCH_KINDS.get(message.text)
if not kind:
return
user = validate(message)
user.search_kind = kind
user.search_results = []
user.setState(f"search:{kind}")
send(message, SEARCH_PROMPTS[kind], reply_markup=Markup.search_results([]))
@bot.message_handler(
content_types=["text"],
func=lambda message: str(validate(message).state).startswith("search:"),
)
def search_query(message: Message):
user = validate(message)
selected = next(
(target for target in user.search_results if target["name"] == message.text),
None,
)
if selected:
return open_schedule_viewer(message, selected)
kind = str(user.state).partition(":")[2]
try:
results = search_targets(kind, message.text or "", limit=10) # type: ignore
except ScheduleAPIError as error:
logger.error("Schedule", f"[{user.id}] {error}")
return send(message, "Сервер расписания сейчас недоступен. Попробуй ещё раз позже.")
user.search_results = results
user.save()
if len(results) == 1:
return open_schedule_viewer(message, results[0])
if not results:
return send(
message,
"Ничего похожего не нашлось. Попробуй другой запрос.",
reply_markup=Markup.search_results([]),
)
send(
message,
"Выбери подходящее расписание",
reply_markup=Markup.search_results([target["name"] for target in results]),
)
@bot.message_handler(
content_types=["text"], func=lambda message: validate(message).state == "scheduleSearch"
)
def search_schedule_viewer(message: Message):
user = validate(message)
target = user.view_target
if not target:
return open_search(message, user)
if message.text == Strings.save:
user.save_schedule(target)
return send(
message,
f"Расписание <b>{target['name']}</b> сохранено",
reply_markup=Markup.schedule(True, from_search=True),
)
if message.text == Strings.delete:
user.remove_schedule(target)
return send(
message,
f"Расписание <b>{target['name']}</b> удалено из сохраненных",
reply_markup=Markup.schedule(False, from_search=True),
)
result = render_schedule_action(user, target, message.text or "")
if result:
return send(
message,
result,
reply_markup=Markup.schedule(user.is_saved(target), from_search=True),
)

5
models/analytics.py
View file

@ -14,9 +14,10 @@ class Analytics:
def add(self, user: User) -> None:
self.users += 1
if not user.group:
if not user.active_target or user.active_target["kind"] != "group":
return
self.groups[user.group] = self.groups.get(user.group, 0) + 1
group = user.active_target["name"]
self.groups[group] = self.groups.get(group, 0) + 1
def toDict(self) -> dict[str, Any]:
return {

78
models/bot.py
View file

@ -2,6 +2,7 @@ import json
import random
import telebot
import datetime
from zoneinfo import ZoneInfo
from telebot.types import ReplyKeyboardMarkup, InlineQueryResultArticle, InputTextMessageContent, Message, ReplyKeyboardRemove, WebAppInfo, KeyboardButton
import config
from telebot import apihelper
@ -67,6 +68,12 @@ class Strings:
support = 'Поддержка 💞'
notifications = 'Уведомления ⏰'
current_schedule_name_F = '{} 🔄'
groups = 'Группы'
rooms = 'Аудитории'
teachers = 'Преподаватели'
add_schedule = 'Добавить расписание 🔆'
delete_active_schedule_F = 'Удалить "{}" 🗑'
delete_other_schedules_F = 'Удалить все кроме "{}" 🗑'
# register
register_not_found_T = 'Моего {} нету ⚠'
@ -105,12 +112,16 @@ class Markup_v2:
@staticmethod
def schedule(is_added: bool, from_search: bool = False) -> ReplyKeyboardMarkup:
odd_even = datetime.datetime.now().isocalendar().week%2 == 0 #! ??????
odd_even = datetime.datetime.now(ZoneInfo('Europe/Moscow')).isocalendar().week % 2 == 0
markup = ReplyKeyboardMarkup(resize_keyboard=True)
markup.add(Strings.back_to_home, row_width=1) if not from_search else markup.add(Strings.back_to_search,Strings.back_to_home, row_width=2)
markup.add(Strings.today, Strings.tomorrow, row_width=2)
markup.add(*Strings.weekdays, row_width=5)
markup.add(f'{"" if odd_even else "→"} {Strings.odd}', f'{"→" if odd_even else ""} {Strings.even}', row_width=2)
markup.add(
f'{"" if odd_even else "→ "}{Strings.odd}',
f'{"→ " if odd_even else ""}{Strings.even}',
row_width=2,
)
markup.add(Strings.delete if is_added else Strings.save, row_width=1)
return markup
@ -123,17 +134,6 @@ class Markup_v1:
@staticmethod
def default(user: user.User) -> ReplyKeyboardMarkup:
markup = ReplyKeyboardMarkup(resize_keyboard=True)
# markup.add(
# KeyboardButton(
# text='Карта 🗺️',
# web_app=WebAppInfo(f'{config.WEB_BASE_URL}/map/?v12')
# ), row_width=1
# )
# markup.add('Сегодня 🔥', 'Завтра 🍀', row_width=2)
# markup.add('На неделю 📅', 'Другие дни 📁', row_width=2)
# markup.add('Уведомления ⏰', 'Поддержать 💞', f'{user.group} 🔄', row_width=3)
markup = ReplyKeyboardMarkup(resize_keyboard=True)
markup.add(
KeyboardButton(
@ -148,23 +148,45 @@ class Markup_v1:
markup.add(Strings.notifications, Strings.support, row_width=2)
return markup
@staticmethod
def search_types() -> ReplyKeyboardMarkup:
markup = ReplyKeyboardMarkup(resize_keyboard=True)
markup.add(Strings.back_to_home, row_width=1)
markup.add(Strings.groups, row_width=1)
markup.add(Strings.rooms, row_width=1)
markup.add(Strings.teachers, row_width=1)
return markup
@staticmethod
def other() -> ReplyKeyboardMarkup:
def search_results(results: list[str]) -> ReplyKeyboardMarkup:
markup = ReplyKeyboardMarkup(resize_keyboard=True)
markup.add('На главную 🏠', row_width=1)
markup.add('ПН', 'ВТ', 'СР', 'ЧТ', 'ПТ', row_width=5)
markup.add('Нечетная 📅', 'Четная 📅', row_width=2)
# markup.add('ЧТ', 'ПТ', row_width=2)
markup.add(Strings.back_to_home, row_width=1)
for result in results[:10]:
markup.add(result, row_width=1)
return markup
@staticmethod
def schedule(is_saved: bool, from_search: bool = False) -> ReplyKeyboardMarkup:
even_week = datetime.datetime.now(ZoneInfo('Europe/Moscow')).isocalendar().week % 2 == 0
markup = ReplyKeyboardMarkup(resize_keyboard=True)
markup.add(Strings.back_to_search if from_search else Strings.back_to_home, row_width=1)
if from_search:
markup.add(Strings.delete if is_saved else Strings.save, row_width=1)
markup.add(Strings.today, Strings.tomorrow, row_width=2)
markup.add(*Strings.weekdays, row_width=5)
markup.add(
f'{"" if even_week else "→ "}{Strings.odd}',
f'{"→ " if even_week else ""}{Strings.even}',
row_width=2,
)
if not from_search:
markup.add(Strings.delete if is_saved else Strings.save, row_width=1)
return markup
@staticmethod
def notifications() -> ReplyKeyboardMarkup:
markup = ReplyKeyboardMarkup(resize_keyboard=True)
markup.add('На главную 🏠', row_width=1)
markup.add(Strings.back_to_home, row_width=1)
markup.add('Перед парой ⌛', row_width=1)
markup.add('Вечером 🌆', row_width=1)
markup.add('Файлом на неделю 📅', row_width=1)
@ -172,22 +194,24 @@ class Markup_v1:
return markup
@staticmethod
def saved_groups(u: user.User) -> ReplyKeyboardMarkup:
def saved_schedules(u: user.User) -> ReplyKeyboardMarkup:
markup = ReplyKeyboardMarkup(resize_keyboard=True)
markup.add('На главную 🏠', row_width=1)
markup.add(*u.saved_groups, row_width=4)
markup.add('Добавить группу 🔆', 'Удалить группы 🗑', row_width=2)
markup.add(Strings.back_to_home, row_width=1)
markup.add(*(target['name'] for target in u.saved_schedules), row_width=4)
markup.add(Strings.add_schedule, row_width=1)
if u.active_target:
markup.add(Strings.delete_active_schedule_F.format(u.group), row_width=1)
markup.add(Strings.delete_other_schedules_F.format(u.group), row_width=1)
return markup
saved_groups = saved_schedules
@staticmethod
def selectGroup(groups: list[str]) -> ReplyKeyboardMarkup:
markup = ReplyKeyboardMarkup(resize_keyboard=True)
for group in groups:
markup.add(group, row_width=1)
markup.add('Моей группы нету ⚠', row_width=1)
return markup

178
models/schedule.py
View file

@ -7,7 +7,7 @@ import json
import re
import threading
import time
from typing import Any
from typing import Any, Literal, TypedDict
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen
@ -16,6 +16,15 @@ from zoneinfo import ZoneInfo
import config
TargetKind = Literal["group", "room", "teacher"]
class ScheduleTarget(TypedDict):
kind: TargetKind
id: str
name: str
dayStrings = [
"Понедельник",
"Вторник",
@ -34,6 +43,12 @@ _cache: dict[str, tuple[float, list[dict[str, Any]]]] = {}
_failures: dict[str, tuple[float, str]] = {}
_cache_lock = threading.RLock()
_ENDPOINTS: dict[TargetKind, str] = {
"group": "groups",
"room": "rooms",
"teacher": "teachers",
}
class ScheduleAPIError(RuntimeError):
pass
@ -85,6 +100,25 @@ def clear_schedule_cache() -> None:
_failures.clear()
def target_key(target: ScheduleTarget) -> str:
return f"{target['kind']}:{target['id']}"
def normalize_target(value: Any) -> ScheduleTarget | None:
if not isinstance(value, dict):
return None
kind = value.get("kind")
target_id = value.get("id")
name = value.get("name")
if kind not in _ENDPOINTS or not all(isinstance(item, str) and item for item in (target_id, name)):
return None
return {"kind": kind, "id": target_id, "name": name}
def group_target(name: str) -> ScheduleTarget:
return {"kind": "group", "id": name, "name": name}
def get_main_groups() -> list[dict[str, Any]]:
groups = _load_result("groups")
if not groups:
@ -95,40 +129,52 @@ def get_main_groups() -> list[dict[str, Any]]:
)
def _compact_group_name(value: str) -> str:
def _compact(value: str) -> str:
return re.sub(r"[^0-9a-zа-яё]", "", value.casefold())
def find_main_groups(query: str, limit: int = 8) -> list[str]:
query = query.strip()
if not query:
def _search_records(records: list[dict[str, Any]], query: str, limit: int) -> list[dict[str, Any]]:
compact_query = _compact(query)
if not compact_query:
return []
groups = [str(group["name"]) for group in get_main_groups()]
compact_query = _compact_group_name(query)
exact = [name for name in groups if _compact_group_name(name) == compact_query]
exact = [record for record in records if _compact(str(record.get("name", ""))) == compact_query]
if exact:
return exact[:1]
return exact[:limit]
contains = [name for name in groups if compact_query in _compact_group_name(name)]
def score(record: dict[str, Any]) -> tuple[float, str]:
name = str(record.get("name", ""))
compact_name = _compact(name)
similarity = difflib.SequenceMatcher(None, compact_query, compact_name).ratio()
return similarity, name.casefold()
contains = [
record for record in records if compact_query in _compact(str(record.get("name", "")))
]
if contains:
return contains[:limit]
return sorted(contains, key=score, reverse=True)[:limit]
scored = sorted(
(
difflib.SequenceMatcher(None, compact_query, _compact_group_name(name)).ratio(),
name,
)
for name in groups
)
return [name for score, name in reversed(scored) if score >= 0.45][:limit]
ranked = sorted(records, key=score, reverse=True)
return [record for record in ranked if score(record)[0] >= 0.45][:limit]
def search_targets(kind: TargetKind, query: str, limit: int = 10) -> list[ScheduleTarget]:
records = _load_result(_ENDPOINTS[kind])
if kind == "group":
records = [record for record in records if record.get("parent_group_id") is None]
return [
{"kind": kind, "id": str(record["id"]), "name": str(record["name"])}
for record in _search_records(records, query.strip(), limit)
]
def find_main_groups(query: str, limit: int = 8) -> list[str]:
return [target["name"] for target in search_targets("group", query, limit)]
def group_exists(group_name: str) -> bool:
target = _compact_group_name(group_name)
return any(
_compact_group_name(str(group["name"])) == target for group in get_main_groups()
)
target = _compact(group_name)
return any(_compact(str(group["name"])) == target for group in get_main_groups())
def _minutes_to_time(value: int | None) -> list[int]:
@ -141,7 +187,6 @@ class Lesson:
def __init__(self, data: dict[str, Any], subgroup_names: list[str]) -> None:
self.start = _minutes_to_time(data.get("time_start"))
self.end = _minutes_to_time(data.get("time_end"))
self.strTime = f"{self.start[0]:02d}:{self.start[1]:02d} - {self.end[0]:02d}:{self.end[1]:02d}"
self.name = str(data.get("name") or "Без названия")
self.type = str(data.get("type") or "пара")
self.teacher_ids = [str(value) for value in data.get("teacher_ids", [])]
@ -153,15 +198,11 @@ class Lesson:
def text(self, isToday: bool = False) -> str:
now = datetime.datetime.now(MOSCOW)
lesson_start = now.replace(
hour=self.start[0], minute=self.start[1], second=0, microsecond=0
)
lesson_end = now.replace(
hour=self.end[0], minute=self.end[1], second=0, microsecond=0
)
lesson_start = now.replace(hour=self.start[0], minute=self.start[1], second=0, microsecond=0)
lesson_end = now.replace(hour=self.end[0], minute=self.end[1], second=0, microsecond=0)
lesson_type = html.escape(self.type)
date_text = f"{lesson_type.capitalize()} <i>в {self.start[0]}:{self.start[1]:02d}</i>"
if isToday and lesson_start > now:
minutes = int((lesson_start - now).total_seconds() // 60)
if minutes < 30:
@ -170,8 +211,7 @@ class Lesson:
minutes = int((lesson_end - now).total_seconds() // 60)
date_text = f"<i>Сейчас идет</i> {lesson_type}, закончится через {max(minutes, 1)} мин."
details: list[str] = []
details.extend(html.escape(teacher) for teacher in self.teacher_ids)
details = [html.escape(teacher) for teacher in self.teacher_ids]
details.extend(
f'<a href="https://rasp.pgups.ru/schedule/room?room={quote(room)}">ауд. {html.escape(room)}</a>'
for room in self.room_ids
@ -180,57 +220,42 @@ class Lesson:
details.append(html.escape(", ".join(self.subgroup_names)))
info = ", ".join(details) if details else "Дополнительная информация отсутствует"
return (
f"<blockquote><b>{html.escape(self.name)}</b>\n"
f"{date_text}\n{info}</blockquote>"
)
return f"<blockquote><b>{html.escape(self.name)}</b>\n{date_text}\n{info}</blockquote>"
class Schedule:
def __init__(
self,
lessons: list[dict[str, Any]],
group_name: str,
target: ScheduleTarget,
groups: list[dict[str, Any]],
now: datetime.datetime | None = None,
) -> None:
current = now or datetime.datetime.now(MOSCOW)
self.weekNumber = current.isocalendar().week
self.weekDay = current.weekday()
base_group = next(
(
group
for group in groups
if _compact_group_name(str(group.get("name", "")))
== _compact_group_name(group_name)
and group.get("parent_group_id") is None
),
None,
)
if base_group is None:
raise ValueError(f"Группа {group_name!r} не найдена")
base_id = str(base_group["id"])
child_names = {
str(group["id"]): str(group["name"])
for group in groups
if group.get("parent_group_id") == base_id
}
target_ids = {base_id, *child_names}
self.days: list[list[Lesson]] = [[] for _ in range(5)]
target_ids = {target["id"]}
child_names: dict[str, str] = {}
field = {"group": "group_ids", "room": "room_ids", "teacher": "teacher_ids"}[target["kind"]]
if target["kind"] == "group":
child_names = {
str(group["id"]): str(group["name"])
for group in groups
if group.get("parent_group_id") == target["id"]
}
target_ids.update(child_names)
for raw_lesson in lessons:
weekday = raw_lesson.get("weekday")
if not isinstance(weekday, int) or weekday not in range(1, 6):
continue
matched_ids = target_ids.intersection(map(str, raw_lesson.get("group_ids", [])))
matched_ids = target_ids.intersection(map(str, raw_lesson.get(field, [])))
if not matched_ids:
continue
subgroup_names = [
child_names[group_id] for group_id in matched_ids if group_id in child_names
]
self.days[weekday - 1].append(Lesson(raw_lesson, sorted(subgroup_names)))
subgroup_names = sorted(child_names[item] for item in matched_ids if item in child_names)
self.days[weekday - 1].append(Lesson(raw_lesson, subgroup_names))
for day in self.days:
day.sort(key=lambda lesson: (lesson.start, lesson.name.casefold()))
@ -246,12 +271,24 @@ class Schedule:
]
def get_group_schedule(group_name: str) -> Schedule:
def get_schedule(target: ScheduleTarget) -> Schedule:
groups = _load_result("groups")
if not groups:
raise ScheduleAPIError("Сервер расписания пока не содержит групп")
lessons = _load_result("schedule/default")
return Schedule(lessons, group_name, groups)
return Schedule(lessons, target, groups)
def get_group_schedule(group_name: str) -> Schedule:
target = next(
(
{"kind": "group", "id": str(group["id"]), "name": str(group["name"])}
for group in get_main_groups()
if _compact(str(group["name"])) == _compact(group_name)
),
None,
)
if target is None:
raise ValueError(f"Группа {group_name!r} не найдена")
return get_schedule(target)
def generateDaySchedule(
@ -259,11 +296,10 @@ def generateDaySchedule(
) -> str:
even_week = schedule.weekNumber % 2 == 0
weekday = schedule.weekDay
day_name = dayStrings[weekday]
header = (
f'<b>{day_name}, {"четная" if even_week else "нечетная"} неделя</b>\n'
f'<b>{dayStrings[weekday]}, {"четная" if even_week else "нечетная"} неделя</b>\n'
if showOddWeekText
else f"<b>{day_name}</b>\n"
else f"<b>{dayStrings[weekday]}</b>\n"
)
lessons = schedule.today()
if not lessons:

172
models/user.py
View file

@ -6,63 +6,116 @@ from telebot.types import CallbackQuery, InlineQuery, Message
from models import logger
from .schedule import Schedule, ScheduleAPIError, get_group_schedule
from .schedule import (
Schedule,
ScheduleAPIError,
ScheduleTarget,
get_schedule,
group_target,
normalize_target,
target_key,
)
class User:
def __init__(self, userID: int) -> None:
self.id: int = userID
self.id = userID
try:
self.data: dict[str, Any] = json.load(
open(f"./users/{self.id}.json", "r", encoding="utf-8")
)
with open(f"./users/{self.id}.json", "r", encoding="utf-8") as file:
self.data: dict[str, Any] = json.load(file)
except FileNotFoundError:
self.data = {}
except (json.JSONDecodeError, OSError):
logger.error("Bot", f'INVALID "{userID}" SAVED DATA! RESET!')
self.data = {}
self.state: str | bool | None = self.data.get("state", None)
self.state: str | bool | None = self.data.get("state")
legacy_group = self.data.get("group")
self.active_target = normalize_target(self.data.get("active_schedule"))
if self.active_target is None and isinstance(legacy_group, str) and legacy_group:
self.active_target = group_target(legacy_group)
self.group: str | None = self.data.get("group", None)
self.notifications: list[str] = self.data.get("notifications", []) # type: ignore
self.valid: bool = self.data.get("valid", True)
saved_groups = self.data.get("saved_groups", [])
self.saved_groups = [
value for value in saved_groups if isinstance(value, str) and value.strip()
]
if self.group and self.group not in self.saved_groups:
self.saved_groups.append(self.group)
self.saved_groups = list(dict.fromkeys(self.saved_groups))
saved = [normalize_target(value) for value in self.data.get("saved_schedules", [])]
self.saved_schedules: list[ScheduleTarget] = [value for value in saved if value]
for name in self.data.get("saved_groups", []):
if isinstance(name, str) and name:
self._append_saved(group_target(name))
if self.active_target:
self._append_saved(self.active_target)
self.user: dict = self.data.get("user", {})
self.usage: dict = self.data.get(
self.view_target = normalize_target(self.data.get("view_schedule"))
self.search_kind = self.data.get("search_kind")
results = [normalize_target(value) for value in self.data.get("search_results", [])]
self.search_results: list[ScheduleTarget] = [value for value in results if value]
self.notifications: list[str] = self.data.get("notifications", [])
self.valid = bool(self.data.get("valid", True))
self.user: dict[str, Any] = self.data.get("user", {})
self.usage: dict[str, Any] = self.data.get(
"usage", {"count": -1, "date": None, "time": None}
)
self.flags: list[str] = self.data.get('flags', [])
if "notifyEvening" in self.data.keys():
if self.data["notifyEvening"] and "evening" not in self.notifications:
self.notifications.append("evening")
if self.data.get("notifyBeforeLesson") and "lesson" not in self.notifications:
self.notifications.append("lesson")
self.flags: list[str] = self.data.get("flags", [])
if self.data.get("notifyEvening") and "evening" not in self.notifications:
self.notifications.append("evening")
if self.data.get("notifyBeforeLesson") and "lesson" not in self.notifications:
self.notifications.append("lesson")
self.save()
def schedule(self) -> Schedule | None:
if not self.group:
self.valid = False
self.notifications = []
self.setState("broken")
@property
def group(self) -> str | None:
return self.active_target["name"] if self.active_target else None
@group.setter
def group(self, name: str | None) -> None:
self.active_target = group_target(name) if name else None
@property
def saved_groups(self) -> list[str]:
return [target["name"] for target in self.saved_schedules if target["kind"] == "group"]
@saved_groups.setter
def saved_groups(self, names: list[str]) -> None:
non_groups = [target for target in self.saved_schedules if target["kind"] != "group"]
self.saved_schedules = non_groups + [group_target(name) for name in names if name]
def _append_saved(self, target: ScheduleTarget) -> None:
if not any(target_key(value) == target_key(target) for value in self.saved_schedules):
self.saved_schedules.append(target)
def is_saved(self, target: ScheduleTarget | None) -> bool:
return bool(
target
and any(target_key(value) == target_key(target) for value in self.saved_schedules)
)
def save_schedule(self, target: ScheduleTarget, make_active: bool = True) -> None:
self._append_saved(target)
if make_active:
self.active_target = target
self.save()
def remove_schedule(self, target: ScheduleTarget) -> None:
key = target_key(target)
self.saved_schedules = [value for value in self.saved_schedules if target_key(value) != key]
if self.active_target and target_key(self.active_target) == key:
self.active_target = self.saved_schedules[-1] if self.saved_schedules else None
self.save()
def keep_only_active(self) -> None:
self.saved_schedules = [self.active_target] if self.active_target else []
self.save()
def find_saved(self, name: str) -> ScheduleTarget | None:
return next((target for target in self.saved_schedules if target["name"] == name), None)
def schedule(self, target: ScheduleTarget | None = None) -> Schedule | None:
selected = target or self.active_target
if not selected:
return None
try:
return get_group_schedule(self.group)
except ValueError:
self.valid = False
self.notifications = []
self.setState("broken")
return None
except ScheduleAPIError as error:
return get_schedule(selected)
except (ValueError, ScheduleAPIError) as error:
logger.error("Schedule", f"[{self.id}] {error}")
return None
@ -71,34 +124,39 @@ class User:
self.save()
def save(self) -> None:
return json.dump(
{
"state": self.state,
"group": self.group,
"notifications": self.notifications,
"valid": self.valid,
"saved_groups": self.saved_groups,
"user": self.user,
"usage": self.usage,
"flags": self.flags,
},
open(f"./users/{self.id}.json", "w", encoding="utf-8"),
ensure_ascii=False,
indent=4,
)
with open(f"./users/{self.id}.json", "w", encoding="utf-8") as file:
json.dump(
{
"state": self.state,
"group": self.group,
"active_schedule": self.active_target,
"saved_groups": self.saved_groups,
"saved_schedules": self.saved_schedules,
"view_schedule": self.view_target,
"search_kind": self.search_kind,
"search_results": self.search_results,
"notifications": self.notifications,
"valid": self.valid,
"user": self.user,
"usage": self.usage,
"flags": self.flags,
},
file,
ensure_ascii=False,
indent=4,
)
def validate(ctx: Message | InlineQuery | CallbackQuery) -> User:
try:
return User(ctx.from_user.id) # type: ignore
except Exception as e:
logger.error("Bot", f'INVALID "{ctx.from_user.id}" VALIDATION! {e}')
except Exception as error:
logger.error("Bot", f'INVALID "{ctx.from_user.id}" VALIDATION! {error}')
return False # type: ignore
def getAllUsers() -> Iterator[User]:
for _, _, filenames in os.walk("./users"):
for userFile in filenames:
if not userFile.endswith(".json"):
continue
yield User(int(userFile.removesuffix(".json")))
if userFile.endswith(".json"):
yield User(int(userFile.removesuffix(".json")))