6413 Белов А. Н. Лаб.2 Вар.1

.gitignore

@@ -0,0 +1,32 @@
+venv/
+env/
+ENV/
+.venv/
+
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+
+debug*.html
+test_*.html
+*.log
+
+cache/
+
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+.DS_Store
+Thumbs.db
+
+*.egg-info/
+*.egg
+dist/
+build/
+
+.env
+.env.local

README.md

@@ -1,61 +1,27 @@
-# Безопасность веб-приложений. Лабораторка №2
-
-## Схема сдачи
-
-1. Получить задание
-2. Сделать форк данного репозитория
-3. Выполнить задание согласно полученному варианту
-4. Сделать PR (pull request) в данный репозиторий 
-6. Исправить замечания после code review
-7. Получить approve 
-8. Прийти на занятие и защитить работу
-
-Что нужно проявить в работе:
-- умение разработать завершенное целое веб-приложение, с клиентской и серверной частями (допустимы открытые АПИ)
-- навыки верстки на html в объеме 200-300 тегов
-- навыки применения css для лейаута и стилизации, желательно с адаптацией к мобилке
-- использование jQuery или аналогичных JS-фреймворков
-- динамическая подгрузка контента
-- динамическое изменение DOM и CSSOM
-
-Если у вас своя идея по заданию, то расскажите, обсудим и подкорректирую.
-
-## Вариант 1. Расписания
-
-Сделать аналог раздела https://ssau.ru/rasp?groupId=531030143
-
-Какие нужны возможности:
-- справочники групп, табличные данные по расписаниям добывать с настоящего сайта на серверной стороне приложения
-- в клиентскую часть подгружать эти сведения динамически по JSON-API
-- обеспечить возможность смотреть расписания в разрезе группы или препода
-- обеспечить возможность выбора учебной недели (по умолчанию выбирается автоматически)
-
-## Вариант 2. Аналог Прибывалки для электричек
-
-Сделать веб-версию Прибывалки, только для электричек
-
-Какие нужны возможности:
-- находить желаемую ЖД-станцию поиском по названию и по карте
-- отображать расписания всех проходящих поездов через выбранную станцию
-- отображать расписания для поездов между двумя станциями
-- работа через АПИ Яндекс.Расписаний https://yandex.ru/dev/rasp/doc/ru/ (доступ получите сами)
-- хорошая работа в условиях экрана смартфона
-- бонус: функция "любимых остановок"
-
-## Вариант 3. Прогноз погоды
-
-Сделать одностраничный сайт с картой, на которой можно выбрать населенный пункт и получить прогноз погоды на несколько дней по нему.
-
-Какие нужны возможности:
- - увидеть на карте точки с населенными пунктами. Координаты населенных пунктов взять из https://tochno.st/datasets/allsettlements - но все 150 тысяч не нужно, выберите 1 тысячу с самым большим населением.
- - при нажатии на точку получить всплывающее окошко с графиками изменения температуры, осадков, силы ветра. API для прогнозов возьмите с https://projecteol.ru/ru/ с соблюдением правил.
- - графики рисовать каким-нибудь приличным компонентом, например, https://www.chartjs.org/
- - находить населенный пункт по названию
- - можете реализовать с собственным серверным компонентом или придумать, как обойтись без него
-
+## Установка и запуск
 
+### 1. Клонирование репозитория
+```bash
+git clone <url-репозитория>
+cd websec-2
+```
 
+### 2. Создание виртуального окружения
+```bash
+python -m venv venv
+venv\Scripts\activate
+```
 
+### 3. Установка зависимостей
+```bash
+pip install -r requirements.txt
+```
 
+### 4. Запуск сервера
+```bash
+python run.py
+```
 
+### 5. Открыть в браузере
 
+http://localhost:8000

