Compare commits

...

5 Commits

Author SHA1 Message Date
5b3c9dc00a Админка в master: секретный путь, ссылка, интервал, force, блокировка входа, история
Админка нужна на «голом» сервере, поэтому возвращаю её в основную ветку. Она
выключена, пока не заданы ADMIN_LOGIN/ADMIN_PASSWORD: маршруты /api/admin/* не
регистрируются, страницы нет — обычная установка ничего не замечает.

Панель (открывается по секретному пути ADMIN_PATH, кнопки входа на сайте нет,
вход по логину/паролю):
- состояние: группы, когда обновлялось, дата файла на Диске, изменился ли
  файл при последнем чтении (sha256), ошибки, путь файла настроек;
- новая ссылка на таблицу: проверяется скачиванием, при ошибке откат;
- интервал обновления (TTL) в минутах;
- «Обновить сейчас»: принудительно перечитывает таблицу (качает файл всегда,
  интервал — только если на Диске изменилась дата);
- история входов: последняя попытка (откуда, когда, успешна ли) и последний
  успешный вход — и в панели, и в логе службы.

Защита входа: ADMIN_MAX_ATTEMPTS неудач (по умолчанию 3), затем блокировка на
ADMIN_LOCKOUT_MINUTES минут (по умолчанию 15) — всё из .env.

Реализация:
- backend/admin.py: cookie-сессия на HMAC (ключ — пароль, 12 ч, HttpOnly,
  SameSite, Secure за https), сравнение за постоянное время, реальный IP из
  X-Forwarded-For, логирование каждой попытки;
- schedule/settings.py: ссылка, интервал и история входов в JSON; путь по
  умолчанию data/settings.json (в docker — том schedule_data), при
  недоступности файла настройки живут в памяти и панель предупреждает;
- cache.py: apply_link (с проверкой и откатом), force_reload, state();
- frontend: страница админки по ADMIN_PATH (вне static, noindex) и прокси
  /api/admin/* с телом и Set-Cookie;
- install.sh пишет переменные админки в .env, Dockerfile копирует admin.py.

Тесты: 156. Проверено локально (страница, 3 попытки -> блокировка, история) и
в собранном docker-образе (том настроек, запись истории).
2026-10-10 11:12:29 +03:00
962681630f Добавлен установщик 2026-10-07 17:28:55 +03:00
cc194b8eee Исправлен деплой скрипт 2026-10-07 16:20:42 +03:00
d897601d88 Скорректирован сценарий запуска, исправление бага с предметом 2026-10-07 15:17:37 +03:00
34b975da69 Перенос в master всего, кроме админки: ТОП-группы, порядок групп, запуск без Docker
Админка осталась отдельной ветвью admin-panel; в master её нет (ни кода, ни
страниц, ни переменных окружения, ни тома настроек).

Что перенесено:
- parsing: имена групп вида «ИВТ-ТОП-11БО» распознаются, порядок групп = порядок
  столбцов таблицы (похожие заголовки с пробелами группой не считаются);
- main: GET /groups отдаёт группы в порядке столбцов; интервал обновления из
  SCHEDULE_INTERVAL_MINUTES (со старым SCHEDULE_CACHE_TTL как fallback);
- cache: по интервалу сверяем modified — если файл не менялся, не скачиваем;
  при загрузке считаем sha256 и пишем в лог и /status, изменился ли файл;
- source: запросы к Яндекс.Диску с no-cache и меткой времени;
- frontend: ответы API с Cache-Control: no-store, общий .env из корня проекта
  (и backend, и frontend), страницы/прокси админки нет;
- Dockerfile/compose/.env.example: без ADMIN_* и без тома настроек; в api-образ
  попадают все модули backend (проверяется тестом test_deploy);
- Makefile: make help/venv/run/run-api/run-front/test + docker-цели — «make run»
  поднимает всё локально без Docker;
- README: быстрый старт, полный список зависимостей, запуск локально и в Docker;
- deploy/: пример nginx и инструкция «второй стенд рядом с основным».

Тесты: 114 (удалены админские), все зелёные.
2026-10-06 15:15:44 +03:00
25 changed files with 2727 additions and 233 deletions

View File

@@ -1,6 +1,30 @@
# --- расписание --------------------------------------------------------------
# Публичная ссылка на Яндекс.Диск с таблицей расписания (обязательна)
SCHEDULE_LINK= SCHEDULE_LINK=
# Как часто перечитывать таблицу, минут
SCHEDULE_INTERVAL_MINUTES=15
# Устаревшее: интервал в секундах (используется, только если не задан предыдущий)
SCHEDULE_CACHE_TTL=900 SCHEDULE_CACHE_TTL=900
# Пауза между повторными попытками, если источник недоступен, секунд
SCHEDULE_RETRY_AFTER=60 SCHEDULE_RETRY_AFTER=60
# --- порты (только для docker) -------------------------------------------------
# Оба порта публикуются на 127.0.0.1 (жёстко задано в docker-compose.yml):
# снаружи сервисы отдаёт nginx, напрямую в интернет они не смотрят.
API_PORT=8000 API_PORT=8000
FRONTEND_PORT=8080 FRONTEND_PORT=8080
SCHEDULE_API=http://api:8000
# --- админка (необязательно) ----------------------------------------------------
# Пока эти переменные пустые, админка выключена и наружу ничего не торчит.
# Секретный путь страницы админки: придумайте длинный, например /panel-8f3a91x7
ADMIN_PATH=
# Логин и пароль администратора
ADMIN_LOGIN=
ADMIN_PASSWORD=
# Сколько неудачных попыток входа допускается и на сколько минут блокировать после них
ADMIN_MAX_ATTEMPTS=3
ADMIN_LOCKOUT_MINUTES=15
# Где хранить настройки админки (ссылка, интервал, история входов).
# Пусто = data/settings.json в корне проекта; в docker задаётся автоматически.
SETTINGS_FILE=

6
.gitignore vendored
View File

@@ -6,3 +6,9 @@ __pycache__/
/.idea/ /.idea/
**/*.env **/*.env
.env .env
# локальные данные инструментов
.kilo/
# состояние админки (ссылка, интервал, история входов)
/data/

View File

@@ -7,11 +7,13 @@ WORKDIR /app
COPY requirements.txt ./ COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt \ RUN pip install --no-cache-dir -r requirements.txt \
&& groupadd --system app && useradd --system --gid app app && groupadd --system app && useradd --system --gid app app \
&& mkdir -p /data && chown app:app /data
FROM base AS api FROM base AS api
COPY --chown=app:app src/backend/main.py ./ COPY --chown=app:app src/backend/main.py ./
COPY --chown=app:app src/backend/admin.py ./
COPY --chown=app:app src/backend/schedule ./schedule COPY --chown=app:app src/backend/schedule ./schedule
USER app USER app
EXPOSE 8000 EXPOSE 8000
@@ -20,6 +22,7 @@ CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers",
FROM base AS frontend FROM base AS frontend
COPY --chown=app:app src/frontend/main.py frontend/main.py COPY --chown=app:app src/frontend/main.py frontend/main.py
COPY --chown=app:app src/frontend/admin.html frontend/admin.html
COPY --chown=app:app src/frontend/static frontend/static COPY --chown=app:app src/frontend/static frontend/static
USER app USER app
EXPOSE 8080 EXPOSE 8080

View File

@@ -1,32 +1,68 @@
.PHONY: build up down logs restart deploy ps dev test PY := $(CURDIR)/.venv/bin/python
PIP := $(CURDIR)/.venv/bin/pip
build: ## Собрать образы микросервисов .PHONY: help venv run dev run-api run-front test build up down restart logs ps deploy nginx-print nginx-install
help: ## Показать этот список команд
@grep -hE '^[a-zA-Z_-]+:.*?## ' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf " \033[36m%-10s\033[0m %s\n", $$1, $$2}'
# --- локальный запуск (без Docker) -------------------------------------------
venv: ## Создать .venv и установить зависимости
@test -x $(PY) || python3 -m venv .venv
@test -x $(PIP) || { echo "Не удалось создать .venv — см. README, раздел «Если установка не идёт» (на Debian/Ubuntu нужен пакет python3-venv)"; exit 1; }
@$(PIP) install -q --upgrade pip
@$(PIP) install -q -r requirements-dev.txt
@echo "Зависимости готовы (.venv). Дальше: make run"
run: venv ## Запустить API и сайт локально, без Docker
@echo "API: http://127.0.0.1:8000/status"
@echo "Сайт: http://127.0.0.1:8080 (открыть в браузере)"
@echo "Остановить: Ctrl+C"
@trap 'kill 0' EXIT; \
(cd src/backend && $(PY) -m uvicorn main:app --reload --port 8000) & \
(cd src/frontend && $(PY) -m uvicorn main:app --reload --port 8080) & \
wait
dev: run ## То же, что run
run-api: venv ## Запустить только API (http://127.0.0.1:8000)
cd src/backend && $(PY) -m uvicorn main:app --reload --port 8000
run-front: venv ## Запустить только сайт (http://127.0.0.1:8080)
cd src/frontend && $(PY) -m uvicorn main:app --reload --port 8080
test: venv ## Прогнать тесты
$(PY) -m pytest tests -q
# --- Docker (продакшен) ------------------------------------------------------
build: ## Собрать образы
docker compose build docker compose build
up: ## Запустить все микросервисы одной командой up: ## Запустить сервисы в фоне
docker compose up -d docker compose up -d
down: ## Остановить и удалить контейнеры down: ## Остановить и удалить контейнеры
docker compose down docker compose down
restart: ## Перезапустить микросервисы restart: ## Перезапустить сервисы
docker compose restart docker compose restart
logs: ## Логи всех микросервисов logs: ## Логи сервисов (Ctrl+C для выхода)
docker compose logs -f docker compose logs -f
ps: ## Статус микросервисов ps: ## Статус сервисов
docker compose ps docker compose ps
deploy: ## Собрать и запустить (деплой на сервере) deploy: ## Собрать и запустить (docker)
./deploy.sh ./deploy.sh
test: ## Прогнать тесты (локальный venv) # --- nginx (на сервере) ------------------------------------------------------
$(CURDIR)/.venv/bin/python -m pytest tests -q
dev: ## Запустить локально без Docker (бэкенд + фронт) # DOMAIN=sched.example.ru make nginx-print / nginx-install
@echo "Запуск бэкенда на :8000 и фронта на :8080..." nginx-print: ## Показать конфиг nginx (DOMAIN=домен), ничего не меняя
@trap 'kill 0' EXIT; \ deploy/nginx-site.sh --print --domain $(DOMAIN)
cd src/backend && $(CURDIR)/.venv/bin/python -m uvicorn main:app --reload --port 8000 & \
cd src/frontend && $(CURDIR)/.venv/bin/python -m uvicorn main:app --reload --port 8080 & \ nginx-install: ## Сгенерировать, включить и перечитать конфиг nginx (DOMAIN=домен, с sudo)
wait sudo deploy/nginx-site.sh --domain $(DOMAIN)

495
README.md
View File

@@ -1,8 +1,19 @@
# Расписание # Расписание
Публичное расписание учебных групп факультета: источник данных — Яндекс.Таблица Публичное расписание учебных групп. Источник данных — Яндекс.Таблица (xlsx на
университета (файл xlsx), который скачивается, разбирается, превращается в JSON Яндекс.Диске): файл скачивается по публичной ссылке, разбирается без сторонних
API и отображается на фронте. Деплой: `schedule.our-place.ru`. библиотек и отдаётся как JSON API и как сайт, который можно «установить» на
главный экран телефона (PWA).
Умеет:
- расписание по группам, преподавателям и аудиториям с разделением
числитель/знаменатель и бейджем текущей недели;
- занятость всех аудиторий по дням и поиск свободных на конкретную пару;
- совпадающие недели показываются одним списком, различающиеся — двумя
колонками, пустая — «Окно»;
- примечания из таблицы (красный текст) — отдельной строкой в карточке;
- PWA: манифест, service worker, иконки — «на главный экран».
## Архитектура ## Архитектура
@@ -10,20 +21,372 @@ API и отображается на фронте. Деплой: `schedule.our-p
Яндекс.Диск (xlsx) Яндекс.Диск (xlsx)
│ скачивание по публичной ссылке │ скачивание по публичной ссылке
▼ ▼
api (FastAPI), порт 8000, доступ только с localhost api — FastAPI, порт 8000
• скачивает и разбирает xlsx без сторонних библиотек • качает и разбирает xlsx стандартной библиотекой (zipfile + xml.etree)
• кэш в памяти: TTL + проверка даты изменения файла + stale-while-error • кэш в памяти: TTL + проверка даты изменения файла + stale-while-error
• один воркер, поэтому кэш общий • один воркер, поэтому кэш общий
▲ SCHEDULE_API=http://api:8000 (внутри docker-сети) ▲ SCHEDULE_API=http://api:8000 (внутри docker-сети)
│ SCHEDULE_API=http://127.0.0.1:8000 (локальная разработка) │
frontend (FastAPI: статика + прокси), прод :8090, dev :8080 frontend — FastAPI, порт из .env (по умолчанию 8080)
• отдаёт index.html / app.js / style.css / картинки • отдаёт index.html / app.js / style.css / иконки
• /api/* — тонкий прокси к api (браузер не знает адрес бэкенда) • /api/* — тонкий прокси к api (браузер не знает адрес бэкенда)
▲ ▲
nginx (https://schedule.our-place.ru) → 127.0.0.1:8090 nginx — порты 80/443, https://<домен сайта> → 127.0.0.1:<порт фронта>
``` ```
## Как устроено расписание в таблице Роли:
- **docker** изолирует оба сервиса, перезапускает их при падении
(`restart: unless-stopped`) и следит за здоровьем (healthcheck на
`/healthz`); frontend стартует, только когда api стал healthy.
- Оба порта публикуются **только на 127.0.0.1** (жёстко задано в
`docker-compose.yml`): из интернета контейнеры напрямую недоступны.
- **nginx** — единственная дверь из интернета: принимает https и проксирует на
фронт. Его конфиг генерирует `deploy/nginx-site.sh` (это делает установщик).
## Установка на сервер
Скачайте zip с проектом, распакуйте и запустите одну команду:
```bash
unzip schedule.zip && cd schedule
bash install.sh
```
Установка требует root — скрипт сам перезапустится через sudo. До запуска DNS
домена уже должен указывать на сервер, а порты 80/443 — быть открытыми:
сертификат Let's Encrypt выпускается прямо в процессе установки.
Скрипт спросит:
| Вопрос | Что ответить |
|---|---|
| Как ставить: Docker или систему | Docker — рекомендуется (разница ниже) |
| Публичная ссылка на таблицу | ссылка Я.Диска, единственная обязательная настройка |
| Домен сайта | например `sched.example.ru`; пусто = без nginx, сайт на 127.0.0.1 |
| Порт api и порт сайта | по умолчанию 8000 и 8080 |
| Пользователь служб (только для «в систему») | от кого работают сервисы |
| Почта (только если задан домен) | для уведомлений Let's Encrypt |
Ответы можно передать и флагами: `--mode`, `--link`, `--domain`, `--api-port`,
`--front-port`, `--user`, `--email`. `install.sh --dry-run` показывает план
установки, ничего не меняя.
### Docker или система — что выбрать
**Docker** — рекомендуется: сервисы изолированы, автоматически перезапускаются,
обновляются одной командой, не зависят от Python в системе. Установщик поставит
Docker (если нет), запишет `.env`, соберёт образы, запустит контейнеры и
дождётся их готовности.
**Система (без Docker)** — если Docker не нужен: установщик проверит Python
3.11+, создаст `.venv`, поставит зависимости и зарегистрирует две службы
systemd — `schedule-api` и `schedule-frontend` (автозапуск после ребута,
перезапуск при падении).
В обоих случаях, если домен задан, установщик дополнительно поставит nginx и
certbot, выпустит сертификат Let's Encrypt (webroot — продление автоматическое,
сайт не останавливается) и настроит конфиг: http → https, прокси на порт
фронта. Снаружи виден только nginx.
### Ручная установка (без скрипта)
Установщик делает ровно следующее — при желании те же шаги выполняются руками.
**Docker:**
```bash
cp .env.example .env && nano .env # SCHEDULE_LINK (+ порты при желании)
sudo ./deploy.sh # собрать и запустить
curl -s http://127.0.0.1:8080/api/status # {"has_data":true} — работает
```
**Система:**
```bash
sudo apt install -y python3 python3-venv python3-pip # Python 3.11+
python3 -m venv .venv
.venv/bin/pip install -r requirements.txt
cp .env.example .env && nano .env # SCHEDULE_LINK
```
Две службы systemd (`/etc/systemd/system/`), затем
`sudo systemctl daemon-reload && sudo systemctl enable --now schedule-api schedule-frontend`:
```ini
# schedule-api.service
[Unit]
Description=Schedule API (FastAPI)
After=network-online.target
Wants=network-online.target
[Service]
User=www-data
WorkingDirectory=/путь/к/проекту/src/backend
ExecStart=/путь/к/проекту/.venv/bin/python -m uvicorn main:app --host 127.0.0.1 --port 8000
Restart=always
RestartSec=3
[Install]
WantedBy=multi-user.target
# schedule-frontend.service
[Unit]
Description=Schedule frontend (FastAPI)
After=network-online.target schedule-api.service
Wants=network-online.target schedule-api.service
[Service]
User=www-data
WorkingDirectory=/путь/к/проекту/src/frontend
Environment=SCHEDULE_API=http://127.0.0.1:8000
ExecStart=/путь/к/проекту/.venv/bin/python -m uvicorn main:app --host 127.0.0.1 --port 8080
Restart=always
RestartSec=3
[Install]
WantedBy=multi-user.target
```
**nginx + сертификат** (общий шаг для обоих способов; `.env` уже должен быть):
```bash
sudo apt install -y nginx certbot
# временный http-конфиг домена с location /.well-known/acme-challenge/ → /var/www/html
sudo certbot certonly --webroot -w /var/www/html -d sched.example.ru
sudo deploy/nginx-site.sh --domain sched.example.ru --cert sched.example.ru --port 8080
curl -sI https://sched.example.ru/ | head -3 # HTTP/2 200 — готово
```
`deploy/nginx-site.sh` генерирует финальный конфиг (http → https, прокси на
`FRONTEND_PORT` из `.env`, таймауты под холодный кэш), включает его, проверяет
`nginx -t` и перечитывает nginx. Посмотреть конфиг, ничего не меняя:
`deploy/nginx-site.sh --print --domain sched.example.ru`.
### Обслуживание
- **Обновление**: Docker — `cd <каталог проекта> && sudo ./deploy.sh`;
система — заменить файлы проекта и
`sudo systemctl restart schedule-api schedule-frontend`.
- **Диагностика**: Docker — `sudo docker compose ps` и
`sudo docker logs api --tail 50`; система — `systemctl status schedule-api`
и `journalctl -u schedule-api -u schedule-frontend`. Состояние кэша —
`/api/status` (смотреть поле `error`).
- **Ротация логов контейнеров** — настроить сразу: без лимита логи растут
вечно и однажды забивают диск. `/etc/docker/daemon.json`:
```json
{ "log-driver": "json-file", "log-opts": { "max-size": "10m", "max-file": "3" } }
```
затем `sudo systemctl restart docker`.
- **Диск**: следить `df -h /`; раз в месяц
`sudo docker system prune -af && sudo journalctl --vacuum-size=200M`.
## Локальная разработка
Нужны Python 3.11+ и bash (Linux, macOS или WSL; на Windows без WSL — Docker).
```bash
python3 --version # должен быть 3.11 или новее
cp .env.example .env # вписать SCHEDULE_LINK
make run # создаст .venv, поставит зависимости, поднимет оба сервиса
```
Открыть <http://127.0.0.1:8080> (API — <http://127.0.0.1:8000>). Остановить —
`Ctrl+C`; отвязать от терминала — `nohup make run &`.
Данные скачиваются сразу на старте (несколько секунд) и дальше обновляются по
интервалу.
Полезные команды (`make help` — весь список):
| Команда | Что делает |
|---|---|
| `make venv` | создать `.venv` и установить зависимости |
| `make run` (или `make dev`) | запустить API и сайт |
| `make run-api` / `make run-front` | запустить только один из сервисов |
| `make test` | прогнать тесты |
| `make nginx-print` / `make nginx-install` | показать / установить конфиг nginx |
Без make — то же самое руками:
```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements-dev.txt
.venv/bin/python -m uvicorn main:app --reload --port 8000 # в каталоге src/backend
.venv/bin/python -m uvicorn main:app --reload --port 8080 # в каталоге src/frontend
```
### Тесты
```bash
make test # 121 тест, сеть не нужна
```
## Зависимости
**Системные требования**
| Что | Зачем |
|---|---|
| Python 3.11+ | локальная разработка и режим «в систему» |
| Docker + docker compose v2 | режим Docker (опционально) |
**Python-пакеты** — закреплены в `requirements.txt` (рабочие) и
`requirements-dev.txt` (тесты):
| Пакет | Версия | Назначение |
|---|---|---|
| fastapi | 0.141.1 | веб-фреймворк обоих сервисов |
| uvicorn | 0.52.4 | ASGI-сервер, которым запускаются сервисы |
| python-dotenv | 1.2.3 | чтение `.env` |
| requests | 2.34.2 | скачивание файла с Я.Диска и прокси к API |
| pytest | 9.1.1 | тесты — только dev |
| httpx2 | 2.12.0 | HTTP-клиент (наследник httpx) для `TestClient` — только dev |
Больше ничего не нужно: xlsx разбирается стандартной библиотекой, базы данных
нет, кэш — в памяти процесса. httpx2 в рабочем коде не используется и нужен
только тестам; модуль импортируется как `httpx2` — httpx перешёл под опеку
Pydantic и был переименован в августе 2026, `import httpx` больше не работает.
### Если установка не идёт
- Проверьте `python3 --version`: нужен 3.11 или новее. Если под `python3`
скрывается старая версия, создайте окружение нужной явно:
`python3.12 -m venv .venv`.
- Debian/Ubuntu: если `make venv` падает на создании окружения, доставьте
пакет venv: `sudo apt install python3-venv` (для нестандартной версии —
например, `python3.12-venv`).
### Если pip не находит пакеты: зеркала
Если pip настроен на корпоративное/национальное зеркало, оно может отставать
от PyPI — свежих пакетов там ещё нет. Диагностика и обход:
```bash
pip index versions httpx2 # есть ли пакет на текущем индексе
pip install -r requirements-dev.txt -i <зеркало> # разово через другое зеркало
pip config set global.index-url <зеркало> # закрепить зеркало навсегда
```
Зеркала, проверенные 07.10.2026 (пакеты `httpx2`/`httpcore2` на месте;
URL подставлять как есть в `-i` / `index-url`):
| Зеркало | URL |
|---|---|
| PyPI, официальный | `https://pypi.org/simple` |
| TUNA (Tsinghua) | `https://pypi.tuna.tsinghua.edu.cn/simple` |
| Huawei Cloud | `https://mirrors.huaweicloud.com/repository/pypi/simple` |
| BFSU | `https://mirrors.bfsu.edu.cn/pypi/web/simple` |
| SJTU | `https://mirror.sjtu.edu.cn/pypi/web/simple` |
| USTC | `https://mirrors.ustc.edu.cn/pypi/simple` |
| SUSTech | `https://mirrors.sustech.edu.cn/pypi/simple` |
Примечания: Alibaba Cloud (`https://mirrors.aliyun.com/pypi/simple`) обычно
тоже свежий, но из части сетей недоступен; Яндекс-зеркало (`mirror.yandex.ru`)
PyPI не раздаёт вовсе. Зеркала отстают по-разному, поэтому для свежих пакетов
надёжнее разово ставить с официального индекса.
## Переменные окружения
| Переменная | По умолчанию | Где | Описание |
|---|---|---|---|
| `SCHEDULE_LINK` | — | api | публичная ссылка Я.Диска, **обязательна** |
| `SCHEDULE_INTERVAL_MINUTES` | `15` | api | как часто перечитывать таблицу, минут |
| `SCHEDULE_CACHE_TTL` | `900` | api | устаревшее: интервал в секундах (если не задан предыдущий) |
| `SCHEDULE_RETRY_AFTER` | `60` | api | пауза между попытками при сбое источника |
| `SCHEDULE_API` | `http://127.0.0.1:8000` | frontend | адрес бэкенда (в docker: `http://api:8000`) |
| `SCHEDULE_PROXY_TIMEOUT` | `150` | frontend | таймаут прокси к бэкенду, сек |
| `ADMIN_PATH` | — | api+frontend | секретный путь админки; пусто = админка выключена |
| `ADMIN_LOGIN` / `ADMIN_PASSWORD` | — | api | логин и пароль администратора |
| `ADMIN_MAX_ATTEMPTS` | `3` | api | сколько неудачных попыток входа допускается |
| `ADMIN_LOCKOUT_MINUTES` | `15` | api | на сколько минут блокировать вход после них |
| `SETTINGS_FILE` | `data/settings.json` | api | где хранить ссылку, интервал и историю входов |
| `API_PORT` | `8000` | docker | внешний порт api (публикуется на 127.0.0.1) |
| `FRONTEND_PORT` | `8080` | docker | внешний порт frontend (публикуется на 127.0.0.1) |
`.env` ищется вверх от каталога модуля: и backend, и frontend находят общий
`.env` в корне репозитория; `load_dotenv` не перезаписывает уже выставленные
переменные окружения.
## Админка
Необязательная страница для управления источником данных. Отдельного входа на
сайте нет: панель открывается по секретному пути `ADMIN_PATH`, а пока
`ADMIN_LOGIN`/`ADMIN_PASSWORD` пусты, админка выключена и наружу ничего не
торчит (маршруты `/api/admin/*` не регистрируются).
Включить: заполнить в `.env` четыре переменные и перезапустить службу.
```ini
ADMIN_PATH=/panel-8f3a91x7 # придумайте длинный и неочевидный
ADMIN_LOGIN=admin
ADMIN_PASSWORD=<длинный пароль>
```
Что умеет:
- **посмотреть состояние**: сколько групп, когда обновлялось, дата файла на
Диске, изменился ли файл при последнем чтении (`sha256`), есть ли ошибки;
- **вставить новую ссылку** на таблицу: она проверяется скачиванием и разбором
до сохранения — нерабочая не применится, данные на сайте не пострадают;
- **задать интервал обновления** в минутах (`1…1440`);
- **обновить сейчас**: скачивает таблицу принудительно, не дожидаясь интервала
(по интервалу файл перечитывается, только если на Диске изменилась дата);
- **видеть входы**: последняя попытка (откуда, когда, успешна ли) и последний
успешный вход — и в панели, и в логе службы.
Защита входа: после `ADMIN_MAX_ATTEMPTS` неудач (по умолчанию 3) вход с этого
адреса блокируется на `ADMIN_LOCKOUT_MINUTES` минут (по умолчанию 15). Сессия —
подписанная cookie на 12 часов.
Настройки (ссылка, интервал, история входов) хранятся в `SETTINGS_FILE`
(по умолчанию `data/settings.json` в корне проекта; в docker — в томе
`schedule_data`). Если файл недоступен для записи, изменения живут в памяти до
перезапуска, а панель предупредит об этом.
## Структура кода
```
install.sh интерактивная установка на сервер: Docker или systemd + nginx
src/
backend/
main.py FastAPI: маршруты, create_app(cache), /healthz
admin.py админка: cookie-сессия, блокировка входа, ссылка, интервал, force
schedule/
xlsx.py чтение xlsx: общие строки, объединённые ячейки,
скрытые столбцы, красный текст, серийные даты
source.py скачивание файла по публичной ссылке Я.Диска
lessons.py разбор ячеек занятий (общий конвейер ролей)
parsing.py лист → расписание, недели, индексы преподавателей/аудиторий
cache.py ScheduleCache: блокировка, интервал, stale-while-error, force
settings.py хранилище настроек админки (ссылка, интервал, история входов)
errors.py ScheduleError
frontend/
main.py FastAPI: статика + прокси /api/* (таймаут 150 c)
admin.html страница админки (вне static: прямой доступ закрыт)
static/
index.html, app.js, style.css
manifest.json PWA: имя, иконки, standalone-режим
sw.js минимальный service worker (без кэша) для установки приложения
img/ логотипы и иконки приложения
tools/
make_icons.py генерация иконок приложения из логотипа (нужен rsvg-convert)
deploy/
nginx-site.sh генерация и включение конфига nginx (используется install.sh)
tests/
conftest.py общие фикстуры
xlsx_builder.py генерация xlsx в памяти (без openpyxl)
fixtures.py общая тестовая таблица
test_lessons.py все форматы ячеек
test_xlsx.py чтение книги
test_parsing.py разбор листов, недели, индексы
test_api.py API и поведение кэша (интервал/stale, 404/400/502)
test_admin.py админка: вход, блокировка, история, ссылка/интервал/force
test_settings.py хранилище настроек админки
test_deploy.py проверка, что образы копируют все нужные файлы
```
Данные админки (`data/settings.json`) — рабочие, в git не попадают.
## Как устроена таблица-источник
Лист «Бакалавры и магистры» — это сетка: Лист «Бакалавры и магистры» — это сетка:
@@ -67,7 +430,10 @@ nginx (https://schedule.our-place.ru) → 127.0.0.1:8090
В одной ячейке может быть несколько занятий: они разделяются пустой строкой, В одной ячейке может быть несколько занятий: они разделяются пустой строкой,
новым маркером «к.в.» или началом нового предмета сразу после строки новым маркером «к.в.» или началом нового предмета сразу после строки
аудитории/преподавателя. аудитории/преподавателя. Если пустой строкой (или красной заметкой, снятой с
целой строки) от предмета оторвался только преподаватель и/или аудитория, этот
кусок достраивает предыдущее занятие, а не создаёт занятие без предмета — иначе
в расписании аудитории оставались бы одни группы.
Итог по паре: `{time, ч: {raw, lessons[]}, з: {raw, lessons[]}}`, где Итог по паре: `{time, ч: {raw, lessons[]}, з: {raw, lessons[]}}`, где
`lesson = {marker, subject, teacher, room, note}`. `lesson = {marker, subject, teacher, room, note}`.
@@ -86,40 +452,6 @@ nginx (https://schedule.our-place.ru) → 127.0.0.1:8090
- Приложение можно «установить» на главный экран (манифест + service - Приложение можно «установить» на главный экран (манифест + service
worker + иконки приложения); на iOS иконку берёт `apple-touch-icon`. worker + иконки приложения); на iOS иконку берёт `apple-touch-icon`.
## Структура кода
```
src/
backend/
main.py FastAPI: маршруты, create_app(cache), /healthz
schedule/
xlsx.py чтение xlsx: общие строки, объединённые ячейки,
скрытые столбцы, красный текст, серийные даты
source.py скачивание файла по публичной ссылке Я.Диска
lessons.py разбор ячеек занятий (общий конвейер ролей)
parsing.py лист → расписание, недели, индексы преподавателей/аудиторий
cache.py ScheduleCache: блокировка, TTL, modified, stale-while-error
errors.py ScheduleError
frontend/
main.py FastAPI: статика + прокси /api/* (таймаут 150 c)
static/
index.html, app.js, style.css
manifest.json PWA: имя, иконки, standalone-режим
sw.js минимальный service worker (без кэша) для установки приложения
img/ логотипы (owl.svg, owl-white.svg), иконки приложения
(app-icon.svg, icon-192/512.png, apple-touch-icon.png, favicon-32.png)
tools/
make_icons.py генерация иконок приложения из логотипа (нужен rsvg-convert)
tests/
conftest.py общие фикстуры
xlsx_builder.py генерация xlsx в памяти (без openpyxl)
fixtures.py общая тестовая таблица
test_lessons.py все форматы ячеек
test_xlsx.py чтение книги
test_parsing.py разбор листов, недели, индексы
test_api.py API и поведение кэша (TTL/modified/stale, 404/400/502)
```
## API ## API
Полный список методов отдаёт сам бэкенд: `GET /`. Полный список методов отдаёт сам бэкенд: `GET /`.
@@ -137,79 +469,10 @@ tests/
| GET | `/rooms-day?day={день}` | занятость всех аудиторий за день | | GET | `/rooms-day?day={день}` | занятость всех аудиторий за день |
| GET | `/rooms-free?day={день}&period={N}` | свободные аудитории | | GET | `/rooms-free?day={день}&period={N}` | свободные аудитории |
| GET | `/week` | недели семестра и текущая чётность | | GET | `/week` | недели семестра и текущая чётность |
| GET | `/status` | состояние кэша (`cache_ttl_seconds`, `last_updated`, `table_modified`, `error`) | | GET | `/status` | состояние кэша (`last_updated`, `table_modified`, `error`, …) |
| GET | `/healthz` | живость процесса (не обращается к таблице) | | GET | `/healthz` | живость процесса (не обращается к таблице) |
Фронт проксирует только то, что использует, под префиксом `/api/...`: Фронт проксирует только то, что использует, под префиксом `/api/...`:
`groups`, `schedule/{group}`, `teachers`, `teacher/{name}`, `rooms`, `groups`, `schedule/{group}`, `teachers`, `teacher/{name}`, `rooms`,
`room/{label}`, `rooms-day`, `rooms-free`, `week`, `status`. `room/{label}`, `rooms-day`, `rooms-free`, `week`, `status`.
Остальное (`today`, `schedule/{group}/{day}`) доступно только на бэкенде. Остальное (`today`, `schedule/{group}/{day}`) доступно только на бэкенде.
## Переменные окружения
| Переменная | По умолчанию | Где | Описание |
|---|---|---|---|
| `SCHEDULE_LINK` | — | api | публичная ссылка Я.Диска, **обязательна** |
| `SCHEDULE_CACHE_TTL` | `900` | api | время жизни кэша, сек |
| `SCHEDULE_RETRY_AFTER` | `60` | api | пауза между попытками при сбое источника |
| `SCHEDULE_API` | `http://127.0.0.1:8000` | frontend | адрес бэкенда (в docker: `http://api:8000`) |
| `SCHEDULE_PROXY_TIMEOUT` | `150` | frontend | таймаут прокси к бэкенду, сек |
| `API_PORT` | `8000` | docker | внешний порт api |
| `FRONTEND_PORT` | `8080` | docker | внешний порт frontend (на проде: `127.0.0.1:8090`) |
`.env` ищется вверх от каталога модуля (backend — до корня репозитория);
`load_dotenv` не перезаписывает уже выставленные переменные окружения.
## Запуск
### Продакшен (Docker)
```bash
cp .env.example .env # вписать SCHEDULE_LINK (и порты при необходимости)
./deploy.sh # = docker compose build && up -d
```
У обоих сервисов есть healthcheck на `/healthz`; frontend стартует после того,
как api стал healthy. Обновление: `git pull && docker compose up -d --build`.
### Разработка
```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements-dev.txt
make dev # api :8000 и фронт :8080, оба с --reload
```
### Тесты
```bash
make test # 104 теста, сеть не нужна
```
## Развёртывание на сервере
Код хранится в Gitea (`git.our-place.ru`, web через nginx, git-ssh на порту
2222) и живёт в `/opt/schedule`:
```bash
# первый запуск
cd /opt/schedule
# .env: SCHEDULE_LINK=... , API_PORT=127.0.0.1:8000, FRONTEND_PORT=127.0.0.1:8090
sudo docker compose up -d --build
# обновление
cd /opt/schedule && git pull && sudo docker compose up -d --build
```
nginx-сайт `schedule.our-place.ru` проксирует 443 → 127.0.0.1:8090 (фронт),
сертификат Let's Encrypt. Порты api и frontend слушаются только на loopback.
## Замечания
- Принудительное обновление таблицы (пропуск кэша, `force`) из API намеренно
убрано; планируется вернуть вместе с авторизацией и ролью администратора.
- Кэш живёт в памяти единственного воркера: при горизонтальном масштабировании
api его придётся выносить в общее хранилище.
- Строки внутри дня в таблице строго чередуются «числитель / знаменатель».
Если строку посреди блока вставят или удалят, бэкенд запишет предупреждение
«нечётное число строк, недели могли сдвинуться».

123
deploy/nginx-site.sh Executable file
View File

@@ -0,0 +1,123 @@
#!/usr/bin/env bash
# Генерирует и включает конфиг nginx для сайта расписания.
#
# SSL обеспечивает сертификат родительского домена (обычно wildcard *.example.ru):
# отдельный certbot для поддомена не нужен — достаточно имени сертификата.
# Список имён: sudo ls /etc/letsencrypt/live/
#
# Имя сертификата берётся из --cert, иначе — родительский домен от --domain
# (sched.example.ru -> example.ru). Порт фронта берётся из --port, иначе из
# FRONTEND_PORT файла .env в корне проекта, иначе 8080.
#
# Примеры:
# deploy/nginx-site.sh --print --domain sched.example.ru # показать конфиг
# sudo deploy/nginx-site.sh --domain sched.example.ru # поставить и перечитать nginx
# sudo deploy/nginx-site.sh --domain sched.example.ru --cert example.ru
set -euo pipefail
usage() {
grep '^#' "$0" | sed 's/^# \{0,1\}//' | tail -n +2
}
die() { echo "Ошибка: $*" >&2; exit 2; }
DOMAIN=""
CERT_NAME=""
FRONTEND_PORT=""
PRINT=0
while [[ $# -gt 0 ]]; do
case "$1" in
--domain) DOMAIN="$2"; shift 2 ;;
--cert) CERT_NAME="$2"; shift 2 ;;
--port) FRONTEND_PORT="$2"; shift 2 ;;
--print) PRINT=1; shift ;;
-h|--help) usage; exit 0 ;;
*) echo "Неизвестный аргумент: $1 (см. --help)" >&2; exit 2 ;;
esac
done
[[ -n $DOMAIN ]] || die "Укажите домен: --domain sched.example.ru (см. --help)"
[[ $DOMAIN =~ ^[a-zA-Z0-9.-]+$ ]] || die "Кривое имя домена: $DOMAIN"
if [[ -z $CERT_NAME ]]; then
CERT_NAME="${DOMAIN#*.}" # sched.example.ru -> example.ru (wildcard родителя)
fi
[[ $CERT_NAME =~ ^[a-zA-Z0-9.-]+$ ]] || { echo "Кривое имя сертификата: $CERT_NAME" >&2; exit 2; }
if [[ -z $FRONTEND_PORT ]]; then
# FRONTEND_PORT из .env; значение вида «127.0.0.1:8090» трактуем как «порт после двоеточия»
FRONTEND_PORT=$(
grep -E '^FRONTEND_PORT=' .env 2>/dev/null | tail -1 | cut -d= -f2- \
| tr -d ' "' | awk -F: '{print $NF}'
) || FRONTEND_PORT=""
fi
FRONTEND_PORT=${FRONTEND_PORT:-8080}
[[ $FRONTEND_PORT =~ ^[0-9]+$ ]] || { echo "Не разобрал FRONTEND_PORT из .env, задайте --port" >&2; exit 2; }
read -r -d '' CONFIG <<'TEMPLATE' || true
# Сгенерировано deploy/nginx-site.sh — правки вносить в скрипт, не сюда.
server {
listen 80;
listen [::]:80;
server_name __DOMAIN__;
location /.well-known/acme-challenge/ {
root /var/www/html;
}
location / {
return 301 https://$host$request_uri;
}
}
server {
listen 443 ssl http2;
listen [::]:443 ssl http2;
server_name __DOMAIN__;
# сертификат родительского домена (обычно wildcard), обновляет certbot родителя
ssl_certificate /etc/letsencrypt/live/__CERT__/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/__CERT__/privkey.pem;
ssl_protocols TLSv1.2 TLSv1.3;
client_max_body_size 8m;
location / {
# порт FRONTEND_PORT из .env стенда
proxy_pass http://127.0.0.1:__PORT__;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto https;
# холодный кэш: бэкенд качает таблицу, до ~2 минут
proxy_read_timeout 300s;
proxy_send_timeout 300s;
}
}
TEMPLATE
CONFIG=${CONFIG//__DOMAIN__/$DOMAIN}
CONFIG=${CONFIG//__CERT__/$CERT_NAME}
CONFIG=${CONFIG//__PORT__/$FRONTEND_PORT}
if [[ $PRINT == 1 ]]; then
printf '%s\n' "$CONFIG"
exit 0
fi
if [[ $EUID -ne 0 ]]; then
echo "Установка требует root: sudo deploy/nginx-site.sh (посмотреть конфиг — deploy/nginx-site.sh --print)" >&2
exit 1
fi
DEST="/etc/nginx/sites-available/${DOMAIN}"
printf '%s\n' "$CONFIG" > "$DEST"
ln -sf "$DEST" "/etc/nginx/sites-enabled/${DOMAIN}"
if nginx -t 2>&1; then
systemctl reload nginx
echo "Готово: https://${DOMAIN} -> 127.0.0.1:${FRONTEND_PORT} (сертификат ${CERT_NAME})"
else
echo "nginx -t не прошёл: конфиг записан в ${DEST} и включён, но nginx НЕ перечитан. Разберитесь и выполните: sudo nginx -t && sudo systemctl reload nginx" >&2
exit 1
fi

View File

@@ -5,10 +5,22 @@ services:
target: api target: api
environment: environment:
SCHEDULE_LINK: ${SCHEDULE_LINK} SCHEDULE_LINK: ${SCHEDULE_LINK}
SCHEDULE_INTERVAL_MINUTES: ${SCHEDULE_INTERVAL_MINUTES:-15}
SCHEDULE_CACHE_TTL: ${SCHEDULE_CACHE_TTL:-900} SCHEDULE_CACHE_TTL: ${SCHEDULE_CACHE_TTL:-900}
SCHEDULE_RETRY_AFTER: ${SCHEDULE_RETRY_AFTER:-60} SCHEDULE_RETRY_AFTER: ${SCHEDULE_RETRY_AFTER:-60}
# админка: пусто = выключена (страница и /api/admin/* не появятся)
ADMIN_PATH: ${ADMIN_PATH:-}
ADMIN_LOGIN: ${ADMIN_LOGIN:-}
ADMIN_PASSWORD: ${ADMIN_PASSWORD:-}
ADMIN_MAX_ATTEMPTS: ${ADMIN_MAX_ATTEMPTS:-3}
ADMIN_LOCKOUT_MINUTES: ${ADMIN_LOCKOUT_MINUTES:-15}
# настройки админки (ссылка, интервал, история входов) живут в томе
SETTINGS_FILE: ${SETTINGS_FILE:-/data/settings.json}
volumes:
- schedule_data:/data
ports: ports:
- "${API_PORT:-8000}:8000" # api нужен только фронтенду внутри docker-сети; наружу — 127.0.0.1 для отладки
- "127.0.0.1:${API_PORT:-8000}:8000"
restart: unless-stopped restart: unless-stopped
healthcheck: healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/healthz', timeout=5)"] test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/healthz', timeout=5)"]
@@ -22,9 +34,13 @@ services:
context: . context: .
target: frontend target: frontend
environment: environment:
# внутри compose api доступен по имени сервиса; менять не нужно
SCHEDULE_API: ${SCHEDULE_API:-http://api:8000} SCHEDULE_API: ${SCHEDULE_API:-http://api:8000}
# страница админки отдаётся фронтендом по этому пути
ADMIN_PATH: ${ADMIN_PATH:-}
ports: ports:
- "${FRONTEND_PORT:-8080}:8080" # фронт обслуживает nginx; наружу напрямую не смотрит
- "127.0.0.1:${FRONTEND_PORT:-8080}:8080"
restart: unless-stopped restart: unless-stopped
depends_on: depends_on:
api: api:
@@ -35,3 +51,6 @@ services:
timeout: 5s timeout: 5s
retries: 3 retries: 3
start_period: 10s start_period: 10s
volumes:
schedule_data:

324
install.sh Executable file
View File

@@ -0,0 +1,324 @@
#!/usr/bin/env bash
# Интерактивная установка «Расписания» на чистый сервер.
#
# Запуск из распакованного каталога проекта (нужен root — скрипт сам
# перезапустится через sudo):
#
# bash install.sh
#
# Скрипт спросит: как ставить (Docker или прямо в систему), публичную ссылку
# на таблицу и домен сайта. Затем поставит всё нужное, запустит сервисы и
# настроит nginx с сертификатом Let's Encrypt.
#
# Для автоматизации ответы можно передать аргументами:
# --mode docker|system способ установки
# --link URL публичная ссылка Яндекс.Диска с таблицей
# --domain DOMAIN домен сайта (пусто = без nginx, сайт на 127.0.0.1)
# --api-port N порт api (по умолчанию 8000)
# --front-port N порт сайта (по умолчанию 8080)
# --user NAME пользователь служб (режим «в систему»)
# --email ADDR почта для уведомлений Let's Encrypt
# --dry-run показать план установки, ничего не меняя
set -euo pipefail
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
log() { printf '\033[1;36m==>\033[0m %s\n' "$*"; }
err() { printf '\033[31m==>\033[0m %s\n' "$*" >&2; }
die() { err "$*"; exit 1; }
usage() { grep '^#' "$0" | sed 's/^# \{0,1\}//' | tail -n +2; }
MODE="" LINK="" DOMAIN="" DOMAIN_SET=0 API_PORT="" FRONT_PORT="" RUN_USER="" EMAIL="" DRY_RUN=0
while [[ $# -gt 0 ]]; do
case "$1" in
--mode) MODE="$2"; shift 2 ;;
--link) LINK="$2"; shift 2 ;;
--domain) DOMAIN="$2"; DOMAIN_SET=1; shift 2 ;;
--api-port) API_PORT="$2"; shift 2 ;;
--front-port) FRONT_PORT="$2"; shift 2 ;;
--user) RUN_USER="$2"; shift 2 ;;
--email) EMAIL="$2"; shift 2 ;;
--dry-run) DRY_RUN=1; shift ;;
-h|--help) usage; exit 0 ;;
*) die "Неизвестный аргумент: $1 (см. --help)" ;;
esac
done
# ask "Вопрос" [по_умолчанию] — ответ в $REPLY
ask() {
local def="${2:-}" input
if ! read -rp "$1${def:+ [$def]}: " input; then
die "Нужен интерактивный ввод — запустите в терминале или передайте все параметры флагами (см. --help)"
fi
REPLY="${input:-$def}"
}
# --- вопросы -------------------------------------------------------------------
if [[ -z $MODE ]]; then
log "Как ставить?"
log " 1) Docker — изолированные контейнеры, автоперезапуск, обновление одной командой (рекомендуется)"
log " 2) Система — без Docker: venv + службы systemd"
while true; do
ask "Способ установки (docker/system)" "docker"
REPLY=$(echo "$REPLY" | tr '[:upper:]' '[:lower:]' | xargs)
if [[ $REPLY == 1 ]]; then
REPLY=docker
elif [[ $REPLY == 2 ]]; then
REPLY=system
fi
[[ $REPLY =~ ^(docker|system)$ ]] && break || err "Ответьте docker или system"
done
MODE=$REPLY
fi
[[ $MODE =~ ^(docker|system)$ ]] || die "--mode: только docker или system"
if [[ -z $LINK ]]; then
while true; do
ask "Публичная ссылка на таблицу (Яндекс.Диск)"
REPLY=$(echo "$REPLY" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//' | tr -d '"')
[[ -n $REPLY ]] && break || err "Без ссылки приложение не сможет скачать расписание"
done
LINK=$REPLY
fi
if [[ $DOMAIN_SET == 0 && -z $DOMAIN ]]; then
ask "Домен сайта, например sched.example.ru (пусто = без nginx, сайт только на 127.0.0.1)"
DOMAIN=$(echo "$REPLY" | tr '[:upper:]' '[:lower:]' | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')
fi
if [[ -n $DOMAIN && ! $DOMAIN =~ ^[a-z0-9.-]+$ ]]; then die "Кривое имя домена: $DOMAIN"; fi
if [[ -z $API_PORT ]]; then
ask "Порт api (нужен только для диагностики)" "8000"
API_PORT=$REPLY
fi
if [[ -z $FRONT_PORT ]]; then
ask "Порт сайта" "8080"
FRONT_PORT=$REPLY
fi
[[ $API_PORT =~ ^[0-9]+$ && $FRONT_PORT =~ ^[0-9]+$ ]] || die "Порты должны быть числами"
if [[ $MODE == system ]]; then
if [[ -z $RUN_USER ]]; then
ask "Пользователь, от которого будут работать службы" "${SUDO_USER:-$(id -un)}"
RUN_USER=$REPLY
fi
id "$RUN_USER" >/dev/null 2>&1 || die "Пользователь $RUN_USER не существует"
fi
if [[ -n $DOMAIN && -z $EMAIL ]]; then
ask "Почта для уведомлений Let's Encrypt" "admin@$DOMAIN"
EMAIL=$REPLY
fi
if [[ $DRY_RUN == 1 ]]; then
log "План установки (dry-run, ничего не выполнено):"
echo " режим: $MODE"
echo " таблица: $LINK"
echo " домен: ${DOMAIN:-(не задан — nginx и https пропускаются, сайт на 127.0.0.1)}"
echo " порты: api $API_PORT, сайт $FRONT_PORT"
if [[ $MODE == system ]]; then
echo " службы от: $RUN_USER"
fi
if [[ -n $DOMAIN ]]; then
echo " сертификат: Let's Encrypt для $DOMAIN (почта: $EMAIL)"
fi
exit 0
fi
# --- права ----------------------------------------------------------------------
if [[ $EUID -ne 0 ]]; then
if [[ ${INSTALL_NO_SUDO:-0} == 1 ]]; then
err "Запущено без root (INSTALL_NO_SUDO=1) — часть шагов может не пройти"
else
log "Нужны права root — перезапускаюсь через sudo"
exec sudo bash "$0" "$@"
fi
fi
# --- помощники ------------------------------------------------------------------
pkg_install() {
local missing=() p
for p in "$@"; do command -v "$p" >/dev/null 2>&1 || missing+=("$p"); done
if [[ ${#missing[@]} -eq 0 ]]; then
return 0
fi
log "Ставлю пакеты: ${missing[*]}"
if command -v apt-get >/dev/null 2>&1; then
apt-get update -y
DEBIAN_FRONTEND=noninteractive apt-get install -y "${missing[@]}"
elif command -v dnf >/dev/null 2>&1; then
dnf install -y "${missing[@]}"
else
die "Не нашёл apt-get/dnf — поставьте вручную: ${missing[*]}"
fi
}
wait_health() {
local url="$1" name="$2" i
printf '%s' "==> Жду $name ($url)"
for i in $(seq 1 60); do
if curl -sf -m 2 "$url" >/dev/null 2>&1; then echo " — готово"; return 0; fi
printf '.'; sleep 1
done
echo
die "$name не отвечает на $url — смотрите логи (docker logs / journalctl)"
}
write_env() {
cd "$ROOT"
if [[ -f .env && ! -f .env.install-backup ]]; then
cp .env .env.install-backup
fi
cat > .env <<EOF
# Сгенерировано install.sh
SCHEDULE_LINK=$LINK
SCHEDULE_INTERVAL_MINUTES=15
SCHEDULE_RETRY_AFTER=60
API_PORT=$API_PORT
FRONTEND_PORT=$FRONT_PORT
# Админка (необязательно). Заполните path/логин/пароль, чтобы включить панель:
# страница будет доступна только по https://<домен><ADMIN_PATH>
ADMIN_PATH=
ADMIN_LOGIN=
ADMIN_PASSWORD=
ADMIN_MAX_ATTEMPTS=3
ADMIN_LOCKOUT_MINUTES=15
EOF
log ".env записан (прежний, если был, сохранён в .env.install-backup)"
}
install_docker() {
if command -v docker >/dev/null 2>&1; then
log "Docker уже установлен"
else
log "Ставлю Docker..."
curl -fsSL https://get.docker.com | sh
fi
if [[ $EUID -eq 0 ]]; then
systemctl enable --now docker
fi
}
setup_nginx() {
pkg_install nginx certbot curl
if [[ ! -d /etc/nginx/sites-available ]]; then
die "Нужен nginx с поддержкой sites-available (Debian/Ubuntu); на других системах настройте конфиг по README"
fi
log "Временный http-конфиг $DOMAIN — для выпуска сертификата"
cat > "/etc/nginx/sites-available/$DOMAIN" <<'NGINX'
# Сгенерировано install.sh: временный конфиг для certbot, потом заменяется
server {
listen 80;
listen [::]:80;
server_name __DOMAIN__;
root /var/www/html;
location /.well-known/acme-challenge/ { }
}
NGINX
sed -i "s/__DOMAIN__/$DOMAIN/" "/etc/nginx/sites-available/$DOMAIN"
ln -sf "/etc/nginx/sites-available/$DOMAIN" "/etc/nginx/sites-enabled/$DOMAIN"
nginx -t && systemctl reload nginx
log "Выпускаю сертификат Let's Encrypt для $DOMAIN (webroot, без остановки nginx)"
certbot certonly --webroot -w /var/www/html -d "$DOMAIN" \
--non-interactive --agree-tos -m "$EMAIL"
log "Финальный конфиг сайта: https + прокси на 127.0.0.1:$FRONT_PORT"
"$ROOT/deploy/nginx-site.sh" --domain "$DOMAIN" --cert "$DOMAIN" --port "$FRONT_PORT"
}
# --- установка ------------------------------------------------------------------
if [[ $MODE == docker ]]; then
install_docker
write_env
log "Собираю и запускаю контейнеры (первая сборка — несколько минут)..."
(cd "$ROOT" && docker compose up -d --build)
wait_health "http://127.0.0.1:$FRONT_PORT/healthz" "frontend"
else
log "Проверяю Python..."
pkg_install python3 python3-pip
if command -v apt-get >/dev/null 2>&1; then
apt-get install -y python3-venv # на Debian/Ubuntu venv без этого пакета не работает
fi
python3 - <<'PY' || die "Нужен Python 3.11 или новее (см. README, «Зависимости»)"
import sys
sys.exit(0 if sys.version_info >= (3, 11) else 1)
PY
write_env
log "Создаю виртуальное окружение и ставлю зависимости..."
[[ -d "$ROOT/.venv" ]] || python3 -m venv "$ROOT/.venv"
"$ROOT/.venv/bin/pip" install -q --upgrade pip
"$ROOT/.venv/bin/pip" install -q -r "$ROOT/requirements.txt"
log "Настраиваю службы systemd..."
cat > /etc/systemd/system/schedule-api.service <<EOF
[Unit]
Description=Schedule API (FastAPI)
After=network-online.target
Wants=network-online.target
[Service]
User=$RUN_USER
WorkingDirectory=$ROOT/src/backend
ExecStart=$ROOT/.venv/bin/python -m uvicorn main:app --host 127.0.0.1 --port $API_PORT
Restart=always
RestartSec=3
[Install]
WantedBy=multi-user.target
EOF
cat > /etc/systemd/system/schedule-frontend.service <<EOF
[Unit]
Description=Schedule frontend (FastAPI)
After=network-online.target schedule-api.service
Wants=network-online.target schedule-api.service
[Service]
User=$RUN_USER
WorkingDirectory=$ROOT/src/frontend
Environment=SCHEDULE_API=http://127.0.0.1:$API_PORT
ExecStart=$ROOT/.venv/bin/python -m uvicorn main:app --host 127.0.0.1 --port $FRONT_PORT
Restart=always
RestartSec=3
[Install]
WantedBy=multi-user.target
EOF
chown -R "$RUN_USER" "$ROOT"
systemctl daemon-reload
systemctl enable --now schedule-api.service schedule-frontend.service
wait_health "http://127.0.0.1:$API_PORT/healthz" "api"
wait_health "http://127.0.0.1:$FRONT_PORT/healthz" "frontend"
fi
if [[ -n $DOMAIN ]]; then
setup_nginx
else
log "Домен не задан — nginx и https пропущены. Сайт работает на http://127.0.0.1:$FRONT_PORT"
fi
# --- итог -----------------------------------------------------------------------
log "Установка завершена."
if [[ -n $DOMAIN ]]; then
echo " Сайт: https://$DOMAIN"
else
echo " Сайт: http://127.0.0.1:$FRONT_PORT (с самого сервера)"
fi
echo " Таблица: $LINK"
echo " Настройки: $ROOT/.env"
if [[ $MODE == docker ]]; then
echo " Логи: sudo docker logs api / frontend"
echo " Обновление: cd $ROOT && sudo docker compose up -d --build"
echo " Напоминание: настройте ротацию логов Docker (/etc/docker/daemon.json), см. README"
else
echo " Службы: sudo systemctl status schedule-api schedule-frontend"
echo " Логи: journalctl -u schedule-api -u schedule-frontend"
echo " Обновление: заменить файлы проекта, затем sudo systemctl restart schedule-api schedule-frontend"
fi

View File

@@ -1,3 +1,2 @@
-r requirements.txt
pytest==9.1.1 pytest==9.1.1
httpx2==2.12.0 httpx2==2.12.0

278
src/backend/admin.py Normal file
View File

@@ -0,0 +1,278 @@
"""Админка расписания.
Отдельного входа на сайте нет: панель открывается по секретному пути из
ADMIN_PATH, логин и пароль задаются в ADMIN_LOGIN/ADMIN_PASSWORD. Если они
пусты, админка выключена и маршруты /api/admin/* не регистрируются.
После нескольких неудачных попыток вход блокируется на время из
ADMIN_LOCKOUT_MINUTES (по умолчанию: 3 попытки и 15 минут). Каждая попытка
запоминается (откуда, когда, успешна ли) — это видно в панели и пишется в лог.
Что умеет администратор:
GET /api/admin/state — состояние кэша, ссылка, интервал, история входов;
POST /api/admin/link — проверить и поставить новую ссылку на таблицу;
POST /api/admin/interval — интервал обновления в минутах;
POST /api/admin/refresh — принудительно перечитать таблицу сейчас.
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import json
import logging
import threading
import time
from dataclasses import dataclass
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from pydantic import BaseModel
from schedule import ScheduleCache, ScheduleError
from schedule.settings import SettingsStore
log = logging.getLogger(__name__)
COOKIE_NAME = "schedule_admin"
SESSION_TTL_SECONDS = 12 * 3600
DEFAULT_MAX_ATTEMPTS = 3
DEFAULT_LOCKOUT_MINUTES = 15
@dataclass
class AdminConfig:
"""Настройки админки: секретный путь, учётные данные и защита от перебора."""
path: str = ""
login: str = ""
password: str = ""
max_attempts: int = DEFAULT_MAX_ATTEMPTS
lockout_minutes: int = DEFAULT_LOCKOUT_MINUTES
@property
def enabled(self) -> bool:
return bool(self.login and self.password)
def _b64e(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def _b64d(text: str) -> bytes:
return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
def _same(left: str, right: str) -> bool:
"""Сравнение строк за постоянное время (compare_digest не умеет не-ASCII)."""
return hmac.compare_digest(left.encode("utf-8"), right.encode("utf-8"))
def make_token(login: str, password: str, ttl: int = SESSION_TTL_SECONDS) -> str:
payload = json.dumps({"sub": login, "exp": int(time.time()) + ttl}, separators=(",", ":")).encode()
sig = hmac.new(password.encode(), payload, hashlib.sha256).digest()
return f"{_b64e(payload)}.{_b64e(sig)}"
def verify_token(token: str, password: str) -> bool:
try:
payload_b64, sig_b64 = token.split(".", 1)
payload = _b64d(payload_b64)
expected = hmac.new(password.encode(), payload, hashlib.sha256).digest()
if not hmac.compare_digest(expected, _b64d(sig_b64)):
return False
return int(json.loads(payload).get("exp", 0)) > time.time()
except (ValueError, TypeError):
return False
class LoginThrottle:
"""Счётчик неудачных попыток: N промахов — пауза на lockout_minutes."""
def __init__(self, max_attempts: int = DEFAULT_MAX_ATTEMPTS, lockout_minutes: int = DEFAULT_LOCKOUT_MINUTES) -> None:
self.max_attempts = max(1, int(max_attempts))
self.lockout_seconds = max(1, int(lockout_minutes)) * 60
self._lock = threading.Lock()
self._failures: dict[str, int] = {}
self._blocked_until: dict[str, float] = {}
def locked_until(self, key: str) -> float | None:
with self._lock:
until = self._blocked_until.get(key)
if until is None:
return None
if until <= time.time():
self._blocked_until.pop(key, None)
self._failures.pop(key, None)
return None
return until
def attempts_left(self, key: str) -> int:
with self._lock:
return max(0, self.max_attempts - self._failures.get(key, 0))
def register_failure(self, key: str) -> None:
with self._lock:
failures = self._failures.get(key, 0) + 1
if failures >= self.max_attempts:
self._blocked_until[key] = time.time() + self.lockout_seconds
self._failures[key] = 0
else:
self._failures[key] = failures
def reset(self, key: str) -> None:
with self._lock:
self._failures.pop(key, None)
self._blocked_until.pop(key, None)
class LoginBody(BaseModel):
login: str
password: str
class LinkBody(BaseModel):
link: str
class IntervalBody(BaseModel):
interval_minutes: int
def _client_ip(request: Request) -> str:
"""Реальный адрес клиента: за nginx он приходит в X-Forwarded-For."""
forwarded = request.headers.get("x-forwarded-for", "")
if forwarded:
return forwarded.split(",")[0].strip() or "неизвестно"
return request.client.host if request.client else "неизвестно"
def _human_dt(iso: str | None) -> str:
if not iso:
return "—"
try:
return datetime.fromisoformat(iso).strftime("%d.%m.%Y %H:%M:%S")
except ValueError:
return iso
def create_admin_router(
cache: ScheduleCache,
settings: SettingsStore | None,
config: AdminConfig,
) -> APIRouter:
router = APIRouter(prefix="/api/admin", tags=["admin"])
throttle = LoginThrottle(config.max_attempts, config.lockout_minutes)
def require_auth(request: Request) -> None:
token = request.cookies.get(COOKIE_NAME, "")
if not token or not verify_token(token, config.password):
raise HTTPException(status_code=401, detail="Требуется вход в админку")
def _secure(request: Request) -> bool:
proto = request.headers.get("x-forwarded-proto", "") or request.url.scheme
return proto.split(",")[0].strip() == "https"
def _remember(ip: str, ok: bool) -> None:
if settings is not None:
settings.record_login(ip, ok)
@router.post("/login")
def login(body: LoginBody, request: Request, response: Response) -> dict:
ip = _client_ip(request)
until = throttle.locked_until(ip)
if until is not None:
minutes = max(1, round((until - time.time()) / 60))
log.warning("Вход в админку заблокирован (%s), осталось ~%d мин", ip, minutes)
raise HTTPException(
status_code=429,
detail=f"Слишком много неудачных попыток. Вход заблокирован ещё на ~{minutes} мин.",
)
ok_login = _same(body.login.strip(), config.login)
ok_password = _same(body.password, config.password)
if not (ok_login and ok_password):
throttle.register_failure(ip)
_remember(ip, False)
left = throttle.attempts_left(ip)
if throttle.locked_until(ip) is not None:
log.warning(
"Неудачная попытка входа (%s): исчерпаны попытки, пауза %d мин",
ip,
config.lockout_minutes,
)
raise HTTPException(
status_code=429,
detail=f"Неверный логин или пароль. Попытки исчерпаны — вход заблокирован на {config.lockout_minutes} мин.",
)
log.warning("Неудачная попытка входа (%s), осталось попыток: %d", ip, left)
raise HTTPException(
status_code=401,
detail=f"Неверный логин или пароль. Осталось попыток: {left}.",
)
throttle.reset(ip)
previous = settings.last_attempt if settings is not None else None
_remember(ip, True)
if previous:
log.info(
"Успешный вход в админку (%s). Предыдущая попытка: %s, %s, %s",
ip,
_human_dt(previous.get("at")),
previous.get("ip"),
"успешная" if previous.get("ok") else "неудачная",
)
else:
log.info("Успешный вход в админку (%s)", ip)
response.set_cookie(
COOKIE_NAME,
make_token(config.login, config.password),
max_age=SESSION_TTL_SECONDS,
httponly=True,
samesite="lax",
secure=_secure(request),
path="/",
)
return {"ok": True}
@router.post("/logout")
def logout(response: Response) -> dict:
response.delete_cookie(COOKIE_NAME, path="/")
return {"ok": True}
@router.get("/state", dependencies=[Depends(require_auth)])
def state(request: Request) -> dict:
ip = _client_ip(request)
until = throttle.locked_until(ip)
payload = cache.state()
payload["throttle"] = {
"max_attempts": config.max_attempts,
"lockout_minutes": config.lockout_minutes,
"attempts_left": throttle.attempts_left(ip),
"locked_until": datetime.fromtimestamp(until).astimezone().isoformat() if until else None,
}
return payload
@router.post("/refresh", dependencies=[Depends(require_auth)])
def refresh() -> dict:
try:
return cache.force_reload()
except ScheduleError as exc:
raise HTTPException(status_code=502, detail=str(exc))
@router.post("/link", dependencies=[Depends(require_auth)])
def set_link(body: LinkBody) -> dict:
try:
return cache.apply_link(body.link)
except ScheduleError as exc:
raise HTTPException(status_code=400, detail=str(exc))
@router.post("/interval", dependencies=[Depends(require_auth)])
def set_interval(body: IntervalBody) -> dict:
try:
minutes = cache.set_interval_minutes(body.interval_minutes)
except ScheduleError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {"interval_minutes": minutes}
return router

View File

@@ -1,27 +1,33 @@
"""API-приложение расписания (FastAPI).""" """API-приложение расписания (FastAPI).
Кроме публичных данных здесь живёт админка (backend/admin.py): она включается
только если заданы ADMIN_LOGIN и ADMIN_PASSWORD, и открывается по секретному
пути ADMIN_PATH. Все параметры — переменные окружения, см. .env.example.
"""
from __future__ import annotations from __future__ import annotations
import logging import logging
import os import os
import threading
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from datetime import date from datetime import date
from pathlib import Path from pathlib import Path
import uvicorn import uvicorn
from dotenv import load_dotenv from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Query from fastapi import FastAPI, HTTPException
from admin import AdminConfig, create_admin_router
from schedule import ScheduleCache, ScheduleError from schedule import ScheduleCache, ScheduleError
from schedule import parsing from schedule import parsing
from schedule.settings import SettingsStore, interval_from_env
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
def _find_dotenv() -> Path | None: def _find_dotenv() -> Path | None:
"""Ищем .env вверх от каталога модуля (локальная разработка).""" """Ищем .env вверх от каталога модуля (общий .env в корне проекта)."""
path = Path(__file__).resolve().parent path = Path(__file__).resolve().parent
for _ in range(5): for _ in range(5):
env = path / ".env" env = path / ".env"
@@ -37,40 +43,56 @@ _dotenv_path = _find_dotenv()
if _dotenv_path: if _dotenv_path:
load_dotenv(_dotenv_path) load_dotenv(_dotenv_path)
SCHEDULE_LINK = (os.getenv("SCHEDULE_LINK") or "").strip()
CACHE_TTL = float(os.getenv("SCHEDULE_CACHE_TTL", "900")) def _normalize_admin_path(raw: str | None) -> str:
RETRY_AFTER = float(os.getenv("SCHEDULE_RETRY_AFTER", "60")) """Секретный путь панели: всегда с ведущим слэшем и без хвостового."""
path = (raw or "").strip()
if not path:
return ""
if not path.startswith("/"):
path = "/" + path
return path.rstrip("/") or "/"
def _default_cache() -> ScheduleCache: def _int_env(name: str, default: int) -> int:
raw = (os.getenv(name) or "").strip()
if not raw:
return default
try: try:
return ScheduleCache(SCHEDULE_LINK, ttl_seconds=CACHE_TTL, retry_after=RETRY_AFTER) return int(raw)
except ScheduleError: except ValueError:
log.error("SCHEDULE_LINK не задан — API данных работать не будет") log.warning("%s=%r не число, беру %d", name, raw, default)
return _BrokenCache() return default
class _BrokenCache(ScheduleCache): SCHEDULE_LINK = (os.getenv("SCHEDULE_LINK") or "").strip()
"""Кэш без ссылки: все запросы данных отвечают понятной ошибкой.""" RETRY_AFTER = float(os.getenv("SCHEDULE_RETRY_AFTER", "60"))
SETTINGS_FILE = (os.getenv("SETTINGS_FILE") or "").strip()
ADMIN_PATH = _normalize_admin_path(os.getenv("ADMIN_PATH"))
ADMIN_LOGIN = (os.getenv("ADMIN_LOGIN") or "").strip()
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD") or ""
ADMIN_MAX_ATTEMPTS = _int_env("ADMIN_MAX_ATTEMPTS", 3)
ADMIN_LOCKOUT_MINUTES = _int_env("ADMIN_LOCKOUT_MINUTES", 15)
def __init__(self) -> None:
self._link = None
self._ttl = CACHE_TTL
self._retry_after = RETRY_AFTER
self._lock = threading.RLock()
self._data = None
self._weeks = []
self._index = None
self._day_times = {}
self._teacher_full = {}
self._room_info = {}
self._loaded_at = 0.0
self._modified = None
self._error = "Не задана переменная окружения SCHEDULE_LINK"
self._last_failed = 0.0
def _ensure_fresh(self) -> None: def default_settings() -> SettingsStore:
raise ScheduleError(self._error) return SettingsStore(
link=SCHEDULE_LINK,
interval_minutes=interval_from_env(),
path=SETTINGS_FILE or None,
)
def default_cache(settings: SettingsStore | None = None) -> ScheduleCache:
settings = settings or default_settings()
if SCHEDULE_LINK and settings.link and settings.link != SCHEDULE_LINK:
log.warning(
"Ссылка в настройках (%s) отличается от SCHEDULE_LINK из окружения (%s); "
"работаем по ссылке из настроек — её видно и можно поменять в админке",
settings.link,
SCHEDULE_LINK,
)
return ScheduleCache(settings=settings, retry_after=RETRY_AFTER)
def _day_arg(day: str) -> str: def _day_arg(day: str) -> str:
@@ -83,8 +105,21 @@ def _day_arg(day: str) -> str:
return normalized return normalized
def create_app(cache: ScheduleCache | None = None) -> FastAPI: def create_app(
cache = cache or _default_cache() cache: ScheduleCache | None = None,
admin: AdminConfig | None = None,
settings: SettingsStore | None = None,
) -> FastAPI:
if cache is None:
settings = settings or default_settings()
cache = ScheduleCache(settings=settings, retry_after=RETRY_AFTER)
admin = admin if admin is not None else AdminConfig(
path=ADMIN_PATH,
login=ADMIN_LOGIN,
password=ADMIN_PASSWORD,
max_attempts=ADMIN_MAX_ATTEMPTS,
lockout_minutes=ADMIN_LOCKOUT_MINUTES,
)
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
@@ -97,6 +132,16 @@ def create_app(cache: ScheduleCache | None = None) -> FastAPI:
lifespan=lifespan, lifespan=lifespan,
) )
app.state.cache = cache app.state.cache = cache
app.state.admin = admin
if admin.enabled:
app.include_router(create_admin_router(cache, settings, admin))
if admin.path:
log.info("Админка включена (путь задан в ADMIN_PATH)")
else:
log.warning("Админка включена, но ADMIN_PATH не задан — страница панели недоступна")
else:
log.info("Админка выключена (не заданы ADMIN_LOGIN/ADMIN_PASSWORD)")
def _data() -> tuple[dict, list[dict]]: def _data() -> tuple[dict, list[dict]]:
try: try:
@@ -128,7 +173,7 @@ def create_app(cache: ScheduleCache | None = None) -> FastAPI:
return { return {
"title": "Расписание", "title": "Расписание",
"endpoints": { "endpoints": {
"/groups": "Список групп", "/groups": "Список групп (в порядке столбцов таблицы)",
"/schedule/{group}": "Расписание группы", "/schedule/{group}": "Расписание группы",
"/schedule/{group}/{day}": "Расписание группы на день", "/schedule/{group}/{day}": "Расписание группы на день",
"/today?group={group}": "Расписание группы на сегодня", "/today?group={group}": "Расписание группы на сегодня",
@@ -145,11 +190,9 @@ def create_app(cache: ScheduleCache | None = None) -> FastAPI:
@app.get("/groups") @app.get("/groups")
def groups() -> list[str]: def groups() -> list[str]:
"""Группы в том порядке, в каком идут столбцы в таблице."""
data, _ = _data() data, _ = _data()
return sorted( return list(data.keys())
data.keys(),
key=lambda g: (0 if g.endswith("БО") else (1 if g.endswith("МО") else 2), g),
)
@app.get("/teachers") @app.get("/teachers")
def teachers() -> list[str]: def teachers() -> list[str]:

View File

@@ -2,15 +2,19 @@
- однопоточная перезагрузка (lock): параллельные запросы во время загрузки - однопоточная перезагрузка (lock): параллельные запросы во время загрузки
ждут один download вместо N скачиваний; ждут один download вместо N скачиваний;
- TTL, а также проверка modified таблицы: если таблица не менялась, - интервал обновления: по его истечении сверяем modified таблицы — если файл
скачивание не повторяется; не менялся, скачивание не повторяется; принудительное обновление из админки
(force_reload) перечитывает таблицу всегда;
- stale-while-error: если таблица перестала скачиваться, отдаём последнее - stale-while-error: если таблица перестала скачиваться, отдаём последнее
удачное состояние и повторяем попытку не чаще, чем раз в retry_after секунд удачное состояние и повторяем попытку не чаще, чем раз в retry_after секунд
(вместо скачивания на каждый запрос). (вместо скачивания на каждый запрос);
- настройки администратора (ссылка, интервал) живут в SettingsStore и
применяются на лету: смена ссылки проверяется до сохранения.
""" """
from __future__ import annotations from __future__ import annotations
import hashlib
import logging import logging
import threading import threading
import time import time
@@ -18,24 +22,32 @@ from datetime import datetime
from . import parsing from . import parsing
from .errors import ScheduleError from .errors import ScheduleError
from .settings import SettingsStore
from .source import download_schedule, fetch_table_modified from .source import download_schedule, fetch_table_modified
from .xlsx import parse_workbook from .xlsx import parse_workbook
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
NO_LINK_ERROR = "Ссылка на расписание не задана"
class ScheduleCache: class ScheduleCache:
def __init__( def __init__(
self, self,
link: str | None, link: str | None = None,
ttl_seconds: float = 900.0, ttl_seconds: float = 900.0,
retry_after: float = 60.0, retry_after: float = 60.0,
settings: SettingsStore | None = None,
) -> None: ) -> None:
if not link: self._settings = settings
if settings is not None:
link = settings.link or link or ""
ttl_seconds = settings.interval_minutes * 60
self._link = (link or "").strip()
if not self._link and settings is None:
raise ScheduleError("Не задана переменная окружения SCHEDULE_LINK") raise ScheduleError("Не задана переменная окружения SCHEDULE_LINK")
self._link = link self._ttl = float(ttl_seconds)
self._ttl = ttl_seconds self._retry_after = float(retry_after)
self._retry_after = retry_after
self._lock = threading.RLock() self._lock = threading.RLock()
self._data: dict | None = None self._data: dict | None = None
@@ -46,8 +58,12 @@ class ScheduleCache:
self._room_info: dict = {} self._room_info: dict = {}
self._loaded_at: float = 0.0 self._loaded_at: float = 0.0
self._modified: str | None = None self._modified: str | None = None
self._error: str | None = None self._error: str | None = None if self._link else NO_LINK_ERROR
self._last_failed: float = 0.0 self._last_failed: float = 0.0
# следим, менялось ли содержимое скачанного файла
self._content_hash: str | None = None
self._changed: bool = False
self._changed_at: float = 0.0
# --- публичное API --------------------------------------------------------- # --- публичное API ---------------------------------------------------------
@@ -59,6 +75,11 @@ class ScheduleCache:
def room_info(self) -> dict: def room_info(self) -> dict:
return self._room_info return self._room_info
@property
def interval_minutes(self) -> int:
with self._lock:
return max(1, round(self._ttl / 60))
def get(self) -> tuple[dict, list[dict]]: def get(self) -> tuple[dict, list[dict]]:
"""(data, weeks); бросает ScheduleError, если данных нет совсем.""" """(data, weeks); бросает ScheduleError, если данных нет совсем."""
with self._lock: with self._lock:
@@ -79,21 +100,79 @@ class ScheduleCache:
return self._day_times return self._day_times
def status(self) -> dict: def status(self) -> dict:
"""Состояние кэша (публичный /status): без ссылки и без админских данных."""
with self._lock: with self._lock:
return { return self._status()
"cache_ttl_seconds": self._ttl,
"last_updated": self._iso(self._loaded_at), def state(self) -> dict:
"table_modified": self._modified, """Состояние для админки: то же плюс текущая ссылка и настройки."""
"error": self._error, with self._lock:
} state = self._status()
state["link"] = self._link
if self._settings is not None:
state["settings_file"] = str(self._settings.path) if self._settings.path else None
state["settings_save_error"] = self._settings.save_error
state["last_attempt"] = self._settings.last_attempt
state["last_success"] = self._settings.last_success
return state
def warmup(self) -> None: def warmup(self) -> None:
"""Первый запуск данных на старте приложения (ошибка не фатальна).""" """Первая загрузка данных на старте приложения (ошибка не фатальна)."""
try: try:
self.get() self.get()
except ScheduleError as exc: except ScheduleError as exc:
log.warning("Начальная загрузка расписания не удалась: %s", exc) log.warning("Начальная загрузка расписания не удалась: %s", exc)
# --- действия администратора ------------------------------------------------
def force_reload(self) -> dict:
"""Принудительно обновить расписание: скачиваем таблицу в любом случае."""
with self._lock:
now = time.time()
modified = fetch_table_modified(self._link) if self._link else None
try:
self._load(now, modified)
except ScheduleError as exc:
self._error = str(exc)
self._last_failed = now
log.warning("Принудительное обновление не удалось: %s", exc)
raise
return self.state()
def apply_link(self, link: str) -> dict:
"""Проверить новую ссылку и, если она рабочая, переключить кэш на неё."""
link = (link or "").strip()
if not link:
raise ScheduleError("Ссылка на расписание не может быть пустой")
with self._lock:
previous_link, previous_modified = self._link, self._modified
self._link = link
self._modified = None
now = time.time()
modified = fetch_table_modified(link)
try:
self._load(now, modified)
except ScheduleError as exc:
# ссылка не подошла — возвращаемся к прежней, данные не теряем
self._link = previous_link
self._modified = previous_modified
self._last_failed = now
if self._data is None:
self._error = str(exc)
raise ScheduleError(f"Ссылка не подошла: {exc}") from exc
if self._settings is not None:
self._settings.set_link(link)
return self.state()
def set_interval_minutes(self, minutes: int | str) -> int:
if self._settings is None:
raise ScheduleError("Настройки не подключены: интервал задаётся в .env")
value = self._settings.set_interval_minutes(minutes)
with self._lock:
self._ttl = value * 60
log.info("Интервал обновления расписания: %d мин", value)
return value
# --- internals -------------------------------------------------------------- # --- internals --------------------------------------------------------------
@staticmethod @staticmethod
@@ -102,6 +181,20 @@ class ScheduleCache:
return None return None
return datetime.fromtimestamp(ts, datetime.now().astimezone().tzinfo).isoformat() return datetime.fromtimestamp(ts, datetime.now().astimezone().tzinfo).isoformat()
def _status(self) -> dict:
return {
"cache_ttl_seconds": self._ttl,
"interval_minutes": max(1, round(self._ttl / 60)),
"last_updated": self._iso(self._loaded_at),
"table_modified": self._modified,
"error": self._error,
"groups": len(self._data) if self._data else 0,
"has_data": self._data is not None,
"changed": self._changed,
"changed_at": self._iso(self._changed_at),
"content_hash": (self._content_hash or "")[:12],
}
def _ensure_fresh(self) -> None: def _ensure_fresh(self) -> None:
now = time.time() now = time.time()
fresh = ( fresh = (
@@ -112,9 +205,9 @@ class ScheduleCache:
) )
if fresh: if fresh:
return return
self._reload(now) self._refresh(now)
def _reload(self, now: float) -> None: def _refresh(self, now: float) -> None:
# служим старые данные, пока не истекло время повторной попытки # служим старые данные, пока не истекло время повторной попытки
if ( if (
self._data is not None self._data is not None
@@ -122,6 +215,12 @@ class ScheduleCache:
and now - self._last_failed < self._retry_after and now - self._last_failed < self._retry_after
): ):
return return
if not self._link:
self._fail(ScheduleError(NO_LINK_ERROR), now)
return
# Обновление по интервалу: сначала сверяем modified — если файл не
# менялся, скачивание не повторяем. Принудительное обновление (кнопка
# в админке) идёт через force_reload и качает файл всегда.
modified = fetch_table_modified(self._link) modified = fetch_table_modified(self._link)
if ( if (
self._data is not None self._data is not None
@@ -133,31 +232,49 @@ class ScheduleCache:
self._error = None self._error = None
return return
try: try:
log.info("Загружаю расписание по ссылке %s ...", self._link) self._load(now, modified)
started = time.time()
xlsx = download_schedule(self._link)
sheets = parse_workbook(xlsx)
data = parsing.parse_schedule(sheets)
weeks = parsing.extract_week_ranges(sheets)
index = parsing.build_teacher_room_index(data)
self._data = data
self._weeks = weeks
self._index = index
self._day_times = parsing.build_day_times(data)
self._teacher_full = parsing.parse_teachers_full(sheets)
self._room_info = parsing.parse_rooms_info(sheets)
self._modified = modified
self._loaded_at = now
self._error = None
log.info(
"Расписание загружено: %d групп за %.1f с", len(data), time.time() - started
)
except ScheduleError as exc: except ScheduleError as exc:
self._fail(exc, now) self._fail(exc, now)
except Exception as exc: # непредвиденное в парсинге/сети except Exception as exc: # непредвиденное в парсинге/сети
log.exception("Ошибка при загрузке расписания") log.exception("Ошибка при загрузке расписания")
self._fail(ScheduleError(f"Не удалось обработать таблицу: {exc}"), now) self._fail(ScheduleError(f"Не удалось обработать таблицу: {exc}"), now)
def _load(self, now: float, modified: str | None) -> None:
"""Скачать и разобрать таблицу, заменить состояние. Бросает ScheduleError."""
if not self._link:
raise ScheduleError(NO_LINK_ERROR)
log.info("Загружаю расписание по ссылке %s ...", self._link)
started = time.time()
xlsx = download_schedule(self._link)
digest = hashlib.sha256(xlsx).hexdigest()
first_load = self._content_hash is None
changed = not first_load and digest != self._content_hash
sheets = parse_workbook(xlsx)
data = parsing.parse_schedule(sheets)
weeks = parsing.extract_week_ranges(sheets)
index = parsing.build_teacher_room_index(data)
self._data = data
self._weeks = weeks
self._index = index
self._day_times = parsing.build_day_times(data)
self._teacher_full = parsing.parse_teachers_full(sheets)
self._room_info = parsing.parse_rooms_info(sheets)
self._modified = modified
self._loaded_at = now
self._error = None
self._content_hash = digest
self._changed = changed
if changed or first_load:
self._changed_at = now
log.info(
"Расписание загружено: %d групп за %.1f с; файл %s (sha256 %s)",
len(data),
time.time() - started,
"изменился" if changed else ("первая загрузка" if first_load else "не изменился"),
digest[:12],
)
def _fail(self, exc: ScheduleError, now: float) -> None: def _fail(self, exc: ScheduleError, now: float) -> None:
self._error = str(exc) self._error = str(exc)
self._last_failed = now self._last_failed = now

View File

@@ -495,6 +495,40 @@ def _normalize_marker_of(line: str) -> Optional[str]:
return _normalize_marker(num or None) return _normalize_marker(num or None)
def _merge_detached_fragments(lessons: list[Lesson]) -> list[Lesson]:
"""Достраивает занятие из оторвавшихся пустой строкой преподавателя/аудитории.
Пустая строка внутри ячейки (или красная заметка, снятая с целой строки)
отрывает хвост от предмета: «Предмет\nЛарина Ю.А.\n\n ауд.210» разбирается
как занятие без аудитории и «ауд. 210» без предмета — в индексе аудиторий
от такого занятия остаётся только список групп. Кусок без предмета цепляем
к предыдущему занятию, но только если у того этих полей ещё нет, иначе
это похоже на отдельное занятие параллельной подгруппы.
"""
merged: list[Lesson] = []
for lesson in lessons:
prev = merged[-1] if merged else None
if (
prev is not None
and prev.subject
and not lesson.subject
and (lesson.teacher or lesson.room)
and (not lesson.marker or lesson.marker == prev.marker)
and (not lesson.teacher or not prev.teacher)
and (not lesson.room or not prev.room)
):
notes = [n for n in (prev.note + "\n" + lesson.note).split("\n") if n]
merged[-1] = replace(
prev,
teacher=lesson.teacher or prev.teacher,
room=lesson.room or prev.room,
note="\n".join(dict.fromkeys(notes)),
)
else:
merged.append(lesson)
return merged
def parse_lesson_cell(cell_text: str) -> list[Lesson]: def parse_lesson_cell(cell_text: str) -> list[Lesson]:
"""Текст ячейки -> список занятий. """Текст ячейки -> список занятий.
@@ -525,6 +559,8 @@ def parse_lesson_cell(cell_text: str) -> list[Lesson]:
lesson = replace(lesson, marker=last_marker) lesson = replace(lesson, marker=last_marker)
lessons.append(lesson) lessons.append(lesson)
lessons = _merge_detached_fragments(lessons)
if not cell_notes: if not cell_notes:
return lessons return lessons
if not lessons: if not lessons:

View File

@@ -35,7 +35,9 @@ _COL_DAY = 4 # день недели
_COL_PERIOD = 5 # пара и время _COL_PERIOD = 5 # пара и время
_FIRST_GROUP_COL = 6 _FIRST_GROUP_COL = 6
_GROUP_RE = re.compile(r"^[А-ЯЁ]{2,4}-\d") # Имя группы: буквенные части через дефис, номер потока и необязательный
# суффикс: «ИВТ-11БО», «ПИЭ-13МО», «ИВТ-ТОП-11БО», «ИТ-ТОП-12БО».
_GROUP_RE = re.compile(r"^[А-ЯЁ]+(?:-[А-ЯЁ]+)*-\d{1,2}[А-ЯЁ]{0,3}$")
_PERIOD_RE = re.compile(r"^\s*(\d{1,2})\s*(.*)$") _PERIOD_RE = re.compile(r"^\s*(\d{1,2})\s*(.*)$")
_ROOM_RE = re.compile(r"ауд\.\s*([\w\-]+)", re.I) _ROOM_RE = re.compile(r"ауд\.\s*([\w\-]+)", re.I)
_SEMESTER_DATE_RE = re.compile(r"(\d{1,2}\.\d{1,2}\.\d{2})") _SEMESTER_DATE_RE = re.compile(r"(\d{1,2}\.\d{1,2}\.\d{2})")
@@ -108,13 +110,16 @@ def current_parity(week_ranges: list[dict], today: date | None = None) -> str |
def _find_group_columns(header: list[str], hidden_cols: set[int]) -> dict[int, str]: def _find_group_columns(header: list[str], hidden_cols: set[int]) -> dict[int, str]:
"""Столбцы групп: имя из заголовка, порядок — как столбцы в таблице."""
columns: dict[int, str] = {} columns: dict[int, str] = {}
seen: set[str] = set() seen: set[str] = set()
for idx, cell in enumerate(header): for idx, cell in enumerate(header):
if idx in hidden_cols: if idx in hidden_cols:
continue continue
name = _normalize_header(cell) name = _normalize_header(cell)
if name and _GROUP_RE.match(name) and name not in seen: if not name or name in seen or len(name) > 24:
continue
if _GROUP_RE.match(name):
columns[idx] = name columns[idx] = name
seen.add(name) seen.add(name)
return columns return columns

View File

@@ -0,0 +1,221 @@
"""Настройки, которые правит администратор.
Хранятся в JSON-файле (`SETTINGS_FILE`, по умолчанию `data/settings.json` в
корне проекта), поэтому переживают перезапуск службы: ссылка на таблицу,
интервал обновления и история входов в админку. Если файл недоступен для
записи — настройки живут в памяти, а сервис продолжает работать (об этом
пишется предупреждение в лог и в состояние админки).
"""
from __future__ import annotations
import json
import logging
import os
import threading
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from .errors import ScheduleError
log = logging.getLogger(__name__)
DEFAULT_INTERVAL_MINUTES = 15
MIN_INTERVAL_MINUTES = 1
MAX_INTERVAL_MINUTES = 24 * 60
def _default_settings_path() -> Path:
"""`<корень проекта>/data/settings.json`.
В репозитории это три уровня вверх от пакета, а в docker-образе файлы лежат
плоско (/app/...), поэтому ищем корень по маркеру — requirements.txt.
"""
here = Path(__file__).resolve()
for parent in here.parents:
if (parent / "requirements.txt").is_file():
return parent / "data" / "settings.json"
return here.parent / "data" / "settings.json"
DEFAULT_SETTINGS_PATH = _default_settings_path()
@dataclass
class Settings:
link: str = ""
interval_minutes: int = DEFAULT_INTERVAL_MINUTES
last_attempt: dict | None = None # {"at": iso, "ip": str, "ok": bool}
last_success: dict | None = None # {"at": iso, "ip": str}
def as_dict(self) -> dict:
return {
"link": self.link,
"interval_minutes": self.interval_minutes,
"last_attempt": self.last_attempt,
"last_success": self.last_success,
}
def clamp_interval(minutes: int) -> int:
return max(MIN_INTERVAL_MINUTES, min(MAX_INTERVAL_MINUTES, int(minutes)))
def interval_from_env(default: int = DEFAULT_INTERVAL_MINUTES) -> int:
"""Интервал из SCHEDULE_INTERVAL_MINUTES, иначе из старого SCHEDULE_CACHE_TTL (сек)."""
raw_minutes = (os.getenv("SCHEDULE_INTERVAL_MINUTES") or "").strip()
if raw_minutes:
try:
return clamp_interval(int(float(raw_minutes)))
except ValueError:
log.warning("SCHEDULE_INTERVAL_MINUTES=%r не число, беру по умолчанию", raw_minutes)
raw_ttl = (os.getenv("SCHEDULE_CACHE_TTL") or "").strip()
if raw_ttl:
try:
return clamp_interval(round(float(raw_ttl) / 60) or 1)
except ValueError:
log.warning("SCHEDULE_CACHE_TTL=%r не число, беру по умолчанию", raw_ttl)
return default
def settings_path_from_env() -> Path | None:
raw = (os.getenv("SETTINGS_FILE") or "").strip()
return Path(raw) if raw else DEFAULT_SETTINGS_PATH
class SettingsStore:
"""Потокобезопасное хранилище настроек администратора."""
def __init__(
self,
link: str = "",
interval_minutes: int = DEFAULT_INTERVAL_MINUTES,
path: str | os.PathLike | None = None,
) -> None:
self._path = Path(path) if path else None
self._lock = threading.RLock()
self._save_error: str | None = None
self._settings = Settings(
link=(link or "").strip(), interval_minutes=clamp_interval(interval_minutes)
)
self._load()
# --- чтение -----------------------------------------------------------------
@property
def path(self) -> Path | None:
return self._path
@property
def link(self) -> str:
with self._lock:
return self._settings.link
@property
def interval_minutes(self) -> int:
with self._lock:
return self._settings.interval_minutes
@property
def last_attempt(self) -> dict | None:
with self._lock:
return dict(self._settings.last_attempt) if self._settings.last_attempt else None
@property
def last_success(self) -> dict | None:
with self._lock:
return dict(self._settings.last_success) if self._settings.last_success else None
@property
def save_error(self) -> str | None:
"""Текст ошибки, если настройки не удаётся сохранить на диск."""
with self._lock:
return self._save_error
def as_dict(self) -> dict:
with self._lock:
return self._settings.as_dict()
# --- запись -----------------------------------------------------------------
def set_link(self, link: str) -> str:
link = (link or "").strip()
if not link:
raise ScheduleError("Ссылка на расписание не может быть пустой")
with self._lock:
self._settings.link = link
self._save()
return link
def set_interval_minutes(self, minutes: int | str) -> int:
try:
value = int(minutes)
except (TypeError, ValueError):
raise ScheduleError("Интервал должен быть целым числом минут") from None
if not MIN_INTERVAL_MINUTES <= value <= MAX_INTERVAL_MINUTES:
raise ScheduleError(
f"Интервал должен быть от {MIN_INTERVAL_MINUTES} до {MAX_INTERVAL_MINUTES} минут"
)
with self._lock:
self._settings.interval_minutes = value
self._save()
return value
def record_login(self, ip: str, ok: bool) -> dict:
"""Запомнить попытку входа: откуда, когда и чем закончилась."""
now = datetime.now().astimezone().replace(microsecond=0)
attempt = {"at": now.isoformat(), "ip": ip or "неизвестно", "ok": bool(ok)}
with self._lock:
self._settings.last_attempt = attempt
if ok:
self._settings.last_success = {"at": attempt["at"], "ip": attempt["ip"]}
self._save()
return attempt
# --- файл -------------------------------------------------------------------
def _load(self) -> None:
if not self._path or not self._path.is_file():
return
try:
raw = json.loads(self._path.read_text(encoding="utf-8"))
except (OSError, ValueError) as exc:
log.warning("Не читается файл настроек %s: %s (беру значения из окружения)", self._path, exc)
return
if not isinstance(raw, dict):
log.warning("Файл настроек %s не JSON-объект, игнорирую", self._path)
return
link = str(raw.get("link") or "").strip()
if link:
self._settings.link = link
if raw.get("interval_minutes") is not None:
try:
self._settings.interval_minutes = clamp_interval(int(raw["interval_minutes"]))
except (TypeError, ValueError):
log.warning("interval_minutes в %s не число, игнорирую", self._path)
for key in ("last_attempt", "last_success"):
value = raw.get(key)
if isinstance(value, dict) and value.get("at"):
setattr(self._settings, key, value)
log.info(
"Настройки из %s: интервал %s мин, ссылка %s",
self._path,
self._settings.interval_minutes,
"задана" if self._settings.link else "пустая",
)
def _save(self) -> None:
if not self._path:
return
try:
self._path.parent.mkdir(parents=True, exist_ok=True)
tmp = self._path.with_suffix(self._path.suffix + ".tmp")
tmp.write_text(
json.dumps(self._settings.as_dict(), ensure_ascii=False, indent=1),
encoding="utf-8",
)
tmp.replace(self._path)
self._save_error = None
except OSError as exc:
self._save_error = str(exc)
log.warning("Не удалось сохранить настройки в %s: %s (живу в памяти)", self._path, exc)

View File

@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
import time
import requests import requests
@@ -12,18 +13,27 @@ log = logging.getLogger(__name__)
CLOUD_API = "https://cloud-api.yandex.net/v1/disk/public/resources" CLOUD_API = "https://cloud-api.yandex.net/v1/disk/public/resources"
# Публичная ссылка может кэшироваться на стороне Диска: просим не кэшировать
# и добавляем в запрос метку времени, чтобы получать текущую версию файла.
_NO_CACHE = {"Cache-Control": "no-cache", "Pragma": "no-cache"}
def _bust() -> dict:
return {"_": int(time.time() * 1000)}
def download_schedule(public_url: str, timeout: float = 60.0) -> bytes: def download_schedule(public_url: str, timeout: float = 60.0) -> bytes:
"""Скачивает файл с диска; при любой ошибке бросает ScheduleError.""" """Скачивает файл с диска; при любой ошибке бросает ScheduleError."""
try: try:
resp = requests.get( resp = requests.get(
f"{CLOUD_API}/download", f"{CLOUD_API}/download",
params={"public_key": public_url}, params={"public_key": public_url, **_bust()},
headers=_NO_CACHE,
timeout=timeout, timeout=timeout,
) )
resp.raise_for_status() resp.raise_for_status()
href = resp.json()["href"] href = resp.json()["href"]
file_resp = requests.get(href, timeout=timeout) file_resp = requests.get(href, headers=_NO_CACHE, timeout=timeout)
file_resp.raise_for_status() file_resp.raise_for_status()
return file_resp.content return file_resp.content
except (requests.RequestException, KeyError, ValueError) as exc: except (requests.RequestException, KeyError, ValueError) as exc:
@@ -35,7 +45,8 @@ def fetch_table_modified(public_url: str, timeout: float = 20.0) -> str | None:
try: try:
resp = requests.get( resp = requests.get(
CLOUD_API, CLOUD_API,
params={"public_key": public_url}, params={"public_key": public_url, **_bust()},
headers=_NO_CACHE,
timeout=timeout, timeout=timeout,
) )
resp.raise_for_status() resp.raise_for_status()

356
src/frontend/admin.html Normal file
View File

@@ -0,0 +1,356 @@
<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta name="robots" content="noindex, nofollow">
<title>Админка расписания</title>
<link rel="icon" type="image/svg+xml" href="/img/app-icon.svg">
<link rel="stylesheet" href="/style.css">
<style>
.admin-wrap { max-width: 760px; margin: 0 auto; padding: 26px 18px 60px; }
.admin-head { display: flex; align-items: center; gap: 12px; margin-bottom: 6px; }
.admin-head img { width: 40px; height: 40px; }
.admin-title { font-size: 22px; font-weight: 700; color: var(--text); }
.admin-sub { color: var(--muted); font-size: 14px; margin-bottom: 18px; }
.admin-card {
background: var(--surface); border: 1px solid var(--border);
padding: 16px 18px; margin-bottom: 14px;
}
.admin-card h2 { font-size: 15px; margin-bottom: 10px; color: var(--text); }
.admin-row { display: flex; gap: 10px; flex-wrap: wrap; align-items: flex-end; }
.admin-field { display: flex; flex-direction: column; gap: 4px; flex: 1 1 260px; min-width: 0; }
.admin-field label { font-size: 13px; color: var(--muted); }
.admin-field input {
font-size: 14px; padding: 8px 10px; border: 1px solid var(--border-2);
background: var(--surface-2); color: var(--text); width: 100%; min-width: 0;
}
.admin-field input:focus { outline: 2px solid var(--accent-blue); outline-offset: -1px; }
.admin-btn {
font-size: 14px; font-weight: 600; padding: 9px 14px; cursor: pointer;
border: 1px solid var(--accent); background: var(--accent); color: #fff;
}
.admin-btn.secondary { background: var(--surface-2); color: var(--text); border-color: var(--border-2); }
.admin-btn.danger { background: var(--accent-red-bright); border-color: var(--accent-red-bright); }
.admin-btn[disabled] { opacity: .55; cursor: default; }
.admin-status { display: flex; flex-wrap: wrap; gap: 10px; }
.admin-chip {
background: var(--surface-2); border: 1px solid var(--border); padding: 8px 12px; min-width: 0;
}
.admin-chip .k { font-size: 12px; color: var(--muted); }
.admin-chip .v { font-size: 14px; font-weight: 600; color: var(--text); word-break: break-word; overflow-wrap: anywhere; }
.admin-error { color: var(--accent-red-bright); font-size: 13px; margin-top: 10px; word-break: break-word; overflow-wrap: anywhere; }
.admin-warn { color: #b8860b; font-size: 13px; margin-top: 10px; word-break: break-word; }
.admin-hint { color: var(--muted-2); font-size: 12px; margin-top: 8px; }
.hidden { display: none !important; }
.bad { color: var(--accent-red-bright); }
.good { color: #2f9e5f; }
/* всплывающие уведомления вместо alert(): сами исчезают через 15 секунд */
#toast-wrap {
position: fixed; left: 50%; bottom: 18px; transform: translateX(-50%);
z-index: 9999; display: flex; flex-direction: column; gap: 8px;
width: min(560px, calc(100vw - 32px));
}
.toast {
display: flex; align-items: flex-start; gap: 10px;
background: var(--surface); border: 1px solid var(--border-2);
border-left: 4px solid var(--accent-blue);
box-shadow: 0 8px 24px var(--shadow);
padding: 12px 14px; font-size: 14px; color: var(--text);
white-space: pre-line; animation: toast-in .18s ease-out;
}
.toast.ok { border-left-color: #2f9e5f; }
.toast.error { border-left-color: var(--accent-red-bright); }
.toast.hide { opacity: 0; transition: opacity .3s ease; }
.toast .toast-text { flex: 1; min-width: 0; overflow-wrap: anywhere; }
.toast .toast-close {
flex: 0 0 auto; border: none; background: none; color: var(--muted);
font-size: 15px; line-height: 1; cursor: pointer; padding: 2px 4px;
}
.toast .toast-close:hover { color: var(--text); }
@keyframes toast-in {
from { opacity: 0; transform: translateY(6px); }
to { opacity: 1; transform: none; }
}
</style>
</head>
<body>
<div class="admin-wrap">
<div class="admin-head">
<img src="/img/owl.svg" alt="">
<div class="admin-title">Админка расписания</div>
</div>
<div class="admin-sub">Управление источником данных и обновлением. Страница доступна только по секретной ссылке.</div>
<section id="login-card" class="admin-card hidden">
<h2>Вход</h2>
<form id="login-form" autocomplete="off">
<div class="admin-field"><label for="login">Логин</label><input id="login" required></div>
<div class="admin-field" style="margin-top:10px"><label for="password">Пароль</label><input id="password" type="password" required></div>
<div class="admin-row" style="margin-top:12px">
<button class="admin-btn" type="submit">Войти</button>
</div>
</form>
<div id="login-error" class="admin-error hidden"></div>
</section>
<section id="panel" class="hidden">
<div class="admin-card">
<h2>Состояние</h2>
<div class="admin-status" id="status-chips"></div>
<div id="state-error" class="admin-error hidden"></div>
<div id="state-warn" class="admin-warn hidden"></div>
</div>
<div class="admin-card">
<h2>Входы в админку</h2>
<div class="admin-status" id="auth-chips"></div>
<div class="admin-hint">Каждая попытка входа запоминается: откуда, когда и чем закончилась. После нескольких неудач вход временно блокируется.</div>
</div>
<div class="admin-card">
<h2>Ссылка на расписание</h2>
<div class="admin-row">
<div class="admin-field">
<label for="link">Публичная ссылка Яндекс.Диска</label>
<input id="link" spellcheck="false" placeholder="https://disk.yandex.ru/i/...">
</div>
<button class="admin-btn" id="save-link" type="button">Сохранить ссылку</button>
</div>
<div class="admin-hint">Ссылка проверяется скачиванием: если не открывается — не сохранится, данные на сайте останутся прежними.
Значение хранится в файле настроек и <b>имеет приоритет над .env</b> — именно по нему сайт берёт расписание.</div>
</div>
<div class="admin-card">
<h2>Обновление</h2>
<div class="admin-row">
<div class="admin-field" style="flex:0 1 200px">
<label for="interval">Интервал, минут</label>
<input id="interval" type="number" min="1" max="1440" step="1">
</div>
<button class="admin-btn secondary" id="save-interval" type="button">Сохранить интервал</button>
<button class="admin-btn" id="refresh" type="button">Обновить сейчас</button>
</div>
<div class="admin-hint">По интервалу расписание перечитывается, только если на Диске изменилась дата файла. Кнопка «Обновить сейчас» скачивает таблицу всегда — её и нажимайте сразу после правок.</div>
</div>
<div class="admin-card">
<button class="admin-btn danger" id="logout" type="button">Выйти</button>
</div>
</section>
</div>
<div id="toast-wrap"></div>
<script>
const $ = (id) => document.getElementById(id);
const TOAST_MS = 15000;
/** Всплывающее уведомление: само скрывается через 15 секунд. */
function toast(message, kind = "info", ms = TOAST_MS) {
const wrap = $("toast-wrap");
const el = document.createElement("div");
el.className = "toast " + kind;
const text = document.createElement("div");
text.className = "toast-text";
text.textContent = message;
const close = document.createElement("button");
close.type = "button";
close.className = "toast-close";
close.textContent = "✕";
close.setAttribute("aria-label", "Закрыть");
el.append(text, close);
wrap.appendChild(el);
const hide = () => {
clearTimeout(timer);
el.classList.add("hide");
setTimeout(() => el.remove(), 300);
};
const timer = setTimeout(hide, ms);
close.addEventListener("click", hide);
return hide;
}
async function api(path, options = {}) {
let resp;
try {
resp = await fetch("/api/admin" + path, { credentials: "same-origin", ...options });
} catch (err) {
return { ok: false, status: 0, data: { detail: "Сеть недоступна: " + err.message } };
}
let data = null;
try { data = await resp.json(); } catch (_) {}
return { ok: resp.ok, status: resp.status, data };
}
function post(path, payload) {
return api(path, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload || {}),
});
}
function fmt(value) {
if (!value) return "—";
const d = new Date(value);
return isNaN(d) ? String(value) : d.toLocaleString("ru-RU", { dateStyle: "short", timeStyle: "short" });
}
function chip(key, value, cls) {
const div = document.createElement("div");
div.className = "admin-chip";
div.innerHTML = `<div class="k">${key}</div><div class="v"></div>`;
const v = div.querySelector(".v");
v.textContent = value;
if (cls) v.classList.add(cls);
return div;
}
function loginText(info) {
if (!info) return "нет данных";
// у успешного входа поля ok нет — там итог и так известен
const status = info.ok === undefined ? "" : ` · ${info.ok ? "успешно" : "неудачно"}`;
return `${fmt(info.at)} · ${info.ip}${status}`;
}
function renderState(state) {
const chips = $("status-chips");
chips.innerHTML = "";
chips.appendChild(chip("Интервал обновления", `${state.interval_minutes} мин`));
chips.appendChild(chip("Обновлено", fmt(state.last_updated)));
chips.appendChild(chip(
"Файл при последнем обновлении",
state.changed ? "изменился" : "без изменений",
state.changed ? "good" : null,
));
chips.appendChild(chip("Дата файла на Диске", fmt(state.table_modified)));
chips.appendChild(chip("Групп в расписании", String(state.groups)));
if (state.content_hash) chips.appendChild(chip("sha256 файла", state.content_hash));
chips.appendChild(chip("Текущая ссылка", state.link || "не задана"));
if (state.settings_file) chips.appendChild(chip("Файл настроек", state.settings_file));
const auth = $("auth-chips");
auth.innerHTML = "";
const attempt = state.last_attempt;
auth.appendChild(chip(
"Последняя попытка входа",
loginText(attempt),
attempt ? (attempt.ok ? "good" : "bad") : null,
));
auth.appendChild(chip("Последний успешный вход", loginText(state.last_success)));
const throttle = state.throttle || {};
if (throttle.locked_until) {
auth.appendChild(chip("Вход заблокирован до", fmt(throttle.locked_until), "bad"));
} else if (throttle.max_attempts) {
auth.appendChild(chip(
"Осталось попыток",
`${throttle.attempts_left} из ${throttle.max_attempts}`,
));
}
$("link").value = state.link || "";
$("interval").value = state.interval_minutes;
const err = $("state-error");
err.textContent = state.error ? "Ошибка загрузки: " + state.error : "";
err.classList.toggle("hidden", !state.error);
const warn = $("state-warn");
warn.textContent = state.settings_save_error
? "Настройки не сохраняются на диск: " + state.settings_save_error +
" — после перезапуска службы ссылка и интервал вернутся к значениям из .env."
: "";
warn.classList.toggle("hidden", !state.settings_save_error);
}
function showLogin(show) {
$("login-card").classList.toggle("hidden", !show);
$("panel").classList.toggle("hidden", show);
if (show) $("login").focus();
}
function busy(button, text) {
const original = button.textContent;
button.disabled = true;
button.textContent = text;
return () => { button.disabled = false; button.textContent = original; };
}
async function loadState() {
const r = await api("/state");
if (r.status === 401) { showLogin(true); return; }
if (!r.ok) {
toast("Не удалось получить состояние: " + ((r.data && r.data.detail) || r.status), "error");
return;
}
showLogin(false);
renderState(r.data);
}
async function withButton(button, text, fn) {
const done = busy(button, text);
try { await fn(); } finally { done(); }
}
$("login-form").addEventListener("submit", async (e) => {
e.preventDefault();
const error = $("login-error");
error.classList.add("hidden");
const r = await post("/login", { login: $("login").value, password: $("password").value });
if (!r.ok) {
error.textContent = (r.data && r.data.detail) || "Не удалось войти";
error.classList.remove("hidden");
return;
}
$("password").value = "";
toast("Вход выполнен.", "ok", 5000);
await loadState();
});
$("save-link").addEventListener("click", () => withButton($("save-link"), "Проверяю…", async () => {
const r = await post("/link", { link: $("link").value });
if (!r.ok) {
toast("Ссылка не принята: " + ((r.data && r.data.detail) || r.status), "error");
return;
}
renderState(r.data);
toast("Ссылка сохранена, расписание загружено.", "ok");
}));
$("save-interval").addEventListener("click", () => withButton($("save-interval"), "Сохраняю…", async () => {
const r = await post("/interval", { interval_minutes: Number($("interval").value) });
if (!r.ok) {
toast("Интервал не сохранён: " + ((r.data && r.data.detail) || r.status), "error");
return;
}
toast(`Интервал обновления: ${r.data.interval_minutes} мин.`, "ok");
await loadState();
}));
$("refresh").addEventListener("click", () => withButton($("refresh"), "Обновляю…", async () => {
const r = await post("/refresh");
if (!r.ok) {
toast("Обновить не удалось: " + ((r.data && r.data.detail) || r.status), "error");
} else if (r.data) {
renderState(r.data);
toast(
r.data.changed
? "Готово: в файле обнаружены изменения — расписание обновлено."
: "Готово: файл не изменился (на Диске та же версия).\nЕсли вы только что правили таблицу, значит Диск ещё не отдаёт новую версию — попробуйте через минуту.",
r.data.changed ? "ok" : "info",
);
}
await loadState();
}));
$("logout").addEventListener("click", async () => {
await post("/logout");
showLogin(true);
});
loadState();
</script>
</body>
</html>

