diff --git a/.gitignore b/.gitignore index da45b79..cf11bf0 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,5 @@ timetables/*.json .timetracker web/analytics/data.json DEBUG -schedule_client.py config.py .env \ No newline at end of file diff --git a/utils/schedule_client.py b/utils/schedule_client.py new file mode 100644 index 0000000..57dd9d8 --- /dev/null +++ b/utils/schedule_client.py @@ -0,0 +1,644 @@ +"""Однофайловый клиент для SheetScheduleParser API. + +Файл можно просто скопировать в другой проект. Внешние зависимости не нужны. + +Пример:: + + import schedule_client as schedule + + schedule.configure("http://schedule.example.com", cache_ttl=300) + db = schedule.ScheduleDB() + + lessons = ( + db.query(schedule.ScheduleEntry) + .filter_by(schedule_type="default", group_ids="АТ-501", weekday=1) + .order_by(schedule.Schedule.time_start) + .all() + ) +""" + +from __future__ import annotations + +import json +import os +import threading +import time +from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any, Generic, TypeVar +from urllib.error import HTTPError, URLError +from urllib.parse import quote +from urllib.request import Request, urlopen + + +# Эти значения можно изменить прямо в скопированном файле или через configure(). +API_URL = os.getenv("SCHEDULE_API_URL", "http://127.0.0.1:8000") +# Должно в точности совпадать с app.version.API_VERSION на сервере. +API_VERSION = "1" +CACHE_TTL = float(os.getenv("SCHEDULE_CACHE_TTL", "300")) +REQUEST_TIMEOUT = float(os.getenv("SCHEDULE_REQUEST_TIMEOUT", "15")) +FAILURE_RETRY_DELAY = float(os.getenv("SCHEDULE_FAILURE_RETRY_DELAY", "15")) + + +class ScheduleClientError(RuntimeError): + """Базовая ошибка клиента.""" + + +class APIError(ScheduleClientError): + """API недоступно или вернуло некорректный ответ.""" + + +class IncompatibleAPIVersionError(ScheduleClientError): + """Версия контракта API не поддерживается этим файлом клиента.""" + + +class NoResultFound(ScheduleClientError): + """Query.one() не нашёл ни одной записи.""" + + +class MultipleResultsFound(ScheduleClientError): + """Query.one() нашёл больше одной записи.""" + + +class Expression: + def __init__(self, predicate: Callable[["Model"], bool], label: str = "") -> None: + self._predicate = predicate + self._label = label + + def __call__(self, item: "Model") -> bool: + return self._predicate(item) + + def __and__(self, other: "Expression") -> "Expression": + return Expression(lambda item: self(item) and other(item)) + + def __or__(self, other: "Expression") -> "Expression": + return Expression(lambda item: self(item) or other(item)) + + def __invert__(self) -> "Expression": + return Expression(lambda item: not self(item)) + + def __repr__(self) -> str: + return self._label or "Expression(...)" + + +@dataclass(frozen=True) +class SortExpression: + name: str + reverse: bool = False + + +class Field: + """Дескриптор поля модели и строитель выражений для Query.filter().""" + + def __init__(self) -> None: + self.name = "" + + def __set_name__(self, owner: type["Model"], name: str) -> None: + self.name = name + + def __get__(self, instance: "Model | None", owner: type["Model"]) -> Any: + if instance is None: + return self + return instance._data.get(self.name) + + def _compare(self, operation: Callable[[Any, Any], bool], value: Any, symbol: str) -> Expression: + return Expression( + lambda item: operation(item._data.get(self.name), value), + f"{self.name} {symbol} {value!r}", + ) + + def __eq__(self, value: Any) -> Expression: # type: ignore[override] + return self._compare(lambda actual, expected: actual == expected, value, "==") + + def __ne__(self, value: Any) -> Expression: # type: ignore[override] + return self._compare(lambda actual, expected: actual != expected, value, "!=") + + def __lt__(self, value: Any) -> Expression: + return self._compare(_safe_lt, value, "<") + + def __le__(self, value: Any) -> Expression: + return self._compare(_safe_le, value, "<=") + + def __gt__(self, value: Any) -> Expression: + return self._compare(_safe_gt, value, ">") + + def __ge__(self, value: Any) -> Expression: + return self._compare(_safe_ge, value, ">=") + + def contains(self, value: Any) -> Expression: + return Expression( + lambda item: _contains(item._data.get(self.name), value), + f"{self.name}.contains({value!r})", + ) + + def in_(self, values: Iterable[Any]) -> Expression: + choices = tuple(values) + return Expression( + lambda item: item._data.get(self.name) in choices, + f"{self.name}.in_({choices!r})", + ) + + def is_(self, value: Any) -> Expression: + return Expression(lambda item: item._data.get(self.name) is value) + + def startswith(self, value: str) -> Expression: + return Expression( + lambda item: str(item._data.get(self.name) or "").startswith(value) + ) + + def asc(self) -> SortExpression: + return SortExpression(self.name) + + def desc(self) -> SortExpression: + return SortExpression(self.name, reverse=True) + + +class Model: + """Неизменяемая по соглашению запись из API с доступом через атрибуты.""" + + def __init__(self, data: Mapping[str, Any]) -> None: + self._data = dict(data) + + def as_dict(self) -> dict[str, Any]: + return dict(self._data) + + def __getitem__(self, key: str) -> Any: + return self._data[key] + + def __repr__(self) -> str: + identity = self._data.get("id", self._data.get("name", "?")) + return f"{type(self).__name__}(id={identity!r})" + + +class Schedule(Model): + id = Field() + schedule_type = Field() + name = Field() + type = Field() + weekday = Field() + is_odd_week = Field() + is_even_week = Field() + time_start = Field() + time_end = Field() + datetime = Field() + group_ids = Field() + room_ids = Field() + teacher_ids = Field() + teacher_assignments = Field() + created_at = Field() + parsed_at = Field() + + +# Более явное имя для импорта; Schedule оставлено как короткий вариант. +ScheduleEntry = Schedule + + +class Group(Model): + id = Field() + name = Field() + faculty = Field() + course = Field() + parent_group_id = Field() + created_at = Field() + updated_at = Field() + last_seen_at = Field() + + +class Teacher(Model): + id = Field() + name = Field() + positions = Field() + created_at = Field() + updated_at = Field() + last_seen_at = Field() + + +class Room(Model): + id = Field() + name = Field() + created_at = Field() + updated_at = Field() + last_seen_at = Field() + + +class GroupHistory(Model): + faculty = Field() + course = Field() + parent_group_id = Field() + valid_from = Field() + valid_to = Field() + + +M = TypeVar("M", bound=Model) + + +class Query(Generic[M]): + """Небольшое неизменяемое подмножество привычного SQLAlchemy Query API.""" + + def __init__( + self, + client: "ScheduleDB", + dataset: str, + model: type[M], + predicates: tuple[Expression, ...] = (), + sort_expressions: tuple[SortExpression, ...] = (), + offset_value: int = 0, + limit_value: int | None = None, + ) -> None: + self._client = client + self._dataset = dataset + self._model = model + self._predicates = predicates + self._sort_expressions = sort_expressions + self._offset = offset_value + self._limit = limit_value + + def _clone(self, **changes: Any) -> "Query[M]": + values = { + "client": self._client, + "dataset": self._dataset, + "model": self._model, + "predicates": self._predicates, + "sort_expressions": self._sort_expressions, + "offset_value": self._offset, + "limit_value": self._limit, + } + values.update(changes) + return Query(**values) + + def filter(self, *criteria: Expression | Callable[[M], bool]) -> "Query[M]": + expressions: list[Expression] = [] + for criterion in criteria: + if isinstance(criterion, Expression): + expressions.append(criterion) + elif callable(criterion): + expressions.append(Expression(criterion)) + else: + raise TypeError("filter() принимает выражения полей или callable") + return self._clone(predicates=self._predicates + tuple(expressions)) + + def filter_by(self, **values: Any) -> "Query[M]": + expressions: list[Expression] = [] + for name, expected in values.items(): + if not hasattr(self._model, name): + raise AttributeError(f"У {self._model.__name__} нет поля {name!r}") + expressions.append( + Expression( + lambda item, key=name, value=expected: _filter_by_match( + item._data.get(key), value + ), + f"{name}={expected!r}", + ) + ) + return self._clone(predicates=self._predicates + tuple(expressions)) + + def order_by(self, *fields: Field | SortExpression | str) -> "Query[M]": + expressions: list[SortExpression] = [] + for value in fields: + if isinstance(value, Field): + expressions.append(value.asc()) + elif isinstance(value, SortExpression): + expressions.append(value) + elif isinstance(value, str): + expressions.append(SortExpression(value)) + else: + raise TypeError("order_by() принимает поле, field.asc()/desc() или имя поля") + return self._clone(sort_expressions=tuple(expressions)) + + def limit(self, value: int) -> "Query[M]": + if value < 0: + raise ValueError("limit не может быть отрицательным") + return self._clone(limit_value=value) + + def offset(self, value: int) -> "Query[M]": + if value < 0: + raise ValueError("offset не может быть отрицательным") + return self._clone(offset_value=value) + + def all(self) -> list[M]: + records = list(self._client._records(self._dataset)) + for predicate in self._predicates: + records = [item for item in records if predicate(item)] + for expression in reversed(self._sort_expressions): + records.sort( + key=lambda item, name=expression.name: _sort_key(item._data.get(name)), + reverse=expression.reverse, + ) + end = None if self._limit is None else self._offset + self._limit + return records[self._offset : end] + + def first(self) -> M | None: + records = self.limit(1).all() + return records[0] if records else None + + def one_or_none(self) -> M | None: + records = self.limit(2).all() + if len(records) > 1: + raise MultipleResultsFound("Ожидалась максимум одна запись") + return records[0] if records else None + + def one(self) -> M: + result = self.one_or_none() + if result is None: + raise NoResultFound("Запись не найдена") + return result + + def count(self) -> int: + return len(self.all()) + + def get(self, entity_id: str) -> M | None: + return self.filter_by(id=entity_id).first() + + def __iter__(self) -> Iterator[M]: + return iter(self.all()) + + +@dataclass +class _CacheState: + lock: threading.RLock = field(default_factory=threading.RLock) + datasets: dict[str, tuple[Model, ...]] = field(default_factory=dict) + refreshed_monotonic: float = 0.0 + refreshed_at: str | None = None + retry_after: float = 0.0 + last_error: str | None = None + + +_CACHE_LOCK = threading.Lock() +_GLOBAL_CACHE: dict[str, _CacheState] = {} + + +def configure( + api_url: str, + *, + cache_ttl: float | None = None, + request_timeout: float | None = None, +) -> None: + """Задаёт глобальные настройки для последующих ScheduleDB().""" + + global API_URL, CACHE_TTL, REQUEST_TIMEOUT + API_URL = api_url.rstrip("/") + if cache_ttl is not None: + CACHE_TTL = float(cache_ttl) + if request_timeout is not None: + REQUEST_TIMEOUT = float(request_timeout) + + +def clear_cache(api_url: str | None = None) -> None: + """Очищает один глобальный снимок или весь кеш процесса.""" + + with _CACHE_LOCK: + if api_url is None: + states = list(_GLOBAL_CACHE.values()) + _GLOBAL_CACHE.clear() + else: + state = _GLOBAL_CACHE.pop(api_url.rstrip("/"), None) + states = [state] if state is not None else [] + # Сбрасываем и уже выданные клиентам state, а не только запись в словаре. + for state in states: + with state.lock: + state.datasets.clear() + state.refreshed_monotonic = 0.0 + state.refreshed_at = None + state.retry_after = 0.0 + state.last_error = None + + +class ScheduleDB: + """Точка входа к кешированной копии всех основных справочников API.""" + + def __init__( + self, + api_url: str | None = None, + *, + cache_ttl: float | None = None, + request_timeout: float | None = None, + eager: bool = True, + ) -> None: + self.api_url = (api_url or API_URL).rstrip("/") + self.cache_ttl = CACHE_TTL if cache_ttl is None else float(cache_ttl) + self.request_timeout = ( + REQUEST_TIMEOUT if request_timeout is None else float(request_timeout) + ) + with _CACHE_LOCK: + self._state = _GLOBAL_CACHE.setdefault(self.api_url, _CacheState()) + # Проверяется при каждом создании клиента, даже если общий кеш уже заполнен. + self._check_compatibility() + if eager: + self._ensure_fresh(require_data=True, compatibility_checked=True) + + @property + def schedule(self) -> Query[Schedule]: + return Query(self, "schedule", Schedule) + + @property + def default(self) -> Query[Schedule]: + return Query(self, "default", Schedule) + + @property + def exam(self) -> Query[Schedule]: + return Query(self, "exam", Schedule) + + @property + def groups(self) -> Query[Group]: + return Query(self, "groups", Group) + + @property + def teachers(self) -> Query[Teacher]: + return Query(self, "teachers", Teacher) + + @property + def rooms(self) -> Query[Room]: + return Query(self, "rooms", Room) + + def query(self, model: type[M]) -> Query[M]: + """SQLAlchemy-подобная форма: db.query(ScheduleEntry).filter(...).""" + + datasets: dict[type[Model], str] = { + Schedule: "schedule", + Group: "groups", + Teacher: "teachers", + Room: "rooms", + } + dataset = datasets.get(model) + if dataset is None: + raise TypeError( + "query() поддерживает ScheduleEntry, Group, Teacher и Room; " + "для истории используйте group_history(group_id)" + ) + return Query(self, dataset, model) + + def group_history(self, group_id: str) -> Query[GroupHistory]: + """Загружает историю группы лениво и дальше хранит в общем кеше.""" + + dataset = f"group_history:{group_id}" + return Query(self, dataset, GroupHistory) + + def refresh(self) -> None: + """Принудительно и атомарно обновляет все основные наборы или бросает APIError.""" + + with self._state.lock: + try: + self._refresh_locked() + except APIError as error: + self._state.last_error = str(error) + self._state.retry_after = time.monotonic() + FAILURE_RETRY_DELAY + raise + + def cache_info(self) -> dict[str, Any]: + with self._state.lock: + return { + "api_url": self.api_url, + "refreshed_at": self._state.refreshed_at, + "last_error": self._state.last_error, + "counts": { + name: len(items) + for name, items in self._state.datasets.items() + if not name.startswith("group_history:") + }, + } + + def _records(self, dataset: str) -> Sequence[Model]: + self._ensure_fresh(require_data=True) + with self._state.lock: + if dataset.startswith("group_history:") and dataset not in self._state.datasets: + group_id = dataset.partition(":")[2] + response = self._fetch(f"/groups/{quote(group_id, safe='')}/history") + self._state.datasets[dataset] = tuple( + GroupHistory(item) for item in response["result"] + ) + return self._state.datasets.get(dataset, ()) + + def _ensure_fresh( + self, + *, + require_data: bool, + compatibility_checked: bool = False, + ) -> None: + with self._state.lock: + now = time.monotonic() + has_data = bool(self._state.datasets) + stale = not has_data or now - self._state.refreshed_monotonic >= self.cache_ttl + if not stale or (has_data and now < self._state.retry_after): + return + try: + self._refresh_locked(check_compatibility=not compatibility_checked) + except APIError as error: + self._state.last_error = str(error) + self._state.retry_after = time.monotonic() + FAILURE_RETRY_DELAY + if require_data and not has_data: + raise + + def _refresh_locked(self, *, check_compatibility: bool = True) -> None: + if check_compatibility: + self._check_compatibility() + endpoints = { + "default": ("/schedule/default", Schedule), + "exam": ("/schedule/exam", Schedule), + "groups": ("/groups", Group), + "teachers": ("/teachers", Teacher), + "rooms": ("/rooms", Room), + } + fresh: dict[str, tuple[Model, ...]] = {} + for name, (path, model) in endpoints.items(): + response = self._fetch(path) + records: list[Model] = [] + for raw_item in response["result"]: + item = dict(raw_item) + if name in {"default", "exam"}: + item.setdefault("schedule_type", name) + records.append(model(item)) + fresh[name] = tuple(records) + fresh["schedule"] = fresh["default"] + fresh["exam"] + + # Истории групп инвалидируются: после обновления основной базы они могли измениться. + self._state.datasets = fresh + self._state.refreshed_monotonic = time.monotonic() + self._state.refreshed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + self._state.retry_after = 0.0 + self._state.last_error = None + + def _fetch(self, path: str) -> dict[str, Any]: + return _request_json(f"{self.api_url}{path}", self.request_timeout) + + def _check_compatibility(self) -> None: + response = self._fetch("/version") + actual = response.get("api_version") + if actual != API_VERSION: + raise IncompatibleAPIVersionError( + f"Несовместимая версия API: сервер={actual!r}, " + f"клиент ожидает {API_VERSION!r}" + ) + + +def _request_json(url: str, timeout: float) -> dict[str, Any]: + request = Request(url, headers={"Accept": "application/json"}) + try: + with urlopen(request, timeout=timeout) as response: + payload = json.load(response) + except HTTPError as error: + raise APIError(f"API вернуло HTTP {error.code} для {url}") from error + except (URLError, TimeoutError, OSError) as error: + raise APIError(f"Не удалось обратиться к {url}: {error}") from error + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise APIError(f"API вернуло невалидный JSON для {url}") from error + if not isinstance(payload, dict) or payload.get("ok") is not True: + raise APIError(f"Некорректный ответ API для {url}") + return payload + + +def _filter_by_match(actual: Any, expected: Any) -> bool: + # Самый частый запрос — filter_by(group_ids="АТ-501"). Для списковых полей + # одиночное значение означает contains, а коллекция — точное сравнение. + if isinstance(actual, (list, tuple, set)) and not isinstance( + expected, (list, tuple, set) + ): + return expected in actual + return actual == expected + + +def _contains(container: Any, value: Any) -> bool: + try: + return value in container + except (TypeError, AttributeError): + return False + + +def _sort_key(value: Any) -> tuple[bool, Any]: + return value is None, value + + +def _safe_lt(left: Any, right: Any) -> bool: + return left is not None and right is not None and left < right + + +def _safe_le(left: Any, right: Any) -> bool: + return left is not None and right is not None and left <= right + + +def _safe_gt(left: Any, right: Any) -> bool: + return left is not None and right is not None and left > right + + +def _safe_ge(left: Any, right: Any) -> bool: + return left is not None and right is not None and left >= right + + +__all__ = [ + "API_URL", + "API_VERSION", + "CACHE_TTL", + "APIError", + "IncompatibleAPIVersionError", + "NoResultFound", + "MultipleResultsFound", + "ScheduleDB", + "Schedule", + "ScheduleEntry", + "Group", + "Teacher", + "Room", + "GroupHistory", + "Query", + "configure", + "clear_cache", +]