initial

api.py

@@ -0,0 +1,193 @@
+import httpx
+
+from datetime import date
+from cachetools import TTLCache
+
+from config import (
+    API_URL,
+    REQUEST_TIMEOUT,
+    GROUPS_CACHE_TTL,
+)
+
+from database import (
+    get_cached_schedule,
+    save_schedule_cache,
+)
+
+
+def get_schedule_ttl(day: date) -> int:
+    """
+    Возвращает TTL кэша в минутах.
+    """
+
+    today = date.today()
+
+    if day == today:
+        # Сегодня — обновляем каждый час
+        return 60
+
+    if day > today:
+        # Будущие даты — раз в сутки
+        return 60 * 24
+
+    # Прошедшие даты — неделю
+    return 60 * 24 * 7
+
+
+class ScheduleAPI:
+
+    def __init__(self):
+
+        self.client = httpx.AsyncClient(
+            timeout=REQUEST_TIMEOUT
+        )
+
+        self._groups_cache = TTLCache(
+            maxsize=1,
+            ttl=GROUPS_CACHE_TTL
+        )
+
+    async def close(self):
+        await self.client.aclose()
+
+    async def _get(
+        self,
+        endpoint: str,
+        params: dict | None = None
+    ):
+
+        response = await self.client.get(
+            f"{API_URL}/{endpoint}",
+            params=params
+        )
+
+        response.raise_for_status()
+
+        return response.json()
+
+    async def get_groups(self):
+
+        if "groups" in self._groups_cache:
+            return self._groups_cache["groups"]
+
+        response = await self._get("Groups")
+
+        groups = (
+            response
+            .get("data", {})
+            .get("groups", [])
+        )
+
+        self._groups_cache["groups"] = groups
+
+        return groups
+
+    async def get_group_by_id(
+        self,
+        group_id: int
+    ):
+
+        groups = await self.get_groups()
+
+        for group in groups:
+
+            if group["groupID"] == group_id:
+                return group
+
+        return None
+
+    async def search_group_by_name(
+        self,
+        text: str
+    ):
+
+        groups = await self.get_groups()
+
+        text = text.upper().strip()
+
+        exact = [
+            g
+            for g in groups
+            if g["groupName"].upper() == text
+        ]
+
+        if exact:
+            return exact
+
+        return [
+            g
+            for g in groups
+            if text in g["groupName"].upper()
+        ]
+
+    async def get_schedule(
+        self,
+        group_id: int,
+        day: date | None = None
+    ):
+
+        request_day = day or date.today()
+
+        schedule_date = request_day.isoformat()
+
+        ttl = get_schedule_ttl(request_day)
+
+        # Проверяем кэш
+        cached = await get_cached_schedule(
+            group_id,
+            schedule_date,
+            ttl
+        )
+
+        if cached is not None:
+            print(f"[CACHE] {group_id} {schedule_date}")
+            return cached
+
+        print(f"[API] {group_id} {schedule_date}")
+
+        params = {
+            "idGroup": group_id,
+            "sdate": schedule_date
+        }
+
+        response = await self._get(
+            "Rasp",
+            params
+        )
+
+        schedule = response.get(
+            "data",
+            {}
+        )
+
+        # Сохраняем в кэш
+        await save_schedule_cache(
+            group_id,
+            schedule_date,
+            schedule
+        )
+
+        return schedule
+
+    async def get_teachers(self):
+
+        response = await self._get("Prepods")
+
+        return (
+            response
+            .get("data", {})
+            .get("prepods", [])
+        )
+
+    async def get_auditories(self):
+
+        response = await self._get("Auds")
+
+        return (
+            response
+            .get("data", {})
+            .get("auds", [])
+        )
+
+
+api = ScheduleAPI()