app/cache.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+import asyncio
+import pickle
+import time
+from pathlib import Path
+from typing import Any, Awaitable, Callable, Dict, Generic, Optional, TypeVar
+
+T = TypeVar("T")
+
+
+class AsyncTTLCache(Generic[T]):
+    """
+    In-memory cache with optional disk persistence.
+    """
+
+    def __init__(self, ttl_seconds: int, max_items: int = 512, persist_dir: Optional[str] = None):
+        self.ttl_seconds = int(ttl_seconds)
+        self.max_items = int(max_items)
+
+        self._data: Dict[Any, tuple[float, T]] = {}
+        self._inflight: Dict[Any, asyncio.Task[T]] = {}
+        self._lock = asyncio.Lock()
+
+        self._persist_dir = persist_dir
+        if persist_dir:
+            Path(persist_dir).mkdir(parents=True, exist_ok=True)
+            self._load_from_disk()
+
+    def _now(self) -> float:
+        return time.monotonic()
+
+    def _is_expired(self, expires_at: float) -> bool:
+        return expires_at <= self._now()
+
+    def _get_key_path(self, key: Any) -> Optional[Path]:
+        if not self._persist_dir:
+            return None
+        safe_key = str(key).replace("/", "_").replace("\\", "_").replace(":", "_")
+        return Path(self._persist_dir) / f"{safe_key}.pickle"
+
+    def _load_from_disk(self):
+        """Загружает кэш с диска при старте"""
+        if not self._persist_dir:
+            return
+        try:
+            for p in Path(self._persist_dir).glob("*.pickle"):
+                try:
+                    with open(p, "rb") as f:
+                        data = pickle.load(f)
+                        if isinstance(data, tuple) and len(data) == 2:
+                            expires_at, value = data
+                            if not self._is_expired(expires_at):
+                                # Восстанавливаем ключ из имени файла
+                                key = p.stem.replace("_", "/")
+                                self._data[key] = (expires_at, value)
+                except Exception:
+                    pass
+        except Exception:
+            pass
+
+    def _save_to_disk(self, key: Any, expires_at: float, value: T):
+        """Сохраняет значение на диск"""
+        path = self._get_key_path(key)
+        if path:
+            try:
+                with open(path, "wb") as f:
+                    pickle.dump((expires_at, value), f)
+            except Exception:
+                pass
+
+    async def get(self, key: Any) -> Optional[T]:
+        async with self._lock:
+            item = self._data.get(key)
+            if not item:
+                return None
+            expires_at, value = item
+            if self._is_expired(expires_at):
+                self._data.pop(key, None)
+                return None
+            return value
+
+    async def set(self, key: Any, value: T) -> None:
+        async with self._lock:
+            if len(self._data) >= self.max_items:
+                self._data.pop(next(iter(self._data.keys())), None)
+            expires_at = self._now() + self.ttl_seconds
+            self._data[key] = (expires_at, value)
+            self._save_to_disk(key, expires_at, value)
+
+    async def get_or_set(self, key: Any, factory: Callable[[], Awaitable[T]]) -> T:
+        cached = await self.get(key)
+        if cached is not None:
+            return cached
+
+        async with self._lock:
+            item = self._data.get(key)
+            if item and not self._is_expired(item[0]):
+                return item[1]
+
+            inflight = self._inflight.get(key)
+            if inflight is None:
+                task = asyncio.create_task(factory())
+                self._inflight[key] = task
+            else:
+                task = inflight
+
+        try:
+            value = await task
+        finally:
+            async with self._lock:
+                self._inflight.pop(key, None)
+
+        await self.set(key, value)
+        return value
+
+    def clear(self):
+        """Очищает кэш (синхронный метод)"""
+        self._data.clear()
+        self._inflight.clear()
+
+        if self._persist_dir:
+            try:
+                for p in Path(self._persist_dir).glob("*.pickle"):
+                    p.unlink()
+            except Exception:
+                pass

