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

migrate bot to schedule API

Browse source
This commit is contained in:
Server 2026-08-30 11:43:38 +00:00
parent 41a5edde19
commit c75f14e588
11 changed files with 498 additions and 493 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

176
bot/inline.py
View file

@ -1,138 +1,74 @@
from telebot.types import (InlineQueryResultArticle, InputTextMessageContent, InlineQuery) from telebot.types import InlineQuery
from models.bot import bot, Inlines
from models.user import validate
from models.schedule import generateDaySchedule, dayStrings
from models import logger 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) @bot.inline_handler(lambda query: True)
def default_query(query: InlineQuery): def default_query(query: InlineQuery):
user = validate(query) user = validate(query)
inline = Inlines(query.id) inline = Inlines(query.id)
if not user: if not user.group:
logger.warn('Bot', 'Inlinue from unknown user invoked')
inline.add( inline.add(
'Расписание недоступно ❌', "Расписание недоступно ❌",
'Запусти бота, чтобы использовать эту функцию', "Запусти бота, чтобы выбрать группу",
'Не удалось получить расписание. Пользователь не запустил бота ❌ \n\nhttps://t.me/zatups_bot' "Не удалось получить расписание. Сначала запусти бота: https://t.me/zatups_bot",
) )
else:
schedule = user.schedule()
else: if not schedule:
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'<b>{description}</b>\nСегодня пар нету 🥰'
inline.add( inline.add(
'Расписание на сегодня 🔥', "Расписание недоступно ❌",
f'{description}, {weektype.lower()} неделя', "Сервер расписания сейчас недоступен",
output "Не удалось получить расписание. Попробуй ещё раз позже.",
) )
else:
if 'tomorrow': week_type = "четная" if schedule.weekNumber % 2 == 0 else "нечетная"
s = user.schedule() description = dayStrings[schedule.weekDay]
output = False
description = dayStrings[(s.weekDay+1)%len(dayStrings)]
if s.weekDay in [4, 5]:
output = f'<b>{description}</b>\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 'нечетная'
inline.add( inline.add(
'Расписание на завтра 🍀', "Расписание на сегодня 🔥",
f'{description}, {weektype.lower()} неделя', f"{description}, {week_type} неделя",
output generateDaySchedule(schedule, isToday=True),
) )
if 'week': tomorrow = user.schedule()
s = user.schedule() if tomorrow:
description = f'Сейчас {"четная" if s.weekNumber%2==0 else "нечетная"} неделя' if tomorrow.weekDay == 6:
output = f'<i>Расписание на </i><b>эту</b><i> {"четную" if s.weekNumber%2==0 else "нечетную"} неделю</i>\n' 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: week = user.schedule()
s.weekDay = 0 if week:
s.weekNumber += 1 title = "эту"
description = f'Следующая {"четная" if s.weekNumber%2==0 else "нечетная"} неделя' if week.weekDay == 6:
output = f'<i>Расписание на </i><b>следующую</b><i> {"четную" if s.weekNumber%2==0 else "нечетную"} неделю</i>\n' week.weekNumber += 1
title = "следующую"
week_type = "четную" if week.weekNumber % 2 == 0 else "нечетную"
output = f"<i>Расписание на </i><b>{title}</b><i> {week_type} неделю</i>\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): logger.log("Bot", f"Inline from {user.id}")
s.weekDay = index
output += generateDaySchedule(s, False) +'\n\n'
inline.add(
'Расписание на неделю 📅',
description,
output
)
logger.log('Bot', f'Inline from {user.id}')
try: try:
return inline.send() return inline.send()
except Exception as e: except Exception as error:
logger.error('Bot', f'Inline.send() -> {e}') logger.error("Bot", f"Inline.send() -> {error}")

19
bot/main/home.py
View file