View File

@@ -1,27 +1,53 @@
"""Фронтенд расписания: статики + прокси к API бэкенда.""" """Фронтенд расписания: статики, прокси к API и страница админки.
Страница админки отдаётся по секретному пути из ADMIN_PATH; сам API админки —
/api/admin/* (проксируется в backend). Если ADMIN_PATH не задан, страницы нет.
"""
from __future__ import annotations from __future__ import annotations
import os
import logging import logging
import os
from pathlib import Path from pathlib import Path
import requests import requests
import uvicorn import uvicorn
from dotenv import load_dotenv from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Query from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.responses import FileResponse from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
load_dotenv(Path(__file__).parent / ".env")
# Бэкенд на холодном кэше скачивает таблицу (до 2 минут), таймаут выше. def _find_dotenv() -> Path | None:
"""Ищем .env вверх от каталога модуля: один общий .env в корне проекта."""
path = Path(__file__).resolve().parent
for _ in range(5):
env = path / ".env"
if env.is_file():
return env
if path.parent == path:
break
path = path.parent
return None
_dotenv_path = _find_dotenv()
if _dotenv_path:
load_dotenv(_dotenv_path)
# Адрес бэкенда: в docker это http://api:8000 (задаётся в docker-compose),
# локально — http://127.0.0.1:8000.
SCHEDULE_API = os.getenv("SCHEDULE_API", "http://127.0.0.1:8000").rstrip("/") SCHEDULE_API = os.getenv("SCHEDULE_API", "http://127.0.0.1:8000").rstrip("/")
# Бэкенд на холодном кэше скачивает таблицу (до 2 минут), таймаут выше.
PROXY_TIMEOUT = float(os.getenv("SCHEDULE_PROXY_TIMEOUT", "150")) PROXY_TIMEOUT = float(os.getenv("SCHEDULE_PROXY_TIMEOUT", "150"))
ADMIN_PATH = (os.getenv("ADMIN_PATH") or "").strip().rstrip("/")
STATIC_DIR = Path(__file__).parent / "static" STATIC_DIR = Path(__file__).parent / "static"
# страница админки вне static: прямой доступ по /admin.html закрыт
ADMIN_PAGE = Path(__file__).parent / "admin.html"
app = FastAPI(title="Расписание — фронт") app = FastAPI(title="Расписание — фронт")
@@ -43,7 +69,33 @@ def _proxy(path: str, params: dict) -> object:
except Exception: except Exception:
detail = resp.text detail = resp.text
raise HTTPException(status_code=resp.status_code, detail=detail) raise HTTPException(status_code=resp.status_code, detail=detail)
return resp.json() # данные не кэшируем: браузер всегда показывает актуальное расписание
return JSONResponse(content=resp.json(), headers={"Cache-Control": "no-store"})
def _proxy_admin(method: str, path: str, body: bytes, headers: dict) -> Response:
"""Проксирует админ-запрос как есть: метод, тело, куки и Set-Cookie."""
try:
resp = requests.request(
method,
f"{SCHEDULE_API}/{path}",
data=body or None,
headers=headers,
timeout=PROXY_TIMEOUT,
)
except requests.RequestException as exc:
raise HTTPException(status_code=502, detail=f"Бэкенд недоступен: {exc}")
out = Response(
content=resp.content,
status_code=resp.status_code,
media_type=resp.headers.get("content-type"),
headers={"Cache-Control": "no-store"},
)
# Set-Cookie (вход/выход админки) передаём как есть
raw_headers = getattr(resp.raw, "headers", None)
for cookie in raw_headers.getlist("Set-Cookie") if raw_headers is not None else []:
out.headers.append("set-cookie", cookie)
return out
# --- API-маршруты ----------------------------------------------------------- # --- API-маршруты -----------------------------------------------------------
@@ -99,7 +151,29 @@ def api_status():
return _proxy("status", {}) return _proxy("status", {})
# --- статики ----------------------------------------------------------------- @app.api_route("/api/admin/{rest:path}", methods=["GET", "POST"])
async def api_admin(rest: str, request: Request):
"""Админ-API: тело и куки передаём в бэкенд, ответ — обратно как есть."""
body = await request.body()
headers = {}
content_type = request.headers.get("content-type")
if content_type:
headers["Content-Type"] = content_type
cookie = request.headers.get("cookie")
if cookie:
headers["Cookie"] = cookie
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
headers["X-Forwarded-For"] = forwarded
forwarded_proto = request.headers.get("x-forwarded-proto")
if forwarded_proto:
headers["X-Forwarded-Proto"] = forwarded_proto
return await run_in_threadpool(
_proxy_admin, request.method, f"api/admin/{rest}", body, headers
)
# --- статики и админка --------------------------------------------------------
@app.get("/") @app.get("/")
@@ -107,6 +181,20 @@ def index():
return FileResponse(STATIC_DIR / "index.html") return FileResponse(STATIC_DIR / "index.html")
if ADMIN_PATH:
@app.get(ADMIN_PATH, include_in_schema=False)
def admin_page():
"""Страница админки: путь задаётся в ADMIN_PATH, входа с сайта нет."""
return FileResponse(
ADMIN_PAGE,
headers={"X-Robots-Tag": "noindex, nofollow", "Cache-Control": "no-store"},
)
log.info("Страница админки включена по секретному пути")
else:
log.info("ADMIN_PATH не задан — страница админки выключена")
app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static") app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")