app/client.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import re
+from typing import List, Tuple, Optional
+from urllib.parse import urljoin
+
+import httpx
+from bs4 import BeautifulSoup
+
+SSAU_BASE = "https://ssau.ru"
+
+
+class SsauClient:
+    """HTTP клиент для взаимодействия с ssau.ru"""
+
+    def __init__(self, timeout: float = 30.0):
+        self._client = httpx.AsyncClient(
+            headers={
+                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+                "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
+            },
+            timeout=httpx.Timeout(timeout),
+            follow_redirects=True,
+        )
+        self._csrf_token: Optional[str] = None
+
+    async def close(self):
+        await self._client.aclose()
+
+    async def fetch_html(self, url: str) -> str:
+        """Получает HTML страницы"""
+        print(f"Fetching URL: {url}")
+        response = await self._client.get(url)
+        response.raise_for_status()
+
+        if "ssau.ru/rasp" not in response.url.path:
+            print(f"Warning: Redirected to {response.url}")
+
+        return response.text
+
+    async def get_institutes(self) -> List[Tuple[int, str, str]]:
+        """Получает список институтов/факультетов"""
+        html = await self.fetch_html(urljoin(SSAU_BASE, "/rasp"))
+        soup = BeautifulSoup(html, "lxml")
+
+        institutes = []
+        for link in soup.select('a[href*="/rasp/faculty/"]'):
+            href = link.get("href", "")
+            match = re.search(r"/rasp/faculty/(\d+)", href)
+            if match:
+                inst_id = int(match.group(1))
+                name = link.get_text(strip=True)
+                if name and len(name) > 1:
+                    institutes.append((inst_id, name, urljoin(SSAU_BASE, href)))
+
+        return sorted(institutes, key=lambda x: x[1])
+
+    async def get_courses(self, institute_id: int) -> List[int]:
+        """Получает доступные курсы для института"""
+        url = urljoin(SSAU_BASE, f"/rasp/faculty/{institute_id}")
+        html = await self.fetch_html(url)
+        soup = BeautifulSoup(html, "lxml")
+
+        courses = set()
+        for link in soup.select('a[href*="course="], button[data-course]'):
+            href = link.get("href", "")
+            match = re.search(r"course=(\d+)", href)
+            if match:
+                course = int(match.group(1))
+                if 1 <= course <= 12:
+                    courses.add(course)
+
+            data_course = link.get("data-course")
+            if data_course:
+                try:
+                    course = int(data_course)
+                    if 1 <= course <= 12:
+                        courses.add(course)
+                except ValueError:
+                    pass
+
+        return sorted(courses) if courses else [1, 2, 3, 4]
+
+    async def get_groups(self, institute_id: int, course: int) -> List[Tuple[int, str]]:
+        """Получает группы для института и курса"""
+        url = urljoin(SSAU_BASE, f"/rasp/faculty/{institute_id}?course={course}")
+        html = await self.fetch_html(url)
+        soup = BeautifulSoup(html, "lxml")
+
+        groups = []
+        for link in soup.select('a[href*="groupId="]'):
+            href = link.get("href", "")
+            match = re.search(r"groupId=(\d+)", href)
+            if match:
+                group_id = int(match.group(1))
+                name = link.get_text(strip=True)
+                if name and len(name) > 1:
+                    groups.append((group_id, name))
+
+        if not groups:
+            for link in soup.select('[data-group-id]'):
+                group_id = link.get("data-group-id")
+                if group_id:
+                    name = link.get_text(strip=True)
+                    groups.append((int(group_id), name))
+
+        return sorted(groups, key=lambda x: x[1])
+
+    async def search_teachers(self, query: str) -> List[Tuple[int, str]]:
+        """Ищет преподавателей по имени"""
+        if len(query) < 2:
+            return []
+
+        if not self._csrf_token:
+            html = await self.fetch_html(urljoin(SSAU_BASE, "/rasp"))
+            match = re.search(r'csrf-token" content="([^"]+)"', html)
+            if match:
+                self._csrf_token = match.group(1)
+
+        if not self._csrf_token:
+            return []
+
+        try:
+            response = await self._client.post(
+                urljoin(SSAU_BASE, "/rasp/search"),
+                data={"text": query},
+                headers={
+                    "X-CSRF-TOKEN": self._csrf_token,
+                    "X-Requested-With": "XMLHttpRequest",
+                    "Content-Type": "application/x-www-form-urlencoded",
+                    "Referer": urljoin(SSAU_BASE, "/rasp"),
+                },
+            )
+
+            if response.status_code == 200:
+                data = response.json()
+                teachers = []
+                for item in data:
+                    url = item.get("url", "")
+                    if "staffId=" in url:
+                        match = re.search(r"staffId=(\d+)", url)
+                        if match:
+                            teachers.append((int(match.group(1)), item.get("text", "")))
+                return teachers
+        except Exception as e:
+            print(f"Search error: {e}")
+
+        return []