Админка в 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-образе (том настроек, запись истории).
This commit is contained in:
2026-10-10 11:12:29 +03:00
parent 962681630f
commit 5b3c9dc00a
17 changed files with 1569 additions and 54 deletions

View File

@@ -14,3 +14,17 @@ SCHEDULE_RETRY_AFTER=60
# снаружи сервисы отдаёт nginx, напрямую в интернет они не смотрят. # снаружи сервисы отдаёт nginx, напрямую в интернет они не смотрят.
API_PORT=8000 API_PORT=8000
FRONTEND_PORT=8080 FRONTEND_PORT=8080
# --- админка (необязательно) ----------------------------------------------------
# Пока эти переменные пустые, админка выключена и наружу ничего не торчит.
# Секретный путь страницы админки: придумайте длинный, например /panel-8f3a91x7
ADMIN_PATH=
# Логин и пароль администратора
ADMIN_LOGIN=
ADMIN_PASSWORD=
# Сколько неудачных попыток входа допускается и на сколько минут блокировать после них
ADMIN_MAX_ATTEMPTS=3
ADMIN_LOCKOUT_MINUTES=15
# Где хранить настройки админки (ссылка, интервал, история входов).
# Пусто = data/settings.json в корне проекта; в docker задаётся автоматически.
SETTINGS_FILE=

3
.gitignore vendored
View File

@@ -9,3 +9,6 @@ __pycache__/
# локальные данные инструментов # локальные данные инструментов
.kilo/ .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