View File

@@ -85,7 +85,7 @@ def make_sheets(sheets: dict[str, SheetSpec]):
return parse_workbook(build_workbook(sheets)) return parse_workbook(build_workbook(sheets))
def make_cache(monkeypatch, xlsx: bytes | None, modified: str = "fixed-modified"): def make_cache(monkeypatch, xlsx: bytes | None, modified: str = "fixed-modified", settings=None):
"""ScheduleCache с заменёнными сетевыми функциями.""" """ScheduleCache с заменёнными сетевыми функциями."""
import schedule.cache as cache_mod import schedule.cache as cache_mod
from schedule import ScheduleCache from schedule import ScheduleCache
@@ -107,5 +107,5 @@ def make_cache(monkeypatch, xlsx: bytes | None, modified: str = "fixed-modified"
monkeypatch.setattr(cache_mod, "download_schedule", fake_download) monkeypatch.setattr(cache_mod, "download_schedule", fake_download)
monkeypatch.setattr(cache_mod, "fetch_table_modified", fake_modified) monkeypatch.setattr(cache_mod, "fetch_table_modified", fake_modified)
cache = ScheduleCache("test-link", ttl_seconds=3600, retry_after=60) cache = ScheduleCache("test-link", ttl_seconds=3600, retry_after=60, settings=settings)
return cache, calls, fail return cache, calls, fail

227
tests/test_admin.py Normal file
View File

@@ -0,0 +1,227 @@
"""Тесты админки: вход, блокировка после неудач, история входов, ссылка/интервал/force."""
import json
import pytest
from fastapi.testclient import TestClient
import main
from admin import AdminConfig, make_token, verify_token
from fixtures import fixture_xlsx, make_cache
from schedule import ScheduleError
from schedule.settings import SettingsStore
ADMIN = AdminConfig(path="/secret-panel", login="admin", password="s3cret")
@pytest.fixture()
def store(tmp_path):
return SettingsStore(link="test-link", interval_minutes=15, path=tmp_path / "settings.json")
@pytest.fixture()
def cache(monkeypatch, store):
c, calls, fail = make_cache(monkeypatch, fixture_xlsx(), settings=store)
c.calls = calls
c.fail = fail
return c
@pytest.fixture()
def client(cache, store):
app = main.create_app(cache, admin=ADMIN, settings=store)
with TestClient(app) as c:
yield c
def do_login(client, login="admin", password="s3cret"):
return client.post("/api/admin/login", json={"login": login, "password": password})
# --- авторизация ---------------------------------------------------------------
@pytest.mark.parametrize(
"method, path, payload",
[
("get", "/api/admin/state", None),
("post", "/api/admin/refresh", {}),
("post", "/api/admin/link", {"link": "https://disk.yandex.ru/i/x"}),
("post", "/api/admin/interval", {"interval_minutes": 30}),
],
)
def test_requires_login(client, method, path, payload):
resp = getattr(client, method)(path, json=payload) if method == "post" else client.get(path)
assert resp.status_code == 401
@pytest.mark.parametrize(
"login, password",
[("admin", "неверный"), ("неверный", "s3cret"), ("", "")],
)
def test_login_rejects_wrong_credentials(client, login, password):
resp = client.post("/api/admin/login", json={"login": login, "password": password})
assert resp.status_code == 401
assert "schedule_admin" not in client.cookies
def test_login_and_state(client, cache):
assert do_login(client).status_code == 200
resp = client.get("/api/admin/state")
assert resp.status_code == 200
state = resp.json()
assert state["link"] == "test-link"
assert state["interval_minutes"] == 15
assert state["groups"] == 2
assert state["error"] is None
assert state["throttle"]["max_attempts"] == ADMIN.max_attempts
def test_logout(client):
do_login(client)
assert client.get("/api/admin/state").status_code == 200
assert client.post("/api/admin/logout").status_code == 200
assert client.get("/api/admin/state").status_code == 401
def test_token_helpers():
token = make_token("admin", "s3cret", ttl=60)
assert verify_token(token, "s3cret")
assert not verify_token(token, "другой-пароль")
assert not verify_token(token + "x", "s3cret")
assert not verify_token("мусор", "s3cret")
assert not verify_token(make_token("admin", "s3cret", ttl=-5), "s3cret")
# --- блокировка после неудачных попыток ------------------------------------------
def test_lockout_after_configured_attempts(client):
"""Разрешённые попытки тратятся, следующая попытка включает паузу."""
for _ in range(ADMIN.max_attempts - 1):
resp = do_login(client, password="неверный")
assert resp.status_code == 401
assert "Осталось попыток" in resp.json()["detail"]
exhausted = do_login(client, password="неверный")
assert exhausted.status_code == 429
assert "заблокирован" in exhausted.json()["detail"].lower()
# пауза действует и для верного пароля
assert do_login(client, password="s3cret").status_code == 429
def test_successful_login_resets_failures(client):
assert do_login(client, password="неверный").status_code == 401
assert do_login(client).status_code == 200
# счётчик сброшен: ещё две неудачи не блокируют
assert do_login(client, password="неверный").status_code == 401
assert do_login(client, password="неверный").status_code == 401
assert do_login(client).status_code == 200
# --- история входов ---------------------------------------------------------------
def test_login_history_visible_and_saved(client, store):
assert do_login(client, password="неверный").status_code == 401
assert do_login(client).status_code == 200
state = client.get("/api/admin/state").json()
assert state["last_attempt"]["ok"] is True
assert state["last_attempt"]["ip"] # откуда
assert state["last_attempt"]["at"] # когда
assert state["last_success"]["ip"] == state["last_attempt"]["ip"]
saved = json.loads(store.path.read_text(encoding="utf-8"))
assert saved["last_attempt"]["ok"] is True
assert saved["last_success"]["at"]
# --- интервал --------------------------------------------------------------------
def test_set_interval(client, cache, store):
do_login(client)
resp = client.post("/api/admin/interval", json={"interval_minutes": 30})
assert resp.status_code == 200
assert resp.json()["interval_minutes"] == 30
assert cache.interval_minutes == 30
assert cache._ttl == 1800
assert client.get("/api/admin/state").json()["interval_minutes"] == 30
saved = json.loads(store.path.read_text(encoding="utf-8"))
assert saved["interval_minutes"] == 30
assert SettingsStore(path=store.path).interval_minutes == 30
@pytest.mark.parametrize("value, status", [(0, 400), (1441, 400), (-5, 400), ("abc", 422)])
def test_set_interval_validation(client, value, status):
do_login(client)
resp = client.post("/api/admin/interval", json={"interval_minutes": value})
assert resp.status_code == status
# --- смена ссылки ---------------------------------------------------------------
def test_set_link(client, cache, store):
do_login(client)
before = cache.calls["download"]
resp = client.post("/api/admin/link", json={"link": " https://disk.yandex.ru/i/new "})
assert resp.status_code == 200
assert resp.json()["link"] == "https://disk.yandex.ru/i/new"
assert cache.calls["download"] == before + 1 # перечитали таблицу
assert json.loads(store.path.read_text(encoding="utf-8"))["link"] == "https://disk.yandex.ru/i/new"
def test_set_link_rejects_broken(client, cache, store):
do_login(client)
old_link = cache.state()["link"]
cache.fail["download"] = True
resp = client.post("/api/admin/link", json={"link": "https://disk.yandex.ru/i/broken"})
assert resp.status_code == 400
assert "не подошла" in resp.json()["detail"]
assert cache.state()["link"] == old_link
assert cache.state()["has_data"] is True
if store.path.exists():
assert json.loads(store.path.read_text(encoding="utf-8"))["link"] != "https://disk.yandex.ru/i/broken"
def test_set_empty_link(client, cache):
do_login(client)
resp = client.post("/api/admin/link", json={"link": " "})
assert resp.status_code == 400
assert cache.state()["has_data"] is True
# --- принудительное обновление ---------------------------------------------------
def test_force_refresh_ignores_ttl_and_modified(client, cache):
do_login(client)
client.get("/api/admin/state")
downloads = cache.calls["download"]
cache._ttl = 3600 # кэш свежий; обычное обновление не сработало бы
resp = client.post("/api/admin/refresh")
assert resp.status_code == 200
assert cache.calls["download"] == downloads + 1
assert resp.json()["has_data"] is True
def test_force_refresh_reports_error(client, cache):
do_login(client)
cache.fail["download"] = True
resp = client.post("/api/admin/refresh")
assert resp.status_code == 502
assert cache.state()["error"] is not None
assert cache.state()["has_data"] is True # старые данные не потеряли
# --- выключенная админка -----------------------------------------------------------
def test_admin_disabled_without_credentials(cache, store):
app = main.create_app(cache, admin=AdminConfig(), settings=store)
with TestClient(app) as c:
assert c.get("/api/admin/state").status_code == 404
assert c.post("/api/admin/login", json={"login": "a", "password": "b"}).status_code == 404
assert c.get("/healthz").status_code == 200

View File

@@ -46,6 +46,21 @@ def test_groups(client):
assert client.get("/groups").json() == ["ИВТ-11БО", "ИТ-12БО"] assert client.get("/groups").json() == ["ИВТ-11БО", "ИТ-12БО"]
def test_groups_keep_sheet_order(monkeypatch):
"""GET /groups отдаёт группы в порядке столбцов таблицы, а не по алфавиту."""
from fixtures import fixture_sheets
from xlsx_builder import build_workbook
sheets = fixture_sheets()
header = sheets["Бакалавры и магистры"].rows[0]
header[6] = "ЯЯ-ТОП-11БО\nсеместр до 11.01.27"
header[7] = "АА-11БО\nсеместр до 29.12.26"
cache, _, _ = make_cache(monkeypatch, build_workbook(sheets))
with TestClient(main.create_app(cache)) as client:
assert client.get("/groups").json() == ["ЯЯ-ТОП-11БО", "АА-11БО"]
def test_teachers(client): def test_teachers(client):
assert client.get("/teachers").json() == [ assert client.get("/teachers").json() == [
"Иванов И.И.", "Петров П.П.", "Сидорова С.С.", "Смирнов А.В.", "Иванов И.И.", "Петров П.П.", "Сидорова С.С.", "Смирнов А.В.",
@@ -155,14 +170,52 @@ def test_cache_no_repeated_download(client, cache):
assert cache.calls["download"] == 1 assert cache.calls["download"] == 1
def test_cache_modified_unchanged_no_download(cache): def test_modified_unchanged_no_download(cache):
cache._ttl = 0 # даже с истёкшим TTL """По интервалу сверяем modified: если таблица не менялась — не скачиваем."""
cache._ttl = 0
cache.get() cache.get()
cache.get() cache.get()
assert cache.calls["download"] == 1 assert cache.calls["download"] == 1
assert cache.calls["modified"] >= 2 assert cache.calls["modified"] >= 2
def test_status_reports_file_changes(monkeypatch):
"""Кэш показывает, изменилось ли содержимое файла при последнем чтении."""
import schedule.cache as cache_mod
from fixtures import fixture_sheets
from xlsx_builder import build_workbook
def xlsx_with_room(room: str) -> bytes:
sheets = fixture_sheets()
sheets["Бакалавры и магистры"].rows[2][6] = f"Математика\nСмирнов А.В.\nауд. {room}"
return build_workbook(sheets)
files = {"data": xlsx_with_room("215"), "modified": "m1"}
monkeypatch.setattr(
cache_mod, "fetch_table_modified", lambda url, timeout=20.0: files["modified"]
)
monkeypatch.setattr(cache_mod, "download_schedule", lambda url, timeout=60.0: files["data"])
cache = ScheduleCache("test-link", ttl_seconds=0, retry_after=60)
cache.get()
first = cache.status()
assert first["content_hash"]
assert first["changed"] is False # первая загрузка — не «изменение»
assert first["changed_at"] is not None
files["data"] = xlsx_with_room("216")
files["modified"] = "m2"
cache.get() # ttl=0 и новый modified — перечитываем таблицу
changed = cache.status()
assert changed["changed"] is True
assert changed["content_hash"] != first["content_hash"]
# новая ревизия файла с тем же содержимым → скачали, но изменений нет
files["modified"] = "m3"
cache.get()
assert cache.status()["changed"] is False
def test_cache_stale_while_error(cache): def test_cache_stale_while_error(cache):
cache._ttl = 0 # форсим повторную попытку cache._ttl = 0 # форсим повторную попытку
cache.get() cache.get()
@@ -196,12 +249,17 @@ def test_cache_failure_without_data_raises(cache):
assert cache.status()["last_updated"] is None assert cache.status()["last_updated"] is None
def test_broken_cache_http(client=None): def test_cache_without_link():
"""Без ссылки приложение живёт: /healthz есть, данные отвечают понятной ошибкой."""
from fastapi.testclient import TestClient as TC from fastapi.testclient import TestClient as TC
app = main.create_app(main._BrokenCache()) from schedule import ScheduleCache
from schedule.settings import SettingsStore
cache = ScheduleCache(settings=SettingsStore(link=""), retry_after=60)
app = main.create_app(cache)
with TC(app) as c: with TC(app) as c:
assert c.get("/healthz").status_code == 200 assert c.get("/healthz").status_code == 200
assert c.get("/groups").status_code == 502 assert c.get("/groups").status_code == 502
body = c.get("/status").json() body = c.get("/status").json()
assert "SCHEDULE_LINK" in body["error"] assert "Ссылка" in body["error"]

51
tests/test_deploy.py Normal file
View File

@@ -0,0 +1,51 @@
"""Проверки сборки образов: все нужные файлы должны копироваться.
Ошибка «в Dockerfile забыли COPY какого-то модуля» роняет контейнер на импорте,
а тесты из исходников её не видят (они запускаются из рабочей копии). Ловим
статически, без docker.
"""
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
DOCKERFILE = ROOT / "Dockerfile"
COMPOSE = ROOT / "docker-compose.yml"
def _stage(name: str) -> str:
dockerfile = DOCKERFILE.read_text(encoding="utf-8")
part = dockerfile.split(f"FROM base AS {name}", 1)[1]
for other in ("api", "frontend"):
if other != name:
part = part.split(f"FROM base AS {other}", 1)[0]
return part
def test_api_image_copies_every_backend_module():
stage = _stage("api")
for module in sorted((ROOT / "src" / "backend").glob("*.py")):
assert module.name in stage, f"{module.name} не копируется в образ api"
assert "src/backend/schedule" in stage, "пакет schedule не копируется в образ api"
def test_frontend_image_copies_page_and_static():
stage = _stage("frontend")
for path in ("src/frontend/main.py", "src/frontend/admin.html", "src/frontend/static"):
assert path in stage, f"{path} не копируется в образ frontend"
def test_compose_passes_required_env():
compose = COMPOSE.read_text(encoding="utf-8")
for var in (
"SCHEDULE_LINK",
"SCHEDULE_INTERVAL_MINUTES",
"SCHEDULE_API",
"ADMIN_PATH",
"ADMIN_LOGIN",
"ADMIN_PASSWORD",
"ADMIN_MAX_ATTEMPTS",
"ADMIN_LOCKOUT_MINUTES",
"SETTINGS_FILE",
):
assert var in compose, f"{var} не передаётся контейнерам"
assert "schedule_data:/data" in compose, "том для настроек админки не подключён"

View File

@@ -14,7 +14,7 @@ def lessons(text):
def one(text): def one(text):
ls = parse_lesson_cell(text) ls = parse_lesson_cell(text)
assert len(ls) == 1, f"ожидалося 1 занятие, got: {ls}" assert len(ls) == 1, f"ожидалось 1 занятие, got: {ls}"
return ls[0].to_dict() return ls[0].to_dict()
@@ -295,6 +295,53 @@ def test_lesson_with_red_between_lessons():
assert ls[1]["subject"] == "Язык UML и CASE-системы" assert ls[1]["subject"] == "Язык UML и CASE-системы"
# --- оторвавшиеся пустой строкой преподаватель/аудитория ---------------------------
def test_room_detached_by_blank_line():
# ПИЭ-31БО, вторник: «ауд.210» оторвана пустой строкой от предмета
result = one("Программная инженерия\nЛарина Ю.А.\n\n ауд.210")
assert result["subject"] == "Программная инженерия"
assert result["teacher"] == "Ларина Ю.А."
assert result["room"] == "ауд. 210"
def test_teacher_and_room_detached_by_red_note():
# ИВТ-13БО, четверг: красная заметка снята со своей строки, осталась пустая
raw = f"Практикум по матанализу\n{R}пара для тех кто не ТОП{Cc}\nШабаршина Г.В.\nауд. 219"
result = one(raw)
assert result["subject"] == "Практикум по матанализу"
assert result["teacher"] == "Шабаршина Г.В."
assert result["room"] == "ауд. 219"
assert result["note"] == "пара для тех кто не ТОП"
def test_detached_room_inherits_marker():
ls = lessons("к.в. - 2\nФизика\nИванов И.И.\n\nауд. 301")
assert len(ls) == 1
assert ls[0]["marker"] == "к.в. - 2"
assert ls[0]["room"] == "ауд. 301"
def test_detached_room_kept_if_prev_has_room():
ls = lessons("Физика\nСмирнов А.В.\nауд. 221\n\nауд. 222")
assert len(ls) == 2
assert ls[0]["room"] == "ауд. 221"
def test_detached_teacher_kept_if_prev_has_teacher():
ls = lessons("Физика\nСмирнов А.В.\nауд. 221\n\nПетров П.П.\nауд. 222")
assert len(ls) == 2
assert ls[0]["teacher"] == "Смирнов А.В."
def test_lesson_after_detached_room_starts_fresh():
ls = lessons("Физика\nСмирнов А.В.\n\nауд. 221\n\nХимия\nПетров П.П.\nауд. 222")
assert len(ls) == 2
assert ls[0]["room"] == "ауд. 221"
assert ls[1]["subject"] == "Химия"
# --- пустые и экзотические ---------------------------------------------------------- # --- пустые и экзотические ----------------------------------------------------------

View File

@@ -23,6 +23,28 @@ def test_hidden_group_excluded(data):
assert set(data) == {"ИВТ-11БО", "ИТ-12БО"} assert set(data) == {"ИВТ-11БО", "ИТ-12БО"}
def test_group_names_with_top_and_sheet_order():
"""ТОП-группы распознаются, а порядок групп — как столбцы в таблице."""
sheets = fixture_sheets()
header = sheets[parsing.MAIN_SHEET].rows[0]
header[6] = "ИТ-ТОП-11БО\nсеместр до 11.01.27"
header[7] = "АА-11БО\nсеместр до 29.12.26"
data = parsing.parse_schedule(make_sheets(sheets))
assert list(data) == ["ИТ-ТОП-11БО", "АА-11БО"] # не по алфавиту, а по столбцам
assert data["ИТ-ТОП-11БО"]["days"]["понедельник"][1]["ч"]
def test_not_a_group_headers_ignored():
"""Похожие, но не групповые заголовки столбцами групп не становятся."""
sheets = fixture_sheets()
header = sheets[parsing.MAIN_SHEET].rows[0]
header[9] = "ИВТ 11БО" # пробел вместо дефиса — это не имя группы
data = parsing.parse_schedule(make_sheets(sheets))
assert set(data) == {"ИВТ-11БО", "ИТ-12БО"}
def test_semester(data): def test_semester(data):
assert data["ИВТ-11БО"]["semester_until"] == "11.01.27" assert data["ИВТ-11БО"]["semester_until"] == "11.01.27"
assert data["ИТ-12БО"]["semester_until"] == "29.12.26" assert data["ИТ-12БО"]["semester_until"] == "29.12.26"
@@ -123,6 +145,18 @@ def test_teacher_room_index(data):
assert room_301["2"]["ч"][0]["groups"] == ["ИВТ-11БО"] assert room_301["2"]["ч"][0]["groups"] == ["ИВТ-11БО"]
def test_room_index_detached_room():
"""Аудитория, оторванная пустой строкой, попадает в индекс с предметом и преподавателем."""
sheets = fixture_sheets()
sheets[parsing.MAIN_SHEET].rows[6][7] = "Химия\nПетров П.П.\n\n ауд. 201"
data = parsing.parse_schedule(make_sheets(sheets))
_, rooms = parsing.build_teacher_room_index(data)
entry = rooms["ауд. 201"]["вторник"]["1"]["ч"][0]
assert entry["subject"] == "Химия"
assert entry["teacher"] == "Петров П.П."
assert entry["groups"] == ["ИТ-12БО"]
def test_day_times(data): def test_day_times(data):
times = parsing.build_day_times(data) times = parsing.build_day_times(data)
assert times["понедельник"] == {"1": "9:00-10:35", "2": "10:45-12:20"} assert times["понедельник"] == {"1": "9:00-10:35", "2": "10:45-12:20"}

125
tests/test_settings.py Normal file
View File

@@ -0,0 +1,125 @@
"""Тесты хранилища настроек админки: ссылка, интервал, история входов."""
import json
import pytest
from schedule import ScheduleError
from schedule.settings import (
DEFAULT_INTERVAL_MINUTES,
SettingsStore,
clamp_interval,
interval_from_env,
)
def test_defaults():
store = SettingsStore()
assert store.link == ""
assert store.interval_minutes == DEFAULT_INTERVAL_MINUTES
assert store.last_attempt is None
assert store.last_success is None
def test_persist_and_reload(tmp_path):
path = tmp_path / "settings.json"
store = SettingsStore(link="l1", interval_minutes=20, path=path)
store.set_link("l2")
store.set_interval_minutes(45)
again = SettingsStore(link="из-окружения", interval_minutes=1, path=path)
assert again.link == "l2"
assert again.interval_minutes == 45
def test_file_overrides_env(tmp_path):
path = tmp_path / "settings.json"
path.write_text(json.dumps({"link": "из-файла", "interval_minutes": 5}), encoding="utf-8")
store = SettingsStore(link="из-окружения", interval_minutes=99, path=path)
assert store.link == "из-файла"
assert store.interval_minutes == 5
def test_interval_validation():
store = SettingsStore()
for bad in (0, 1441, -5, "abc"):
with pytest.raises(ScheduleError):
store.set_interval_minutes(bad)
assert store.set_interval_minutes("30") == 30
def test_empty_link_rejected():
store = SettingsStore()
with pytest.raises(ScheduleError):
store.set_link(" ")
with pytest.raises(ScheduleError):
store.set_link("")
def test_corrupt_file_is_tolerated(tmp_path):
path = tmp_path / "settings.json"
path.write_text("{это не json", encoding="utf-8")
store = SettingsStore(link="env-link", interval_minutes=7, path=path)
assert store.link == "env-link"
assert store.interval_minutes == 7
def test_login_history_saved(tmp_path):
path = tmp_path / "settings.json"
store = SettingsStore(link="l", path=path)
store.record_login("10.0.0.1", ok=False)
assert store.last_attempt["ip"] == "10.0.0.1"
assert store.last_attempt["ok"] is False
assert store.last_attempt["at"]
assert store.last_success is None # успешных ещё не было
store.record_login("10.0.0.2", ok=True)
assert store.last_success["ip"] == "10.0.0.2"
# история переживает перезапуск: читаем тем же путём
again = SettingsStore(path=path)
assert again.last_attempt["ip"] == "10.0.0.2"
assert again.last_attempt["ok"] is True
assert again.last_success["ip"] == "10.0.0.2"
def test_unwritable_path_marks_save_error(tmp_path):
blocker = tmp_path / "blocker"
blocker.write_text("x", encoding="utf-8")
store = SettingsStore(link="l", interval_minutes=3, path=blocker / "settings.json")
assert store.set_interval_minutes(10) == 10 # в памяти меняется
assert store.interval_minutes == 10
assert store.save_error # и видно, что на диск не сохранилось
def test_default_settings_path_is_inside_project():
"""Путь по умолчанию вычисляется от корня проекта, а не от глубины файла.
В docker-образе файлы лежат плоско (/app/...), поэтому глубину не считаем.
"""
from schedule.settings import DEFAULT_SETTINGS_PATH
assert DEFAULT_SETTINGS_PATH.name == "settings.json"
assert DEFAULT_SETTINGS_PATH.parent.name == "data"
root = DEFAULT_SETTINGS_PATH.parent.parent
assert (root / "requirements.txt").is_file() # корень найден по маркеру
def test_clamp_interval():
assert clamp_interval(0) == 1
assert clamp_interval(99999) == 1440
assert clamp_interval(30) == 30
def test_interval_from_env(monkeypatch):
monkeypatch.delenv("SCHEDULE_INTERVAL_MINUTES", raising=False)
monkeypatch.delenv("SCHEDULE_CACHE_TTL", raising=False)
assert interval_from_env() == DEFAULT_INTERVAL_MINUTES
# старый SCHEDULE_CACHE_TTL (в секундах) продолжает работать
monkeypatch.setenv("SCHEDULE_CACHE_TTL", "900")
assert interval_from_env() == 15
monkeypatch.setenv("SCHEDULE_INTERVAL_MINUTES", "7")
assert interval_from_env() == 7