@ -30,21 +30,23 @@ def forceReturnToDefaultState(query: CallbackQuery):
@bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'default') @bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'default')
def defaultState(message: Message): def defaultState(message: Message):
u = validate(message) u = validate(message)
s = u.schedule()
# print(u.id, message.text)
if message.text[0] == '/': return ContinueHandling() # type: ignore if message.text[0] == '/': return ContinueHandling() # type: ignore
if message.text == f'{u.group} 🔄': if message.text == f'{u.group} 🔄':
u.setState('savedGroups') u.setState('savedGroups')
return send(u.id, f'<b>Сохраненные группы</b>\nЗдесь список групп, которые ранее были использованы, чтобы быстро между ними переключаться. \n\n<i>Активная группа: {u.group}</i>', Markup.saved_groups(u)) return send(u.id, f'<b>Сохраненные группы</b>\nЗдесь список групп, которые ранее были использованы, чтобы быстро между ними переключаться. \n\n<i>Активная группа: {u.group}</i>', 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: match message.text:
case 'Сегодня 🔥': case 'Сегодня 🔥':
if s.weekDay > 4:
return send(u.id, f'<b>{dayStrings[s.weekDay]}</b>\nСегодня пар нет 🥰')
return send(u.id, generateDaySchedule(s, isToday = True), reply_markup = Markup.default(u)) return send(u.id, generateDaySchedule(s, isToday = True), reply_markup = Markup.default(u))
case 'Завтра 🍀': case 'Завтра 🍀':
@ -73,7 +75,7 @@ def defaultState(message: Message):
case 'Другие дни 📁': case 'Другие дни 📁':
u.setState('otherDays') 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()) return send(u.id, f'<b>Сегодня {dayStrings[s.weekDay].lower()}, {"четная" if s.weekNumber%2==0 else "нечетная"} неделя</b>\n\nВыбери нужный день/пункт', reply_markup = Markup.other())
case 'Уведомления ⏰': case 'Уведомления ⏰':
u.setState('notifications') 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 [bot.forward_message(adminID, message.from_user.id, message.id) for adminID in ADMINS] # type: ignore
send(message, '<b>Сообщение переслано разработчику 🔧</b>\nСпасибо за обращение', Markup.default(u)) send(message, '<b>Сообщение переслано разработчику 🔧</b>\nСпасибо за обращение', Markup.default(u))

179
bot/main/register_user.py
View file

