From c75f14e58814a82f774a256ccca69a326da59bc3 Mon Sep 17 00:00:00 2001 From: Server <10.9.8.204@example.com> Date: Sun, 30 Aug 2026 11:43:38 +0000 Subject: [PATCH] migrate bot to schedule API --- bot/inline.py | 176 ++++++----------- bot/main/home.py | 19 +- bot/main/register_user.py | 179 +++++++++-------- bot/schedule/other_days.py | 4 +- config.example.py | 4 +- event/lesson.py | 13 +- models/analytics.py | 121 +++--------- models/bot.py | 16 +- models/schedule.py | 385 +++++++++++++++++++++++++------------ models/user.py | 65 +++---- utils/calendar.py | 9 +- 11 files changed, 498 insertions(+), 493 deletions(-) diff --git a/bot/inline.py b/bot/inline.py index e2bef15..00a5ebc 100644 --- a/bot/inline.py +++ b/bot/inline.py @@ -1,138 +1,74 @@ -from telebot.types import (InlineQueryResultArticle, InputTextMessageContent, InlineQuery) - -from models.bot import bot, Inlines -from models.user import validate -from models.schedule import generateDaySchedule, dayStrings +from telebot.types import InlineQuery from models import logger +from models.bot import Inlines, bot +from models.schedule import dayStrings, generateDaySchedule +from models.user import validate + @bot.inline_handler(lambda query: True) def default_query(query: InlineQuery): user = validate(query) inline = Inlines(query.id) - if not user: - logger.warn('Bot', 'Inlinue from unknown user invoked') + if not user.group: inline.add( - 'Расписание недоступно ❌', - 'Запусти бота, чтобы использовать эту функцию', - 'Не удалось получить расписание. Пользователь не запустил бота ❌ \n\nhttps://t.me/zatups_bot' + "Расписание недоступно ❌", + "Запусти бота, чтобы выбрать группу", + "Не удалось получить расписание. Сначала запусти бота: https://t.me/zatups_bot", ) - - - else: - if 'today': - s = user.schedule() - weektype = 'четная' if s.weekNumber%2 == 0 else 'нечетная' - description = dayStrings[s.weekDay] - output = generateDaySchedule(s, isToday = True) if s.weekDay < 5 else f'{description}\nСегодня пар нету 🥰' - + else: + schedule = user.schedule() + if not schedule: inline.add( - 'Расписание на сегодня 🔥', - f'{description}, {weektype.lower()} неделя', - output + "Расписание недоступно ❌", + "Сервер расписания сейчас недоступен", + "Не удалось получить расписание. Попробуй ещё раз позже.", ) - - if 'tomorrow': - s = user.schedule() - output = False - description = dayStrings[(s.weekDay+1)%len(dayStrings)] - if s.weekDay in [4, 5]: - output = f'{description}\nЗавтра пар нету 🥰' - - elif s.weekDay == 6: - s.weekDay = 0 - s.weekNumber += 1 - - else: - s.weekDay += 1 - - if not output: output = generateDaySchedule(s) - - weektype = 'четная' if s.weekNumber%2 == 0 else 'нечетная' - + else: + week_type = "четная" if schedule.weekNumber % 2 == 0 else "нечетная" + description = dayStrings[schedule.weekDay] inline.add( - 'Расписание на завтра 🍀', - f'{description}, {weektype.lower()} неделя', - output + "Расписание на сегодня 🔥", + f"{description}, {week_type} неделя", + generateDaySchedule(schedule, isToday=True), ) - if 'week': - s = user.schedule() - description = f'Сейчас {"четная" if s.weekNumber%2==0 else "нечетная"} неделя' - output = f'Расписание на эту {"четную" if s.weekNumber%2==0 else "нечетную"} неделю\n' + tomorrow = user.schedule() + if tomorrow: + if tomorrow.weekDay == 6: + tomorrow.weekDay = 0 + tomorrow.weekNumber += 1 + else: + tomorrow.weekDay += 1 + tomorrow_description = dayStrings[tomorrow.weekDay] + tomorrow_week_type = ( + "четная" if tomorrow.weekNumber % 2 == 0 else "нечетная" + ) + inline.add( + "Расписание на завтра 🍀", + f"{tomorrow_description}, {tomorrow_week_type} неделя", + generateDaySchedule(tomorrow), + ) - if s.weekDay == 6: - s.weekDay = 0 - s.weekNumber += 1 - description = f'Следующая {"четная" if s.weekNumber%2==0 else "нечетная"} неделя' - output = f'Расписание на следующую {"четную" if s.weekNumber%2==0 else "нечетную"} неделю\n' + week = user.schedule() + if week: + title = "эту" + if week.weekDay == 6: + week.weekNumber += 1 + title = "следующую" + week_type = "четную" if week.weekNumber % 2 == 0 else "нечетную" + output = f"Расписание на {title} {week_type} неделю\n" + for weekday in range(len(week.days)): + week.weekDay = weekday + output += generateDaySchedule(week, False) + "\n\n" + inline.add( + "Расписание на неделю 📅", + f"{week_type.capitalize()} неделя", + output, + ) - for index, _ in enumerate(s.days): - s.weekDay = index - output += generateDaySchedule(s, False) +'\n\n' - - inline.add( - 'Расписание на неделю 📅', - description, - output - ) - - logger.log('Bot', f'Inline from {user.id}') + logger.log("Bot", f"Inline from {user.id}") try: return inline.send() - except Exception as e: - logger.error('Bot', f'Inline.send() -> {e}') - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + except Exception as error: + logger.error("Bot", f"Inline.send() -> {error}") diff --git a/bot/main/home.py b/bot/main/home.py index a28ef8b..e4e3ebd 100644 --- a/bot/main/home.py +++ b/bot/main/home.py @@ -30,21 +30,23 @@ def forceReturnToDefaultState(query: CallbackQuery): @bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'default') def defaultState(message: Message): u = validate(message) - s = u.schedule() - # print(u.id, message.text) - - 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)) + schedule_actions = {'Сегодня 🔥', 'Завтра 🍀', 'На неделю 📅', 'Другие дни 📁'} + s = u.schedule() if message.text in schedule_actions else None + if message.text in schedule_actions and not s: + return send( + u.id, + 'Не удалось получить расписание. Попробуй ещё раз позже или перезапусти бота через /start.', + reply_markup=Markup.default(u), + ) + match message.text: case 'Сегодня 🔥': - if s.weekDay > 4: - return send(u.id, f'{dayStrings[s.weekDay]}\nСегодня пар нет 🥰') - return send(u.id, generateDaySchedule(s, isToday = True), reply_markup = Markup.default(u)) case 'Завтра 🍀': @@ -73,7 +75,7 @@ def defaultState(message: Message): case 'Другие дни 📁': u.setState('otherDays') - return send(u.id, f'Сегодня {dayStrings[s.weekDay].lower()}, {"четная" if s.weekNumber%2==0 else "нечентная"} неделя\n\nВыбери нужный день/пункт', reply_markup = Markup.other()) + return send(u.id, f'Сегодня {dayStrings[s.weekDay].lower()}, {"четная" if s.weekNumber%2==0 else "нечетная"} неделя\n\nВыбери нужный день/пункт', reply_markup = Markup.other()) case 'Уведомления ⏰': u.setState('notifications') @@ -170,4 +172,3 @@ def sendHelp(message: Message): [bot.forward_message(adminID, message.from_user.id, message.id) for adminID in ADMINS] # type: ignore send(message, 'Сообщение переслано разработчику 🔧\nСпасибо за обращение', Markup.default(u)) - diff --git a/bot/main/register_user.py b/bot/main/register_user.py index c78e339..992b2e0 100644 --- a/bot/main/register_user.py +++ b/bot/main/register_user.py @@ -1,97 +1,114 @@ import json -from telebot.types import Message + +from telebot.types import Message, ReplyKeyboardRemove from telebot.util import quick_markup from models import logger -from models.user import User, validate, getAllUsers -from models.bot import (Markup_v1 as Markup, bot, send) -from models.schedule import generateDaySchedule +from models.bot import Markup_v1 as Markup, bot, send +from models.schedule import ScheduleAPIError, find_main_groups, group_exists +from models.user import getAllUsers, validate -@bot.message_handler(commands = ['start']) + +@bot.message_handler(commands=["start"]) def start(message: Message): - u = validate(message) - if u and 'migrate' in message.text: # type: ignore - u.setState('default') - logger.log('Bot', f'{message.from_user.id} migrated') # type: ignore - send(message, 'Добро пожаловать снова 💞', reply_markup = Markup.default(u)) - send(message, f'''Что нового -
1. Теперь доступны все факультеты, курсы и группы -2. Затупсика можно упомянуть в любом чате, даже в личке и группе, и он отправит актуальное расписание! -Для этого напиши @zatups_bot в любом чате -3. Теперь расписания отображаются корректно всегда, даже с подгруппами
+ user = validate(message) + if user.group and "migrate" in (message.text or ""): + user.setState("default") + logger.log("Bot", f"{message.from_user.id} migrated") + return send(message, "Добро пожаловать снова 💞", reply_markup=Markup.default(user)) -Чтобы воспользоваться функцией отправки расписания в любом чате, напиши @zatups_bot в любом чате, выбери бота из списка, а после нажми на нужный день - сегодня, завтра, или на всю неделю.\nКстати - в следующий раз достаточно просто написать @ и телеграм сам выдаст затупсика в списке!''') - return - - logger.log('Bot', f'[{u.id}] /start detected, Total users: {len(list(getAllUsers()))+1}') - json.dump( - { - 'state': 'selectTopic', + was_registered = bool(user.group) + logger.log("Bot", f"[{user.id}] /start detected, Total users: {len(list(getAllUsers()))}") + json.dump( + { + "state": "selectGroup", + "group": None, + "saved_groups": user.saved_groups, + "notifications": [], + "valid": True, + "user": user.user, + "usage": user.usage, + "flags": user.flags, + }, + open(f"./users/{message.from_user.id}.json", "w", encoding="utf-8"), + ensure_ascii=False, + indent=4, + ) - 'topic': '', - 'course': '', - 'group': '', - 'saved_groups': u.saved_groups if u else [], - 'notifyEvening': False, - 'notifyBeforeLesson': False, - }, - open(f'./users/{message.from_user.id}.json', 'w', encoding='utf-8'), # type: ignore - ensure_ascii = False, - indent = 4 - ) - send(message, ''' -Бот на стадии разработки! -- Доступны не все функции, группы -- Возможны баги, фризы, затупы итд итп -- Бот изначально не планировался для публичного использования - -Просто имей это ввиду😥 -''') - - #EDIT - send(message, 'Выбери свой факультет' if not u else 'Вы начали процедуру перезапуска бота. Все настройки и данные удалены, включая уведомления\nВыбери новый факультет', reply_markup = Markup.selectTopic()) + prefix = ( + "Вы начали процедуру перезапуска бота. Уведомления отключены.\n\n" + if was_registered + else "" + ) + send( + message, + prefix + "Напиши название своей группы. Например: АТ-501", + reply_markup=ReplyKeyboardRemove(), + ) - -@bot.message_handler(content_types = ['text'], func = lambda m: ' нету ⚠' in m.text) +@bot.message_handler(content_types=["text"], func=lambda message: "нету ⚠" in message.text) def notFoundHandler(message: Message): - bot.reply_to(message, ''' + bot.reply_to( + message, + """ Не расстраивайся! -Напиши разработчику, чтобы он добавил твой факультет/курс/группу - -
-К сожалению, для поддержки работы бота нужно много мощностей и материальные вложения. -Бюджет, выделенный на разработку бота - 2 банки пива, поэтому в целях экономии в боте пока доступны не все группы -
''', reply_markup = quick_markup({'Написать разработчику 💞': {'url': 'https://t.me/m/Gg00FvJmZDM6'}, 'Перезапустить бота 🤖': {'url': 'https://t.me/zatups_bot?start=r'}}, row_width = 1)) +Напиши разработчику, если твоей группы нет в расписании. +""", + reply_markup=quick_markup( + { + "Написать разработчику 💞": {"url": "https://t.me/m/Gg00FvJmZDM6"}, + "Перезапустить бота 🤖": {"url": "https://t.me/zatups_bot?start=r"}, + }, + row_width=1, + ), + ) -@bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'selectTopic') -def selectTopicState(message: Message): - u = validate(message) - u.topic = message.text - u.setState('selectCourse') - - send(message, 'Выбери свой курс', reply_markup = Markup.selectCourse(u.topic)) - -@bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'selectCourse') -def selectCourseState(message: Message): - u = validate(message) - u.course = message.text - u.setState('selectGroup') - - send(message, 'Выбери свою группу', reply_markup = Markup.selectGroup(u.topic, u.course)) - - -@bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'selectGroup') -def selectGroupState(message: Message): - u = validate(message) - u.group = message.text - u.setState('default') - - - send(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.valid = True + user.setState("default") + send( + message, + """ Приятного использования! -Здесь можно посмотреть расписание в более приятном виде, а также настроить уведомления о парах! +Здесь можно посмотреть расписание и настроить уведомления о парах. -Изменить факультет/курс/группу можно просто перезапустив бота -''', reply_markup = Markup.default(u)) \ No newline at end of file +Изменить группу можно через /start. +""", + reply_markup=Markup.default(user), + ) + + +@bot.message_handler( + content_types=["text"], func=lambda message: validate(message).state == "selectGroup" +) +def selectGroupState(message: Message): + query = (message.text or "").strip() + try: + matches = find_main_groups(query) + if len(matches) == 1 and group_exists(query): + return _finish_registration(message, matches[0]) + except ScheduleAPIError as error: + logger.error("Schedule", f"[{message.from_user.id}] {error}") + return send( + message, + "Сервер расписания сейчас недоступен. Попробуй отправить название группы позже.", + ) + + if not matches: + return send( + message, + "Похожих групп не нашлось. Проверь название и попробуй ещё раз.", + reply_markup=Markup.selectGroup([]), + ) + + send( + message, + "Выбери свою группу из самых похожих вариантов или введи название точнее.", + reply_markup=Markup.selectGroup(matches), + ) diff --git a/bot/schedule/other_days.py b/bot/schedule/other_days.py index 8940699..fee0514 100644 --- a/bot/schedule/other_days.py +++ b/bot/schedule/other_days.py @@ -11,6 +11,8 @@ def otherState(message: Message): 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) @@ -22,4 +24,4 @@ def otherState(message: Message): s.weekDay = index result += generateDaySchedule(s, False) +'\n\n' - return send(u.id, result, reply_markup = Markup.other()) \ No newline at end of file + return send(u.id, result, reply_markup = Markup.other()) diff --git a/config.example.py b/config.example.py index 8c1f5f3..041b9d7 100644 --- a/config.example.py +++ b/config.example.py @@ -8,7 +8,7 @@ TELEGRAM_PROXY: str | None = None WEB_BASE_URL: str = 'https://zatups.example.com' # Base URL for the schedule (should be runned separately) -SCHEDULE_BASE_URL: str = 'http://schedule_backend:8000/' +SCHEDULE_BASE_URL: str = 'http://10.9.8.3:18822' # Telegram user IDs -ADMINS: list[int] = [5016590523, 5001115363] \ No newline at end of file +ADMINS: list[int] = [5016590523, 5001115363] diff --git a/event/lesson.py b/event/lesson.py index 2583920..b092fc4 100644 --- a/event/lesson.py +++ b/event/lesson.py @@ -19,18 +19,11 @@ class Lesson(Event): t = s.today() if t == []: continue w = f'Скоро начнется пара\n' - for lessonList in t: - if len(lessonList) == 1: - if [lessonTime.hour, lessonTime.minute] == lessonList[0].start: - w += lessonList[0].text(isToday = True) - - else: - lesson = lessonList[1 if s.weekNumber%2 == 0 else 0] - if lesson and [lessonTime.hour, lessonTime.minute] == lesson.start: - w += lesson.text(isToday = True) + for lesson in t: + if [lessonTime.hour, lessonTime.minute] == lesson.start: + w += lesson.text(isToday=True) if len(w.split('\n')) > 2: send(user.id, w) addEvent(Lesson) - diff --git a/models/analytics.py b/models/analytics.py index 98800dd..35bb3f3 100644 --- a/models/analytics.py +++ b/models/analytics.py @@ -1,111 +1,50 @@ import datetime +import html import json -from typing import Any, Callable, Literal +from typing import Any -from .schedule import getFacCourseByGroup from .user import User -def I(condition: bool): return 1 if condition else 0 - -FEATURE_LIST: dict[str, Callable[[User], int]] = { - "accounts": lambda _: 1, - - "today_accounts": lambda u: I(u.usage["date"] == datetime.datetime.now().strftime("%Y.%m.%d")), - "today_interactions": lambda u: ( - max(u.usage["count"], 0) - if u.usage["date"] == datetime.datetime.now().strftime("%Y.%m.%d") - else 0 - ), - - "feature_groups": lambda u: I(len(u.saved_groups) > 1), - "feature_notifications": lambda u: I(len(u.notifications) > 0), - "feature_extended_schedule": lambda u: I('extended_schedule' in u.flags), - "fetaure_v2_ui": lambda u: I('v2_ui' in u.flags) -} - class Analytics: def __init__(self): - self.issued_at: datetime.datetime = datetime.datetime.now() - self.custom: dict[str, Any] = { - "group_list": [], - } - self.total: dict[str, int] = {} - self.data: dict[ - str, # faculty - dict[ - str, # course - dict[str, int], # features - ], - ] = {} + self.issued_at = datetime.datetime.now() + self.users = 0 + self.groups: dict[str, int] = {} - def add(self, u: User): - if u.state == "broken": + def add(self, user: User) -> None: + self.users += 1 + if not user.group: return + self.groups[user.group] = self.groups.get(user.group, 0) + 1 - faculty, course = getFacCourseByGroup(u.group) - if not faculty: - return - - faculty: str - course: str - - # custom - [self.custom["group_list"].append(groupname) for groupname in u.saved_groups] - self.custom["group_list"] = list(set(self.custom["group_list"])) - - # features - for feature_name, func in FEATURE_LIST.items(): - result = func(u) - - if feature_name not in self.total: - self.total[feature_name] = 0 - self.total[feature_name] += result - - if faculty not in self.data: - self.data[faculty] = {} - if course not in self.data[faculty]: - self.data[faculty][course] = {} - if feature_name not in self.data[faculty][course]: - self.data[faculty][course][feature_name] = 0 - self.data[faculty][course][feature_name] += result - - def toDict(self): + def toDict(self) -> dict[str, Any]: return { - "version": 2, + "version": 3, "issued_at": self.issued_at.isoformat(), - "total": self.total, - "data": self.data, - "custom": self.custom, + "users": self.users, + "groups": dict(sorted(self.groups.items())), } - def append(self): + def append(self) -> None: try: - w: list[dict[str, Any]] = json.load( - open("./web/analytics/data-v2.json", "r", encoding="utf-8") - ) - except: - w = [] - w.append(self.toDict()) - w = w[-30:] - json.dump( - w, - open("./web/analytics/data-v2.json", "w", encoding="utf-8"), - ensure_ascii=False, - # indent=0, - ) + with open("./web/analytics/data-v2.json", "r", encoding="utf-8") as file: + history: list[dict[str, Any]] = json.load(file) + except (FileNotFoundError, json.JSONDecodeError, OSError): + history = [] + + history.append(self.toDict()) + with open("./web/analytics/data-v2.json", "w", encoding="utf-8") as file: + json.dump(history[-30:], file, ensure_ascii=False) def text(self) -> str: + group_lines = [ + f"{html.escape(group)}: {count}" + for group, count in sorted(self.groups.items()) + ] + groups = "\n".join(group_lines) if group_lines else "Нет зарегистрированных групп" return ( - f""" -Аналитика за {self.issued_at.strftime("%m.%d")} 📊""" - + "\n".join( - [ - f"{feature.capitalize()}: {number}" - for feature, number in self.total.items() - ] - ) - + """ -Сайт -""" + f'Аналитика за {self.issued_at.strftime("%d.%m")} 📊\n' + f"Пользователи: {self.users}\n\n" + f"Группы\n{groups}" ) diff --git a/models/bot.py b/models/bot.py index 6906ca9..fc24349 100644 --- a/models/bot.py +++ b/models/bot.py @@ -5,7 +5,6 @@ import datetime from telebot.types import ReplyKeyboardMarkup, InlineQueryResultArticle, InputTextMessageContent, Message, ReplyKeyboardRemove, WebAppInfo, KeyboardButton from telebot import apihelper import config -import utils.schedule_client as sc if config.TELEGRAM_PROXY: apihelper.proxy = { @@ -166,19 +165,18 @@ class Markup_v1: return markup @staticmethod - def selectGroup(query: str) -> ReplyKeyboardMarkup: - #EDIT - schedule = sc.ScheduleDB(api_url=config.SCHEDULE_BASE_URL) - groupList = schedule.query(sc.Group).filter_by(name=query) - + def selectGroup(groups: list[str]) -> ReplyKeyboardMarkup: markup = ReplyKeyboardMarkup(resize_keyboard=True) - for group in groupList: + for group in groups: markup.add(group, row_width=1) - markup.add('Моей группы нет ⚠', row_width=1) + markup.add('Моей группы нету ⚠', row_width=1) return markup + +Markup = Markup_v1 + def Receive(state: str, text: str | None = None, strict: bool = True): return bot.message_handler( content_types = ['text'], @@ -188,4 +186,4 @@ def Receive(state: str, text: str | None = None, strict: bool = True): ) def Default(text: str, strict: bool = True): - return Receive('default', text, strict) \ No newline at end of file + return Receive('default', text, strict) diff --git a/models/schedule.py b/models/schedule.py index 460b5af..2916b8c 100644 --- a/models/schedule.py +++ b/models/schedule.py @@ -1,143 +1,274 @@ -import json -import datetime -import config as _config +from __future__ import annotations + +import datetime +import difflib +import html +import json +import re +import threading +import time +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import quote +from urllib.request import Request, urlopen +from zoneinfo import ZoneInfo + +import config + + +dayStrings = [ + "Понедельник", + "Вторник", + "Среда", + "Четверг", + "Пятница", + "Суббота", + "Воскресенье", +] + +MOSCOW = ZoneInfo("Europe/Moscow") +_CACHE_TTL = 300.0 +_REQUEST_TIMEOUT = 10.0 +_FAILURE_RETRY_DELAY = 30.0 +_cache: dict[str, tuple[float, list[dict[str, Any]]]] = {} +_failures: dict[str, tuple[float, str]] = {} +_cache_lock = threading.RLock() + + +class ScheduleAPIError(RuntimeError): + pass + + +def _api_url(path: str) -> str: + return f"{config.SCHEDULE_BASE_URL.rstrip('/')}/{path.lstrip('/')}" + + +def _load_result(path: str) -> list[dict[str, Any]]: + now = time.monotonic() + with _cache_lock: + cached = _cache.get(path) + if cached and now - cached[0] < _CACHE_TTL: + return cached[1] + failure = _failures.get(path) + if failure and now < failure[0]: + if cached: + return cached[1] + raise ScheduleAPIError(failure[1]) + + request = Request(_api_url(path), headers={"Accept": "application/json"}) + try: + with urlopen(request, timeout=_REQUEST_TIMEOUT) as response: + payload = json.load(response) + except (HTTPError, URLError, TimeoutError, OSError, ValueError) as error: + message = f"Не удалось получить {path}: {error}" + with _cache_lock: + _failures[path] = (now + _FAILURE_RETRY_DELAY, message) + if cached: + return cached[1] + raise ScheduleAPIError(message) from error + + if payload.get("ok") is not True or not isinstance(payload.get("result"), list): + if cached: + return cached[1] + raise ScheduleAPIError(f"Сервер вернул некорректный ответ для {path}") + + result = payload["result"] + with _cache_lock: + _cache[path] = (now, result) + _failures.pop(path, None) + return result + + +def clear_schedule_cache() -> None: + with _cache_lock: + _cache.clear() + _failures.clear() + + +def get_main_groups() -> list[dict[str, Any]]: + groups = _load_result("groups") + if not groups: + raise ScheduleAPIError("Сервер расписания пока не содержит групп") + return sorted( + (group for group in groups if group.get("parent_group_id") is None), + key=lambda group: str(group.get("name", "")).casefold(), + ) + + +def _compact_group_name(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: + 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] + if exact: + return exact[:1] + + contains = [name for name in groups if compact_query in _compact_group_name(name)] + if contains: + return contains[: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] + + +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() + ) + + +def _minutes_to_time(value: int | None) -> list[int]: + if value is None: + return [0, 0] + return [value // 60, value % 60] -dayStrings = ['Понедельник', 'Вторник', 'Среда', 'Четверг', 'Пятница', 'Суббота', 'Воскресенье'] class Lesson: - #EDIT: фулл переделать потому-что формат поменялся сильно - def __init__(self, start_end: str, data: dict) -> None: - self.start: list[int] = [ - int(start_end.split(' - ')[0].split(':')[0]), - int(start_end.split(' - ')[0].split(':')[1]) - ] - - self.end: list[int] = [ - int(start_end.split(' - ')[1].split(':')[0]), - int(start_end.split(' - ')[1].split(':')[1]) - ] - self.strTime: str = start_end - - self.raw: str = data['raw'] - self.ok: bool = data['ok'] - if not self.ok: return - - self.name: str = data['name'] - self.type: str = data['type'] - self.info: list[str] = data['info'] + 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", [])] + self.room_ids = [str(value) for value in data.get("room_ids", [])] + self.subgroup_names = subgroup_names + self.info = self.teacher_ids + [f"ауд. {room}" for room in self.room_ids] + self.is_odd_week = bool(data.get("is_odd_week")) + self.is_even_week = bool(data.get("is_even_week")) def text(self, isToday: bool = False) -> str: - date = datetime.datetime.now() - lessonStart = datetime.datetime(date.year, date.month, date.day, self.start[0], self.start[1], 10) - + 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 + ) - - if self.ok: - dateStr: str = f'{self.type.capitalize()} в {self.start[0]}:{self.start[1] if self.start[1] > 9 else "0"+str(self.start[1])}' - if isToday: - if lessonStart > date and (lessonStart-date).seconds < 30*60: - dateStr = f'{self.type.capitalize()} начнется через {max((lessonStart-date).seconds//60, 1)} мин.' - elif lessonStart < date and (date-lessonStart).seconds < 31*60: - dateStr = f'Сейчас идет {self.type}' - elif lessonStart < date and (date-lessonStart).seconds >= 31*60 and (date-lessonStart).seconds < 89*60: - dateStr = f'{self.type.capitalize()} закончится через {max(90-(date-lessonStart).seconds//60, 1)} мин.' - wrb = '' - if self.info: - nd: list[str] = [] - for data in self.info: - if data in self.type: continue - if 'подгр' in data: - nd.append(data.split('подгр')[0]) - nd.append('\nподгр'+data.split('подгр')[1]+f'\n') - - else: nd.append(data) - - - for data in nd: - # print() - # print(data, wrb) - if ('\n' in data or '\n' in wrb[-2:]): - wrb += data - elif 'ауд' in data.lower(): - audn = '.'.join(data.replace(' ', '').split('.')[1:]) - wrb += ', '+f'{data}' - else: - wrb += ', '+data - - - return f'''\ -
\ -{self.name} -{dateStr} -{wrb[2:] if wrb else 'Дополнительная информация отсуствует'}
''' - - else: - return f'''\ -
\
-Пара в {self.start[0]}:{self.start[1] if self.start[1] > 9 else '0'+str(self.start[1])}
-{self.raw}
''' + 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: + date_text = f"{lesson_type.capitalize()} начнется через {max(minutes, 1)} мин." + elif isToday and lesson_start <= now < lesson_end: + 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.extend( + f'ауд. {html.escape(room)}' + for room in self.room_ids + ) + if self.subgroup_names: + 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}
" + ) class Schedule: - def __init__(self, responce: dict) -> None: - self.weekNumber: int = responce['date']['weekNumber'] - self.weekDay: int = responce['date']['weekDay'] + def __init__( + self, + lessons: list[dict[str, Any]], + group_name: str, + 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() - self.raw: dict = responce['schedule'] - self.days: list[list[list[Lesson]]] = [] + 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} не найдена") - for dayInfo in self.raw.values(): - currentDay = [] - for start_end, lessons in dayInfo.items(): - if lessons == [None] or lessons == [None, None]: - continue - - lessonList: list[Lesson | None] = [] - for lesson in lessons: - if lesson: lessonList.append(Lesson(start_end, lesson)) - else: lessonList.append(None) + 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)] - currentDay.append(lessonList) + 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", []))) + 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))) - self.days.append(currentDay) + for day in self.days: + day.sort(key=lambda lesson: (lesson.start, lesson.name.casefold())) - def today(self) -> list[list[Lesson]]: - return self.days[self.weekDay] if self.weekDay < 5 else [] + def today(self) -> list[Lesson]: + if self.weekDay not in range(5): + return [] + even_week = self.weekNumber % 2 == 0 + return [ + lesson + for lesson in self.days[self.weekDay] + if (lesson.is_even_week if even_week else lesson.is_odd_week) + ] -def getFacCourseByGroup(group: str) -> tuple[str | None, str | None]: - for faculty, courses in _config.LOAD.items(): - for course in courses: - path = f'./timetables/{faculty} {course}.json'.replace(' ', '_') - data: dict[str, dict] = json.load( - open( - path, - 'r', - encoding = 'utf-8' - ) - ) - if group.lower() in [_.lower() for _ in data.keys()]: - return faculty, course - return None, None - -def getSchedulePathByGroup(group: str) -> str | None: - faculty, course = getFacCourseByGroup(group) - return None if not faculty else f'./timetables/{faculty} {course}.json'.replace(' ', '_') -def generateDaySchedule(schedule: Schedule, showOddWeekText: bool = True, isToday: bool = False) -> str: - isWeekOdd = schedule.weekNumber%2 == 0 - weekDay = schedule.weekDay - schedule_today = schedule.today() - lastLessonEnds = [0,0] - scheduleResult = f'{dayStrings[weekDay]}, {"четная" if isWeekOdd else "нечетная"} неделя\n' if showOddWeekText else f'{dayStrings[weekDay]}' - for lessonList in schedule_today: - if len(lessonList) == 1: - scheduleResult += lessonList[0].text(isToday) + '\n' - lastLessonEnds = lessonList[0].end - else: - lesson = lessonList[1 if isWeekOdd else 0] - if lesson: - scheduleResult += lesson.text(isToday) + '\n' - lastLessonEnds = lesson.end - - scheduleResult += f'Пары закончатся в {lastLessonEnds[0]}:{lastLessonEnds[1] if lastLessonEnds[1] > 9 else "0"+str(lastLessonEnds[1])}' - return scheduleResult \ No newline at end of file +def get_group_schedule(group_name: str) -> Schedule: + groups = _load_result("groups") + if not groups: + raise ScheduleAPIError("Сервер расписания пока не содержит групп") + lessons = _load_result("schedule/default") + return Schedule(lessons, group_name, groups) + + +def generateDaySchedule( + schedule: Schedule, showOddWeekText: bool = True, isToday: bool = False +) -> str: + even_week = schedule.weekNumber % 2 == 0 + weekday = schedule.weekDay + day_name = dayStrings[weekday] + header = ( + f'{day_name}, {"четная" if even_week else "нечетная"} неделя\n' + if showOddWeekText + else f"{day_name}\n" + ) + lessons = schedule.today() + if not lessons: + return header + "Пар нет 🥰" + + result = header + "\n".join(lesson.text(isToday) for lesson in lessons) + last_end = max(lesson.end for lesson in lessons) + return result + f"\nПары закончатся в {last_end[0]}:{last_end[1]:02d}" diff --git a/models/user.py b/models/user.py index 102abe1..9209798 100644 --- a/models/user.py +++ b/models/user.py @@ -1,39 +1,39 @@ -import datetime -import hashlib import json import os +from typing import Any, Iterator -from telebot.types import CallbackQuery, InlineQuery, Message, ReplyKeyboardRemove +from telebot.types import CallbackQuery, InlineQuery, Message from models import logger -from .schedule import Schedule, getSchedulePathByGroup +from .schedule import Schedule, ScheduleAPIError, get_group_schedule class User: def __init__(self, userID: int) -> None: self.id: int = userID try: - self.data: dict[str, str] = json.load( + self.data: dict[str, Any] = json.load( open(f"./users/{self.id}.json", "r", encoding="utf-8") ) - except: + 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.topic: str | None = self.data.get("topic", None) - self.course: str | None = self.data.get("course", None) 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) - self.saved_groups: list[str] = self.data.get("saved_groups", []) - self.saved_groups.append(self.group) - self.saved_groups = [t if "-" in t else "{null}" for t in self.saved_groups] - self.saved_groups = list(set(self.saved_groups)) - try: self.saved_groups.remove("{null}") - except: ... + 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)) self.user: dict = self.data.get("user", {}) self.usage: dict = self.data.get( @@ -42,34 +42,29 @@ class User: self.flags: list[str] = self.data.get('flags', []) if "notifyEvening" in self.data.keys(): - self.notifyEvening: str = self.data["notifyEvening"] - self.notifyBeforeLesson: str = self.data["notifyBeforeLesson"] if self.data["notifyEvening"] and "evening" not in self.notifications: self.notifications.append("evening") - if self.data["notifyBeforeLesson"] and "lesson" not in self.notifications: + if self.data.get("notifyBeforeLesson") and "lesson" not in self.notifications: self.notifications.append("lesson") self.save() def schedule(self) -> Schedule | None: - date = datetime.datetime.now() - path = getSchedulePathByGroup(self.group) - if not path: + if not self.group: self.valid = False self.notifications = [] self.setState("broken") - self.save() return None - - return Schedule( - { - "date": { - "weekNumber": date.isocalendar().week, - "weekDay": date.weekday(), - }, - "schedule": json.load(open(path, "r", encoding="utf-8"))[self.group], - } - ) + try: + return get_group_schedule(self.group) + except ValueError: + self.valid = False + self.notifications = [] + self.setState("broken") + return None + except ScheduleAPIError as error: + logger.error("Schedule", f"[{self.id}] {error}") + return None def setState(self, newState: str) -> None: self.state = newState @@ -79,8 +74,6 @@ class User: return json.dump( { "state": self.state, - "topic": self.topic, - "course": self.course, "group": self.group, "notifications": self.notifications, "valid": self.valid, @@ -103,9 +96,9 @@ def validate(ctx: Message | InlineQuery | CallbackQuery) -> User: return False # type: ignore -def getAllUsers() -> list[User]: # type: ignore +def getAllUsers() -> Iterator[User]: for _, _, filenames in os.walk("./users"): for userFile in filenames: - if ".json" not in userFile: + if not userFile.endswith(".json"): continue - yield User(int(userFile.replace(".json", ""))) # type: ignore + yield User(int(userFile.removesuffix(".json"))) diff --git a/utils/calendar.py b/utils/calendar.py index d1972a2..4f32f58 100644 --- a/utils/calendar.py +++ b/utils/calendar.py @@ -20,7 +20,6 @@ class ICS: else: # on this week self.start_week_date = timezone.localize(datetime.now()) - timedelta(days=schedule.weekDay) - self.isWeekOdd = schedule.weekNumber%2 == 0 self.calendar = Calendar() for target in range(5): @@ -28,10 +27,7 @@ class ICS: event_date = self.start_week_date + timedelta(days=target) # print(f'\n\n--- {target} -> {event_date.strftime("%Y-%m-%d")} ---') - for lessonList in schedule.today(): - lesson = lessonList[0] if len(lessonList) == 1 else lessonList[1 if self.isWeekOdd else 0] - if not lesson: continue - + for lesson in schedule.today(): event = Event() event.name = f'{lesson.name}' @@ -47,7 +43,7 @@ class ICS: # break # break - + def save(self) -> str: """returns path to file""" content = self.calendar.serialize() @@ -59,4 +55,3 @@ class ICS: path = f'./temp/{self.userID}.ics' with open(path, 'w', encoding='utf-8', newline='') as file: file.writelines(content) return path - \ No newline at end of file