360 lines
12 KiB
Python
360 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import datetime
|
||
import difflib
|
||
import html
|
||
import json
|
||
import re
|
||
import threading
|
||
import time
|
||
from typing import Any, Literal, TypedDict
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import quote
|
||
from urllib.request import Request, urlopen
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import config
|
||
from utils import convert
|
||
|
||
|
||
TargetKind = Literal["group", "room", "teacher"]
|
||
|
||
|
||
class ScheduleTarget(TypedDict):
|
||
kind: TargetKind
|
||
id: str
|
||
name: str
|
||
|
||
|
||
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()
|
||
|
||
_ENDPOINTS: dict[TargetKind, str] = {
|
||
"group": "groups",
|
||
"room": "rooms",
|
||
"teacher": "teachers",
|
||
}
|
||
|
||
|
||
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 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:
|
||
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(value: str) -> str:
|
||
return re.sub(r"[^0-9a-zа-яё]", "", value.casefold())
|
||
|
||
|
||
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 []
|
||
|
||
exact = [record for record in records if _compact(str(record.get("name", ""))) == compact_query]
|
||
if exact:
|
||
return exact[:limit]
|
||
|
||
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 sorted(contains, key=score, reverse=True)[: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)
|
||
return any(_compact(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]
|
||
|
||
|
||
class Lesson:
|
||
def __init__(
|
||
self,
|
||
data: dict[str, Any],
|
||
target_kind: TargetKind,
|
||
group_names: list[str],
|
||
) -> None:
|
||
self.start = _minutes_to_time(data.get("time_start"))
|
||
self.end = _minutes_to_time(data.get("time_end"))
|
||
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.target_kind = target_kind
|
||
self.group_names = group_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:
|
||
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_type = html.escape(self.type)
|
||
date_text = f"{lesson_type.capitalize()} <i>в {self.start[0]}:{self.start[1]:02d}</i>"
|
||
|
||
if isToday and lesson_start > now:
|
||
minutes = int((lesson_start - now).total_seconds() // 60)
|
||
if minutes < 30:
|
||
date_text = f"{lesson_type.capitalize()} <i>начнется через {max(minutes, 1)} мин.</i>"
|
||
elif isToday and lesson_start <= now < lesson_end:
|
||
minutes = int((lesson_end - now).total_seconds() // 60)
|
||
date_text = f"<i>Сейчас идет</i> {lesson_type}, закончится через {max(minutes, 1)} мин."
|
||
|
||
def generate_url(kind, target):
|
||
return f'<a href="{config.TELEGRAM_URL}?start=_s{kind[:1]}{convert.encode(html.escape(target))}">{html.escape(target)}</a>'
|
||
|
||
teachers = [
|
||
generate_url("teacher", teacher)
|
||
for teacher in self.teacher_ids
|
||
]
|
||
rooms = [
|
||
generate_url("room", room)
|
||
for room in self.room_ids
|
||
]
|
||
groups = [
|
||
generate_url("group", group)
|
||
for group in self.group_names
|
||
]
|
||
|
||
if self.target_kind == "teacher":
|
||
details = rooms + groups
|
||
elif self.target_kind == "room":
|
||
details = groups + teachers
|
||
elif self.target_kind == "group":
|
||
details = teachers + rooms
|
||
else:
|
||
details = teachers + rooms + groups
|
||
|
||
info = ", ".join(details) if details else "Дополнительная информация отсутствует"
|
||
return f"<blockquote><b>{html.escape(self.name)}</b>\n{date_text}\n{info}</blockquote>"
|
||
|
||
|
||
class Schedule:
|
||
def __init__(
|
||
self,
|
||
lessons: list[dict[str, Any]],
|
||
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()
|
||
self.days: list[list[Lesson]] = [[] for _ in range(5)]
|
||
|
||
groups_by_id = {
|
||
str(group["id"]): str(group["name"])
|
||
for group in groups
|
||
}
|
||
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(field, [])))
|
||
if not matched_ids:
|
||
continue
|
||
if target["kind"] == "group":
|
||
group_names = sorted(child_names[item] for item in matched_ids if item in child_names)
|
||
else:
|
||
group_names = sorted(
|
||
{
|
||
groups_by_id.get(str(group_id), str(group_id))
|
||
for group_id in raw_lesson.get("group_ids", [])
|
||
}
|
||
)
|
||
self.days[weekday - 1].append(Lesson(raw_lesson, target["kind"], group_names))
|
||
|
||
for day in self.days:
|
||
day.sort(key=lambda lesson: (lesson.start, lesson.name.casefold()))
|
||
|
||
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 get_schedule(target: ScheduleTarget) -> Schedule:
|
||
groups = _load_result("groups")
|
||
lessons = _load_result("schedule/default")
|
||
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(
|
||
schedule: Schedule, showOddWeekText: bool = True, isToday: bool = False
|
||
) -> str:
|
||
even_week = schedule.weekNumber % 2 == 0
|
||
weekday = schedule.weekDay
|
||
header = (
|
||
f'<b>{dayStrings[weekday]}, {"четная" if even_week else "нечетная"} неделя</b>\n'
|
||
if showOddWeekText
|
||
else f"<b>{dayStrings[weekday]}</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>"
|
||
|
||
|
||
def generateWeekSchedule(schedule: Schedule) -> str:
|
||
original_weekday = schedule.weekDay
|
||
days = []
|
||
try:
|
||
for weekday in range(len(schedule.days)):
|
||
schedule.weekDay = weekday
|
||
if schedule.today():
|
||
days.append(generateDaySchedule(schedule, False))
|
||
finally:
|
||
schedule.weekDay = original_weekday
|
||
|
||
return "\n\n".join(days) if days else "Пар на этой неделе нет 🥰"
|