diff --git a/bot/__init__.py b/bot/__init__.py index 8284f8f..20e5c45 100644 --- a/bot/__init__.py +++ b/bot/__init__.py @@ -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}') \ No newline at end of file + logger.error('Bot', f'{e}') diff --git a/bot/main/home.py b/bot/main/home.py index e4e3ebd..59cd737 100644 --- a/bot/main/home.py +++ b/bot/main/home.py @@ -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'Сохраненные группы\nЗдесь список групп, которые ранее были использованы, чтобы быстро между ними переключаться. \n\nАктивная группа: {u.group}', 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'Сохраненные расписания\n\nАктивное расписание: {u.group}', + 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'{dayStrings[s.weekDay+1]}\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'Сегодня {dayStrings[s.weekDay].lower()}, {"четная" if s.weekNumber%2==0 else "нечетная"} неделя\n\nВыбери нужный день/пункт', reply_markup = Markup.other()) + u.view_target = u.active_target + u.save() + return send( + u.id, + f'Сегодня {dayStrings[s.weekDay].lower()}, {"четная" if s.weekNumber%2==0 else "нечетная"} неделя\n\nВыбери нужный день/пункт', + reply_markup=Markup.schedule(u.is_saved(u.active_target)), + ) - case 'Уведомления ⏰': + case Strings.notifications: u.setState('notifications') return send(u.id, f'''\ Перед парой ⌛ — {"включено" if 'lesson' in u.notifications else "выключено"} @@ -92,7 +116,7 @@ def defaultState(message: Message): Выбери нужную кнопку, чтобы включить/выключить уведомления\ ''', 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)) diff --git a/bot/main/register_user.py b/bot/main/register_user.py index 992b2e0..1edd246 100644 --- a/bot/main/register_user.py +++ b/bot/main/register_user.py @@ -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( diff --git a/bot/schedule/other_days.py b/bot/schedule/other_days.py index fee0514..104cdce 100644 --- a/bot/schedule/other_days.py +++ b/bot/schedule/other_days.py @@ -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'{message.text.split(" ")[0]} неделя\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"Расписание {target['name']} сохранено", + reply_markup=Markup.schedule(True), + ) + if message.text == Strings.delete: + user.remove_schedule(target) + return send( + message, + f"Расписание {target['name']} удалено из сохраненных", + 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)), + ) diff --git a/bot/schedule/saved_groups.py b/bot/schedule/saved_groups.py index 3074fd0..3a81af2 100644 --- a/bot/schedule/saved_groups.py +++ b/bot/schedule/saved_groups.py @@ -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)) \ No newline at end of file +@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'Расписание {removed_name} удалено. Выбери новое расписание.') + return open_search(message, user) + user.setState("savedGroups") + return send( + message, + f'Расписание {removed_name} удалено. Активно: {user.group}', + 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'Оставлено только расписание {user.group}', + 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), + ) diff --git a/bot/schedule/search.py b/bot/schedule/search.py new file mode 100644 index 0000000..7724090 --- /dev/null +++ b/bot/schedule/search.py @@ -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"{target['name']}\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"{action.split()[0]} неделя\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"Расписание {target['name']} сохранено", + reply_markup=Markup.schedule(True, from_search=True), + ) + if message.text == Strings.delete: + user.remove_schedule(target) + return send( + message, + f"Расписание {target['name']} удалено из сохраненных", + 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), + ) diff --git a/models/analytics.py b/models/analytics.py index 35bb3f3..136f360 100644 --- a/models/analytics.py +++ b/models/analytics.py @@ -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 { diff --git a/models/bot.py b/models/bot.py index d3954f5..4dc7763 100644 --- a/models/bot.py +++ b/models/bot.py @@ -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 diff --git a/models/schedule.py b/models/schedule.py index 2916b8c..761b349 100644 --- a/models/schedule.py +++ b/models/schedule.py @@ -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()} в {self.start[0]}:{self.start[1]:02d}" + 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"Сейчас идет {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'ауд. {html.escape(room)}' 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"
{html.escape(self.name)}\n" - f"{date_text}\n{info}
" - ) + return f"
{html.escape(self.name)}\n{date_text}\n{info}
" 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'{day_name}, {"четная" if even_week else "нечетная"} неделя\n' + f'{dayStrings[weekday]}, {"четная" if even_week else "нечетная"} неделя\n' if showOddWeekText - else f"{day_name}\n" + else f"{dayStrings[weekday]}\n" ) lessons = schedule.today() if not lessons: diff --git a/models/user.py b/models/user.py index 9209798..3ca1567 100644 --- a/models/user.py +++ b/models/user.py @@ -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")))