@@ -294,6 +294,11 @@ PyPI не раздаёт вовсе. Зеркала отстают по-разн
| `SCHEDULE_RETRY_AFTER` | `60` | api | пауза между попытками при сбое источника | | `SCHEDULE_RETRY_AFTER` | `60` | api | пауза между попытками при сбое источника |
| `SCHEDULE_API` | `http://127.0.0.1:8000` | frontend | адрес бэкенда (в docker: `http://api:8000`) | | `SCHEDULE_API` | `http://127.0.0.1:8000` | frontend | адрес бэкенда (в docker: `http://api:8000`) |
| `SCHEDULE_PROXY_TIMEOUT` | `150` | frontend | таймаут прокси к бэкенду, сек | | `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) | | `API_PORT` | `8000` | docker | внешний порт api (публикуется на 127.0.0.1) |
| `FRONTEND_PORT` | `8080` | docker | внешний порт frontend (публикуется на 127.0.0.1) | | `FRONTEND_PORT` | `8080` | docker | внешний порт frontend (публикуется на 127.0.0.1) |
@@ -301,6 +306,42 @@ PyPI не раздаёт вовсе. Зеркала отстают по-разн
`.env` в корне репозитория; `load_dotenv` не перезаписывает уже выставленные `.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`). Если файл недоступен для записи, изменения живут в памяти до
перезапуска, а панель предупредит об этом.
## Структура кода ## Структура кода
``` ```
@@ -308,16 +349,19 @@ install.sh интерактивная установка на сер
src/ src/
backend/ backend/
main.py FastAPI: маршруты, create_app(cache), /healthz main.py FastAPI: маршруты, create_app(cache), /healthz
admin.py админка: cookie-сессия, блокировка входа, ссылка, интервал, force
schedule/ schedule/
xlsx.py чтение xlsx: общие строки, объединённые ячейки, xlsx.py чтение xlsx: общие строки, объединённые ячейки,
скрытые столбцы, красный текст, серийные даты скрытые столбцы, красный текст, серийные даты
source.py скачивание файла по публичной ссылке Я.Диска source.py скачивание файла по публичной ссылке Я.Диска
lessons.py разбор ячеек занятий (общий конвейер ролей) lessons.py разбор ячеек занятий (общий конвейер ролей)
parsing.py лист → расписание, недели, индексы преподавателей/аудиторий parsing.py лист → расписание, недели, индексы преподавателей/аудиторий
cache.py ScheduleCache: блокировка, интервал, stale-while-error cache.py ScheduleCache: блокировка, интервал, stale-while-error, force
settings.py хранилище настроек админки (ссылка, интервал, история входов)
errors.py ScheduleError errors.py ScheduleError
frontend/ frontend/
main.py FastAPI: статика + прокси /api/* (таймаут 150 c) main.py FastAPI: статика + прокси /api/* (таймаут 150 c)
admin.html страница админки (вне static: прямой доступ закрыт)
static/ static/
index.html, app.js, style.css index.html, app.js, style.css
manifest.json PWA: имя, иконки, standalone-режим manifest.json PWA: имя, иконки, standalone-режим
@@ -335,9 +379,13 @@ tests/
test_xlsx.py чтение книги test_xlsx.py чтение книги
test_parsing.py разбор листов, недели, индексы test_parsing.py разбор листов, недели, индексы
test_api.py API и поведение кэша (интервал/stale, 404/400/502) test_api.py API и поведение кэша (интервал/stale, 404/400/502)
test_admin.py админка: вход, блокировка, история, ссылка/интервал/force
test_settings.py хранилище настроек админки
test_deploy.py проверка, что образы копируют все нужные файлы test_deploy.py проверка, что образы копируют все нужные файлы
``` ```
Данные админки (`data/settings.json`) — рабочие, в git не попадают.
## Как устроена таблица-источник ## Как устроена таблица-источник
Лист «Бакалавры и магистры» — это сетка: Лист «Бакалавры и магистры» — это сетка:

View File

@@ -8,6 +8,16 @@ services:
SCHEDULE_INTERVAL_MINUTES: ${SCHEDULE_INTERVAL_MINUTES:-15} 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 нужен только фронтенду внутри docker-сети; наружу — 127.0.0.1 для отладки # api нужен только фронтенду внутри docker-сети; наружу — 127.0.0.1 для отладки
- "127.0.0.1:${API_PORT:-8000}:8000" - "127.0.0.1:${API_PORT:-8000}:8000"
@@ -26,6 +36,8 @@ services:
environment: environment:
# внутри compose api доступен по имени сервиса; менять не нужно # внутри compose api доступен по имени сервиса; менять не нужно
SCHEDULE_API: ${SCHEDULE_API:-http://api:8000} SCHEDULE_API: ${SCHEDULE_API:-http://api:8000}
# страница админки отдаётся фронтендом по этому пути
ADMIN_PATH: ${ADMIN_PATH:-}
ports: ports:
# фронт обслуживает nginx; наружу напрямую не смотрит # фронт обслуживает nginx; наружу напрямую не смотрит
- "127.0.0.1:${FRONTEND_PORT:-8080}:8080" - "127.0.0.1:${FRONTEND_PORT:-8080}:8080"
@@ -39,3 +51,6 @@ services:
timeout: 5s timeout: 5s
retries: 3 retries: 3
start_period: 10s start_period: 10s
volumes:
schedule_data:

View File

@@ -180,6 +180,14 @@ SCHEDULE_INTERVAL_MINUTES=15
SCHEDULE_RETRY_AFTER=60 SCHEDULE_RETRY_AFTER=60
API_PORT=$API_PORT API_PORT=$API_PORT
FRONTEND_PORT=$FRONT_PORT FRONTEND_PORT=$FRONT_PORT
# Админка (необязательно). Заполните path/логин/пароль, чтобы включить панель:
# страница будет доступна только по https://<домен><ADMIN_PATH>
ADMIN_PATH=
ADMIN_LOGIN=
ADMIN_PASSWORD=
ADMIN_MAX_ATTEMPTS=3
ADMIN_LOCKOUT_MINUTES=15
EOF EOF
log ".env записан (прежний, если был, сохранён в .env.install-backup)" log ".env записан (прежний, если был, сохранён в .env.install-backup)"
} }

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,4 +1,9 @@
"""API-приложение расписания (FastAPI).""" """API-приложение расписания (FastAPI).
Кроме публичных данных здесь живёт админка (backend/admin.py): она включается
только если заданы ADMIN_LOGIN и ADMIN_PASSWORD, и открывается по секретному
пути ADMIN_PATH. Все параметры — переменные окружения, см. .env.example.
"""
from __future__ import annotations from __future__ import annotations
@@ -12,15 +17,17 @@ import uvicorn
from dotenv import load_dotenv from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException 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"
@@ -36,36 +43,58 @@ _dotenv_path = _find_dotenv()
if _dotenv_path: if _dotenv_path:
load_dotenv(_dotenv_path) load_dotenv(_dotenv_path)
DEFAULT_INTERVAL_MINUTES = 15
def _normalize_admin_path(raw: str | None) -> str:
"""Секретный путь панели: всегда с ведущим слэшем и без хвостового."""
path = (raw or "").strip()
if not path:
return ""
if not path.startswith("/"):
path = "/" + path
return path.rstrip("/") or "/"
def _int_env(name: str, default: int) -> int:
raw = (os.getenv(name) or "").strip()
if not raw:
return default
try:
return int(raw)
except ValueError:
log.warning("%s=%r не число, беру %d", name, raw, default)
return default
SCHEDULE_LINK = (os.getenv("SCHEDULE_LINK") or "").strip() SCHEDULE_LINK = (os.getenv("SCHEDULE_LINK") or "").strip()
RETRY_AFTER = float(os.getenv("SCHEDULE_RETRY_AFTER", "60")) 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 interval_seconds() -> float: def default_settings() -> SettingsStore:
"""Интервал обновления: SCHEDULE_INTERVAL_MINUTES, иначе старый SCHEDULE_CACHE_TTL (сек).""" return SettingsStore(
raw_minutes = (os.getenv("SCHEDULE_INTERVAL_MINUTES") or "").strip()
if raw_minutes:
try:
return max(1.0, float(raw_minutes)) * 60
except ValueError:
log.warning("SCHEDULE_INTERVAL_MINUTES=%r не число, беру значение по умолчанию", raw_minutes)
raw_ttl = (os.getenv("SCHEDULE_CACHE_TTL") or "").strip()
if raw_ttl:
try:
return max(1.0, float(raw_ttl))
except ValueError:
log.warning("SCHEDULE_CACHE_TTL=%r не число, беру значение по умолчанию", raw_ttl)
return DEFAULT_INTERVAL_MINUTES * 60
def default_cache() -> ScheduleCache:
return ScheduleCache(
link=SCHEDULE_LINK, link=SCHEDULE_LINK,
ttl_seconds=interval_seconds(), interval_minutes=interval_from_env(),
retry_after=RETRY_AFTER, 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:
normalized = day.strip().lower() normalized = day.strip().lower()
if normalized not in parsing.DAYS: if normalized not in parsing.DAYS:
@@ -76,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):
@@ -90,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:

View File

@@ -3,10 +3,13 @@
- однопоточная перезагрузка (lock): параллельные запросы во время загрузки - однопоточная перезагрузка (lock): параллельные запросы во время загрузки
ждут один download вместо N скачиваний; ждут один download вместо N скачиваний;
- интервал обновления: по его истечении сверяем modified таблицы — если файл - интервал обновления: по его истечении сверяем modified таблицы — если файл
не менялся, скачивание не повторяется; не менялся, скачивание не повторяется; принудительное обновление из админки
(force_reload) перечитывает таблицу всегда;
- stale-while-error: если таблица перестала скачиваться, отдаём последнее - stale-while-error: если таблица перестала скачиваться, отдаём последнее
удачное состояние и повторяем попытку не чаще, чем раз в retry_after секунд удачное состояние и повторяем попытку не чаще, чем раз в retry_after секунд
(вместо скачивания на каждый запрос). (вместо скачивания на каждый запрос);
- настройки администратора (ссылка, интервал) живут в SettingsStore и
применяются на лету: смена ссылки проверяется до сохранения.
""" """
from __future__ import annotations from __future__ import annotations
@@ -19,12 +22,13 @@ 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 = "Ссылка на расписание не задана (SCHEDULE_LINK)" NO_LINK_ERROR = "Ссылка на расписание не задана"
class ScheduleCache: class ScheduleCache:
@@ -33,8 +37,15 @@ class ScheduleCache:
link: str | None = 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:
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() self._link = (link or "").strip()
if not self._link and settings is None:
raise ScheduleError("Не задана переменная окружения SCHEDULE_LINK")
self._ttl = float(ttl_seconds) self._ttl = float(ttl_seconds)
self._retry_after = float(retry_after) self._retry_after = float(retry_after)
self._lock = threading.RLock() self._lock = threading.RLock()
@@ -64,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:
@@ -84,20 +100,21 @@ 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,
"interval_minutes": max(1, round(self._ttl / 60)), def state(self) -> dict:
"last_updated": self._iso(self._loaded_at), """Состояние для админки: то же плюс текущая ссылка и настройки."""
"table_modified": self._modified, with self._lock:
"error": self._error, state = self._status()
"groups": len(self._data) if self._data else 0, state["link"] = self._link
"has_data": self._data is not None, if self._settings is not None:
# при последнем чтении файла: изменилось ли его содержимое state["settings_file"] = str(self._settings.path) if self._settings.path else None
"changed": self._changed, state["settings_save_error"] = self._settings.save_error
"changed_at": self._iso(self._changed_at), state["last_attempt"] = self._settings.last_attempt
"content_hash": (self._content_hash or "")[:12], state["last_success"] = self._settings.last_success
} return state
def warmup(self) -> None: def warmup(self) -> None:
"""Первая загрузка данных на старте приложения (ошибка не фатальна).""" """Первая загрузка данных на старте приложения (ошибка не фатальна)."""
@@ -106,6 +123,56 @@ class ScheduleCache:
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
@@ -114,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 = (
@@ -137,6 +218,9 @@ class ScheduleCache:
if not self._link: if not self._link:
self._fail(ScheduleError(NO_LINK_ERROR), now) self._fail(ScheduleError(NO_LINK_ERROR), now)
return 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

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)

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,4 +1,8 @@
"""Фронтенд расписания: статики + прокси к API бэкенда.""" """Фронтенд расписания: статики, прокси к API и страница админки.
Страница админки отдаётся по секретному пути из ADMIN_PATH; сам API админки —
/api/admin/* (проксируется в backend). Если ADMIN_PATH не задан, страницы нет.
"""
from __future__ import annotations from __future__ import annotations
@@ -9,7 +13,8 @@ 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 from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse, JSONResponse from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
@@ -39,7 +44,10 @@ if _dotenv_path:
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 минут), таймаут выше. # Бэкенд на холодном кэше скачивает таблицу (до 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="Расписание — фронт")
@@ -65,6 +73,31 @@ def _proxy(path: str, params: dict) -> object:
return JSONResponse(content=resp.json(), headers={"Cache-Control": "no-store"}) 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-маршруты -----------------------------------------------------------
@@ -118,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("/")
@@ -126,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

@@ -254,11 +254,12 @@ def test_cache_without_link():
from fastapi.testclient import TestClient as TC from fastapi.testclient import TestClient as TC
from schedule import ScheduleCache from schedule import ScheduleCache
from schedule.settings import SettingsStore
cache = ScheduleCache(link="", retry_after=60) cache = ScheduleCache(settings=SettingsStore(link=""), retry_after=60)
app = main.create_app(cache) 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"]

View File

@@ -28,13 +28,24 @@ def test_api_image_copies_every_backend_module():
assert "src/backend/schedule" in stage, "пакет schedule не копируется в образ api" assert "src/backend/schedule" in stage, "пакет schedule не копируется в образ api"
def test_frontend_image_copies_static(): def test_frontend_image_copies_page_and_static():
stage = _stage("frontend") stage = _stage("frontend")
for path in ("src/frontend/main.py", "src/frontend/static"): for path in ("src/frontend/main.py", "src/frontend/admin.html", "src/frontend/static"):
assert path in stage, f"{path} не копируется в образ frontend" assert path in stage, f"{path} не копируется в образ frontend"
def test_compose_passes_required_env(): def test_compose_passes_required_env():
compose = COMPOSE.read_text(encoding="utf-8") compose = COMPOSE.read_text(encoding="utf-8")
for var in ("SCHEDULE_LINK", "SCHEDULE_INTERVAL_MINUTES", "SCHEDULE_API"): 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 var in compose, f"{var} не передаётся контейнерам"
assert "schedule_data:/data" in compose, "том для настроек админки не подключён"

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