@ -1,97 +1,114 @@
import json import json
from telebot.types import Message
from telebot.types import Message, ReplyKeyboardRemove
from telebot.util import quick_markup from telebot.util import quick_markup
from models import logger from models import logger
from models.user import User, validate, getAllUsers from models.bot import Markup_v1 as Markup, bot, send
from models.bot import (Markup_v1 as Markup, bot, send) from models.schedule import ScheduleAPIError, find_main_groups, group_exists
from models.schedule import generateDaySchedule from models.user import getAllUsers, validate
@bot.message_handler(commands = ['start'])
@bot.message_handler(commands=["start"])
def start(message: Message): def start(message: Message):
u = validate(message) user = validate(message)
if u and 'migrate' in message.text: # type: ignore if user.group and "migrate" in (message.text or ""):
u.setState('default') user.setState("default")
logger.log('Bot', f'{message.from_user.id} migrated') # type: ignore logger.log("Bot", f"{message.from_user.id} migrated")
send(message, 'Добро пожаловать снова 💞', reply_markup = Markup.default(u)) return send(message, "Добро пожаловать снова 💞", reply_markup=Markup.default(user))
send(message, f'''<b>Что нового</b>
<blockquote>1. Теперь доступны все факультеты, курсы и группы
2. Затупсика можно упомянуть в любом чате, даже в личке и группе, и он отправит актуальное расписание!
<i>Для этого напиши <code>@zatups_bot</code> в любом чате</i>
3. Теперь расписания отображаются корректно всегда, даже с подгруппами</blockquote>
<i>Чтобы воспользоваться функцией отправки расписания в любом чате, напиши <code>@zatups_bot</code> в любом чате, выбери бота из списка, а после нажми на нужный день - сегодня, завтра, или на всю неделю.\nКстати - в следующий раз достаточно просто написать @ и телеграм сам выдаст затупсика в списке!</i>''') was_registered = bool(user.group)
return logger.log("Bot", f"[{user.id}] /start detected, Total users: {len(list(getAllUsers()))}")
json.dump(
logger.log('Bot', f'[{u.id}] /start detected, Total users: {len(list(getAllUsers()))+1}') {
json.dump( "state": "selectGroup",
{ "group": None,
'state': 'selectTopic', "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': '', prefix = (
'course': '', "<i>Вы начали процедуру перезапуска бота. Уведомления отключены.</i>\n\n"
'group': '', if was_registered
'saved_groups': u.saved_groups if u else [], else ""
'notifyEvening': False, )
'notifyBeforeLesson': False, send(
}, message,
open(f'./users/{message.from_user.id}.json', 'w', encoding='utf-8'), # type: ignore prefix + "Напиши название своей группы. Например: <code>АТ-501</code>",
ensure_ascii = False, reply_markup=ReplyKeyboardRemove(),
indent = 4 )
)
send(message, '''
<b>Бот на стадии разработки!</b>
- <i>Доступны не все функции, группы</i>
- <i>Возможны баги, фризы, затупы итд итп</i>
- <i>Бот изначально не планировался для публичного использования</i>
Просто имей это ввиду😥
''')
#EDIT
send(message, 'Выбери свой факультет' if not u else '<i>Вы начали процедуру перезапуска бота. Все настройки и данные удалены, включая уведомления</i>\nВыбери новый факультет', reply_markup = Markup.selectTopic())
@bot.message_handler(content_types=["text"], func=lambda message: "нету ⚠" in message.text)
@bot.message_handler(content_types = ['text'], func = lambda m: ' нету ⚠' in m.text)
def notFoundHandler(message: Message): def notFoundHandler(message: Message):
bot.reply_to(message, ''' bot.reply_to(
message,
"""
<b>Не расстраивайся!</b> <b>Не расстраивайся!</b>
<i>Напиши разработчику, чтобы он добавил твой факультет/курс/группу</i> <i>Напиши разработчику, если твоей группы нет в расписании.</i>
""",
<blockquote> reply_markup=quick_markup(
К сожалению, для поддержки работы бота нужно много мощностей и материальные вложения. {
Бюджет, выделенный на разработку бота - <i>2 банки пива</i>, поэтому в целях экономии в боте пока доступны не все группы "Написать разработчику 💞": {"url": "https://t.me/m/Gg00FvJmZDM6"},
</blockquote>''', reply_markup = quick_markup({'Написать разработчику 💞': {'url': 'https://t.me/m/Gg00FvJmZDM6'}, 'Перезапустить бота 🤖': {'url': 'https://t.me/zatups_bot?start=r'}}, row_width = 1)) "Перезапустить бота 🤖": {"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 _finish_registration(message: Message, group_name: str) -> None:
def selectTopicState(message: Message): user = validate(message)
u = validate(message) user.group = group_name
u.topic = message.text if group_name not in user.saved_groups:
u.setState('selectCourse') user.saved_groups.append(group_name)
user.valid = True
send(message, 'Выбери свой курс', reply_markup = Markup.selectCourse(u.topic)) user.setState("default")
send(
@bot.message_handler(content_types = ['text'], func = lambda m: validate(m).state == 'selectCourse') message,
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, '''
<b>Приятного использования!</b> <b>Приятного использования!</b>
Здесь можно посмотреть расписание в более приятном виде, а также настроить уведомления о парах! Здесь можно посмотреть расписание и настроить уведомления о парах.
<i>Изменить факультет/курс/группу можно просто перезапустив бота</i> <i>Изменить группу можно через /start.</i>
''', reply_markup = Markup.default(u)) """,
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),
)

4
bot/schedule/other_days.py
View file

@ -11,6 +11,8 @@ def otherState(message: Message):
if message.text not in possibleResponces and message.text not in ['Четная 📅', 'Нечетная 📅']: return if message.text not in possibleResponces and message.text not in ['Четная 📅', 'Нечетная 📅']: return
result: str = '' result: str = ''
s = u.schedule() s = u.schedule()
if not s:
return send(message, 'Не удалось получить расписание. Попробуй ещё раз позже.')
if message.text in possibleResponces: if message.text in possibleResponces:
s.weekDay = possibleResponces.index(message.text) s.weekDay = possibleResponces.index(message.text)
result = generateDaySchedule(s) result = generateDaySchedule(s)
@ -22,4 +24,4 @@ def otherState(message: Message):
s.weekDay = index s.weekDay = index
result += generateDaySchedule(s, False) +'\n\n' result += generateDaySchedule(s, False) +'\n\n'
return send(u.id, result, reply_markup = Markup.other()) return send(u.id, result, reply_markup = Markup.other())

4
config.example.py
View file

@ -8,7 +8,7 @@ TELEGRAM_PROXY: str | None = None
WEB_BASE_URL: str = 'https://zatups.example.com' WEB_BASE_URL: str = 'https://zatups.example.com'
# Base URL for the schedule (should be runned separately) # 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 # Telegram user IDs
ADMINS: list[int] = [5016590523, 5001115363] ADMINS: list[int] = [5016590523, 5001115363]

13
event/lesson.py
View file

@ -19,18 +19,11 @@ class Lesson(Event):
t = s.today() t = s.today()
if t == []: continue if t == []: continue
w = f'<b>Скоро начнется пара</b>\n' w = f'<b>Скоро начнется пара</b>\n'
for lessonList in t: for lesson in t:
if len(lessonList) == 1: if [lessonTime.hour, lessonTime.minute] == lesson.start:
if [lessonTime.hour, lessonTime.minute] == lessonList[0].start: w += lesson.text(isToday=True)
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)
if len(w.split('\n')) > 2: send(user.id, w) if len(w.split('\n')) > 2: send(user.id, w)
addEvent(Lesson) addEvent(Lesson)

121
models/analytics.py
View file

@ -1,111 +1,50 @@
import datetime import datetime
import html
import json import json
from typing import Any, Callable, Literal from typing import Any
from .schedule import getFacCourseByGroup
from .user import User 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: class Analytics:
def __init__(self): def __init__(self):
self.issued_at: datetime.datetime = datetime.datetime.now() self.issued_at = datetime.datetime.now()
self.custom: dict[str, Any] = { self.users = 0
"group_list": [], self.groups: dict[str, int] = {}
}
self.total: dict[str, int] = {}
self.data: dict[
str, # faculty
dict[
str, # course
dict[str, int], # features
],
] = {}
def add(self, u: User): def add(self, user: User) -> None:
if u.state == "broken": self.users += 1
if not user.group:
return return
self.groups[user.group] = self.groups.get(user.group, 0) + 1
faculty, course = getFacCourseByGroup(u.group) def toDict(self) -> dict[str, Any]:
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):
return { return {
"version": 2, "version": 3,
"issued_at": self.issued_at.isoformat(), "issued_at": self.issued_at.isoformat(),
"total": self.total, "users": self.users,
"data": self.data, "groups": dict(sorted(self.groups.items())),
"custom": self.custom,
} }
def append(self): def append(self) -> None:
try: try:
w: list[dict[str, Any]] = json.load( with open("./web/analytics/data-v2.json", "r", encoding="utf-8") as file:
open("./web/analytics/data-v2.json", "r", encoding="utf-8") history: list[dict[str, Any]] = json.load(file)
) except (FileNotFoundError, json.JSONDecodeError, OSError):
except: history = []
w = []
w.append(self.toDict()) history.append(self.toDict())
w = w[-30:] with open("./web/analytics/data-v2.json", "w", encoding="utf-8") as file:
json.dump( json.dump(history[-30:], file, ensure_ascii=False)
w,
open("./web/analytics/data-v2.json", "w", encoding="utf-8"),
ensure_ascii=False,
# indent=0,
)
def text(self) -> str: def text(self) -> str:
group_lines = [
f"{html.escape(group)}: <code>{count}</code>"
for group, count in sorted(self.groups.items())
]
groups = "\n".join(group_lines) if group_lines else "Нет зарегистрированных групп"
return ( return (
f""" f'<b>Аналитика за {self.issued_at.strftime("%d.%m")} 📊</b>\n'
<b>Аналитика за {self.issued_at.strftime("%m.%d")} 📊</b>""" f"Пользователи: <code>{self.users}</code>\n\n"
+ "\n".join( f"<b>Группы</b>\n{groups}"
[
f"{feature.capitalize()}: <code>{number}</code>"
for feature, number in self.total.items()
]
)
+ """
<i><a href="https://zatups.sinya.ru/analytics">Сайт</a></i>
"""
) )

16
models/bot.py
View file

@ -5,7 +5,6 @@ import datetime
from telebot.types import ReplyKeyboardMarkup, InlineQueryResultArticle, InputTextMessageContent, Message, ReplyKeyboardRemove, WebAppInfo, KeyboardButton from telebot.types import ReplyKeyboardMarkup, InlineQueryResultArticle, InputTextMessageContent, Message, ReplyKeyboardRemove, WebAppInfo, KeyboardButton
from telebot import apihelper from telebot import apihelper
import config import config
import utils.schedule_client as sc
if config.TELEGRAM_PROXY: if config.TELEGRAM_PROXY:
apihelper.proxy = { apihelper.proxy = {
@ -166,19 +165,18 @@ class Markup_v1:
return markup return markup
@staticmethod @staticmethod
def selectGroup(query: str) -> ReplyKeyboardMarkup: def selectGroup(groups: list[str]) -> ReplyKeyboardMarkup:
#EDIT
schedule = sc.ScheduleDB(api_url=config.SCHEDULE_BASE_URL)
groupList = schedule.query(sc.Group).filter_by(name=query)
markup = ReplyKeyboardMarkup(resize_keyboard=True) markup = ReplyKeyboardMarkup(resize_keyboard=True)
for group in groupList: for group in groups:
markup.add(group, row_width=1) markup.add(group, row_width=1)
markup.add('Моей группы нет ⚠', row_width=1) markup.add('Моей группы нету ⚠', row_width=1)
return markup return markup
Markup = Markup_v1
def Receive(state: str, text: str | None = None, strict: bool = True): def Receive(state: str, text: str | None = None, strict: bool = True):
return bot.message_handler( return bot.message_handler(
content_types = ['text'], 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): def Default(text: str, strict: bool = True):
return Receive('default', text, strict) return Receive('default', text, strict)

385
models/schedule.py
View file

@ -1,143 +1,274 @@
import json from __future__ import annotations
import datetime
import config as _config 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: class Lesson:
#EDIT: фулл переделать потому-что формат поменялся сильно def __init__(self, data: dict[str, Any], subgroup_names: list[str]) -> None:
def __init__(self, start_end: str, data: dict) -> None: self.start = _minutes_to_time(data.get("time_start"))
self.start: list[int] = [ self.end = _minutes_to_time(data.get("time_end"))
int(start_end.split(' - ')[0].split(':')[0]), self.strTime = f"{self.start[0]:02d}:{self.start[1]:02d} - {self.end[0]:02d}:{self.end[1]:02d}"
int(start_end.split(' - ')[0].split(':')[1]) 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.end: list[int] = [ self.room_ids = [str(value) for value in data.get("room_ids", [])]
int(start_end.split(' - ')[1].split(':')[0]), self.subgroup_names = subgroup_names
int(start_end.split(' - ')[1].split(':')[1]) self.info = self.teacher_ids + [f"ауд. {room}" for room in self.room_ids]
] self.is_odd_week = bool(data.get("is_odd_week"))
self.strTime: str = start_end self.is_even_week = bool(data.get("is_even_week"))
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 text(self, isToday: bool = False) -> str: def text(self, isToday: bool = False) -> str:
date = datetime.datetime.now() now = datetime.datetime.now(MOSCOW)
lessonStart = datetime.datetime(date.year, date.month, date.day, self.start[0], self.start[1], 10) 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)
if self.ok: date_text = f"{lesson_type.capitalize()} <i>в {self.start[0]}:{self.start[1]:02d}</i>"
dateStr: str = f'{self.type.capitalize()} <i>в {self.start[0]}:{self.start[1] if self.start[1] > 9 else "0"+str(self.start[1])}</i>' if isToday and lesson_start > now:
if isToday: minutes = int((lesson_start - now).total_seconds() // 60)
if lessonStart > date and (lessonStart-date).seconds < 30*60: if minutes < 30:
dateStr = f'{self.type.capitalize()} <i>начнется через {max((lessonStart-date).seconds//60, 1)} мин.</i>' date_text = f"{lesson_type.capitalize()} <i>начнется через {max(minutes, 1)} мин.</i>"
elif lessonStart < date and (date-lessonStart).seconds < 31*60: elif isToday and lesson_start <= now < lesson_end:
dateStr = f'<i>Сейчас идет</i> {self.type}' minutes = int((lesson_end - now).total_seconds() // 60)
elif lessonStart < date and (date-lessonStart).seconds >= 31*60 and (date-lessonStart).seconds < 89*60: date_text = f"<i>Сейчас идет</i> {lesson_type}, закончится через {max(minutes, 1)} мин."
dateStr = f'{self.type.capitalize()} <i>закончится через {max(90-(date-lessonStart).seconds//60, 1)} мин.</i>'
wrb = '' details: list[str] = []
if self.info: details.extend(html.escape(teacher) for teacher in self.teacher_ids)
nd: list[str] = [] details.extend(
for data in self.info: f'<a href="https://rasp.pgups.ru/schedule/room?room={quote(room)}">ауд. {html.escape(room)}</a>'
if data in self.type: continue for room in self.room_ids
if 'подгр' in data: )
nd.append(data.split('подгр')[0]) if self.subgroup_names:
nd.append('\n<b>подгр'+data.split('подгр')[1]+f'</b>\n') details.append(html.escape(", ".join(self.subgroup_names)))
else: nd.append(data) info = ", ".join(details) if details else "Дополнительная информация отсутствует"
return (
f"<blockquote><b>{html.escape(self.name)}</b>\n"
for data in nd: f"{date_text}\n{info}</blockquote>"
# 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'<a href="https://rasp.pgups.ru/schedule/room?room={audn}">{data}</a>'
else:
wrb += ', '+data
return f'''\
<blockquote>\
<b>{self.name}</b>
{dateStr}
{wrb[2:] if wrb else 'Дополнительная информация отсуствует'}</blockquote>'''
else:
return f'''\
<pre language="Error">\
Пара в {self.start[0]}:{self.start[1] if self.start[1] > 9 else '0'+str(self.start[1])}
{self.raw}</pre>'''
class Schedule: class Schedule:
def __init__(self, responce: dict) -> None: def __init__(
self.weekNumber: int = responce['date']['weekNumber'] self,
self.weekDay: int = responce['date']['weekDay'] 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'] base_group = next(
self.days: list[list[list[Lesson]]] = [] (
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(): base_id = str(base_group["id"])
currentDay = [] child_names = {
for start_end, lessons in dayInfo.items(): str(group["id"]): str(group["name"])
if lessons == [None] or lessons == [None, None]: for group in groups
continue if group.get("parent_group_id") == base_id
}
lessonList: list[Lesson | None] = [] target_ids = {base_id, *child_names}
for lesson in lessons: self.days: list[list[Lesson]] = [[] for _ in range(5)]
if lesson: lessonList.append(Lesson(start_end, lesson))
else: lessonList.append(None)
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]]: def today(self) -> list[Lesson]:
return self.days[self.weekDay] if self.weekDay < 5 else [] 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: def get_group_schedule(group_name: str) -> Schedule:
isWeekOdd = schedule.weekNumber%2 == 0 groups = _load_result("groups")
weekDay = schedule.weekDay if not groups:
schedule_today = schedule.today() raise ScheduleAPIError("Сервер расписания пока не содержит групп")
lastLessonEnds = [0,0] lessons = _load_result("schedule/default")
scheduleResult = f'<b>{dayStrings[weekDay]}, {"четная" if isWeekOdd else "нечетная"} неделя</b>\n' if showOddWeekText else f'<b>{dayStrings[weekDay]}</b>' return Schedule(lessons, group_name, groups)
for lessonList in schedule_today:
if len(lessonList) == 1:
scheduleResult += lessonList[0].text(isToday) + '\n' def generateDaySchedule(
lastLessonEnds = lessonList[0].end schedule: Schedule, showOddWeekText: bool = True, isToday: bool = False
else: ) -> str:
lesson = lessonList[1 if isWeekOdd else 0] even_week = schedule.weekNumber % 2 == 0
if lesson: weekday = schedule.weekDay
scheduleResult += lesson.text(isToday) + '\n' day_name = dayStrings[weekday]
lastLessonEnds = lesson.end header = (
f'<b>{day_name}, {"четная" if even_week else "нечетная"} неделя</b>\n'
scheduleResult += f'Пары закончатся в <i>{lastLessonEnds[0]}:{lastLessonEnds[1] if lastLessonEnds[1] > 9 else "0"+str(lastLessonEnds[1])}</i>' if showOddWeekText
return scheduleResult else f"<b>{day_name}</b>\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Пары закончатся в <i>{last_end[0]}:{last_end[1]:02d}</i>"

65
models/user.py
View file

@ -1,39 +1,39 @@
import datetime
import hashlib
import json import json
import os 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 models import logger
from .schedule import Schedule, getSchedulePathByGroup from .schedule import Schedule, ScheduleAPIError, get_group_schedule
class User: class User:
def __init__(self, userID: int) -> None: def __init__(self, userID: int) -> None:
self.id: int = userID self.id: int = userID
try: 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") 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!') logger.error("Bot", f'INVALID "{userID}" SAVED DATA! RESET!')
self.data = {} self.data = {}
self.state: str | bool | None = self.data.get("state", None) 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.group: str | None = self.data.get("group", None)
self.notifications: list[str] = self.data.get("notifications", []) # type: ignore self.notifications: list[str] = self.data.get("notifications", []) # type: ignore
self.valid: bool = self.data.get("valid", True) self.valid: bool = self.data.get("valid", True)
self.saved_groups: list[str] = self.data.get("saved_groups", []) saved_groups = self.data.get("saved_groups", [])
self.saved_groups.append(self.group) self.saved_groups = [
self.saved_groups = [t if "-" in t else "{null}" for t in self.saved_groups] value for value in saved_groups if isinstance(value, str) and value.strip()
self.saved_groups = list(set(self.saved_groups)) ]
try: self.saved_groups.remove("{null}") if self.group and self.group not in self.saved_groups:
except: ... self.saved_groups.append(self.group)
self.saved_groups = list(dict.fromkeys(self.saved_groups))
self.user: dict = self.data.get("user", {}) self.user: dict = self.data.get("user", {})
self.usage: dict = self.data.get( self.usage: dict = self.data.get(
@ -42,34 +42,29 @@ class User:
self.flags: list[str] = self.data.get('flags', []) self.flags: list[str] = self.data.get('flags', [])
if "notifyEvening" in self.data.keys(): 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: if self.data["notifyEvening"] and "evening" not in self.notifications:
self.notifications.append("evening") 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.notifications.append("lesson")
self.save() self.save()
def schedule(self) -> Schedule | None: def schedule(self) -> Schedule | None:
date = datetime.datetime.now() if not self.group:
path = getSchedulePathByGroup(self.group)
if not path:
self.valid = False self.valid = False
self.notifications = [] self.notifications = []
self.setState("broken") self.setState("broken")
self.save()
return None return None
try:
return Schedule( return get_group_schedule(self.group)
{ except ValueError:
"date": { self.valid = False
"weekNumber": date.isocalendar().week, self.notifications = []
"weekDay": date.weekday(), self.setState("broken")
}, return None
"schedule": json.load(open(path, "r", encoding="utf-8"))[self.group], except ScheduleAPIError as error:
} logger.error("Schedule", f"[{self.id}] {error}")
) return None
def setState(self, newState: str) -> None: def setState(self, newState: str) -> None:
self.state = newState self.state = newState
@ -79,8 +74,6 @@ class User:
return json.dump( return json.dump(
{ {
"state": self.state, "state": self.state,
"topic": self.topic,
"course": self.course,
"group": self.group, "group": self.group,
"notifications": self.notifications, "notifications": self.notifications,
"valid": self.valid, "valid": self.valid,
@ -103,9 +96,9 @@ def validate(ctx: Message | InlineQuery | CallbackQuery) -> User:
return False # type: ignore return False # type: ignore
def getAllUsers() -> list[User]: # type: ignore def getAllUsers() -> Iterator[User]:
for _, _, filenames in os.walk("./users"): for _, _, filenames in os.walk("./users"):
for userFile in filenames: for userFile in filenames:
if ".json" not in userFile: if not userFile.endswith(".json"):
continue continue
yield User(int(userFile.replace(".json", ""))) # type: ignore yield User(int(userFile.removesuffix(".json")))

9
utils/calendar.py
View file

@ -20,7 +20,6 @@ class ICS:
else: # on this week else: # on this week
self.start_week_date = timezone.localize(datetime.now()) - timedelta(days=schedule.weekDay) self.start_week_date = timezone.localize(datetime.now()) - timedelta(days=schedule.weekDay)
self.isWeekOdd = schedule.weekNumber%2 == 0
self.calendar = Calendar() self.calendar = Calendar()
for target in range(5): for target in range(5):
@ -28,10 +27,7 @@ class ICS:
event_date = self.start_week_date + timedelta(days=target) event_date = self.start_week_date + timedelta(days=target)
# print(f'\n\n--- {target} -> {event_date.strftime("%Y-%m-%d")} ---') # print(f'\n\n--- {target} -> {event_date.strftime("%Y-%m-%d")} ---')
for lessonList in schedule.today(): for lesson in schedule.today():
lesson = lessonList[0] if len(lessonList) == 1 else lessonList[1 if self.isWeekOdd else 0]
if not lesson: continue
event = Event() event = Event()
event.name = f'{lesson.name}' event.name = f'{lesson.name}'
@ -47,7 +43,7 @@ class ICS:
# break # break
# break # break
def save(self) -> str: def save(self) -> str:
"""returns path to file""" """returns path to file"""
content = self.calendar.serialize() content = self.calendar.serialize()
@ -59,4 +55,3 @@ class ICS:
path = f'./temp/{self.userID}.ics' path = f'./temp/{self.userID}.ics'
with open(path, 'w', encoding='utf-8', newline='') as file: file.writelines(content) with open(path, 'w', encoding='utf-8', newline='') as file: file.writelines(content)
return path return path