add: Трекать бд клиент

.gitignore

@@ -9,6 +9,5 @@ timetables/*.json
 .timetracker
 web/analytics/data.json
 DEBUG
-schedule_client.py
 config.py
 .env

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",
+]