Vladislav Kopnin
475eb3801e
В таблице появились группы с «ТОП» в имени (ИВТ-ТОП-11БО, ИТ-ТОП-12БО): шаблон требовал цифру сразу после первого дефиса, поэтому такие столбцы пропускались целиком. Теперь имя группы — буквенные части через дефис, номер потока и необязательный суффикс, т.е. распознаются и «ИВТ-11БО», и «ПИЭ-13МО», и «ИВТ-ТОП-11БО» (похожие заголовки с пробелами — нет). GET /groups больше не сортирует по алфавиту с приоритетом БО/МО, а отдаёт группы в порядке столбцов таблицы — как просили, «как в расписании». Состав групп (появление/переименование/скрытие) подхватывается сам при следующем обновлении, потому что источник истины — заголовок листа. Тесты: ТОП-имена + порядок столбцов, отсев похожих не-групповых заголовков.
289 lines
11 KiB
Python
289 lines
11 KiB
Python
"""API-приложение расписания (FastAPI)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
from contextlib import asynccontextmanager
|
||
from datetime import date
|
||
from pathlib import Path
|
||
|
||
import uvicorn
|
||
from dotenv import load_dotenv
|
||
from fastapi import FastAPI, HTTPException
|
||
|
||
from admin import AdminConfig, create_admin_router
|
||
from schedule import ScheduleCache, ScheduleError
|
||
from schedule import parsing
|
||
from schedule.settings import SettingsStore, interval_from_env
|
||
|
||
log = logging.getLogger(__name__)
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||
|
||
|
||
def _find_dotenv() -> Path | None:
|
||
"""Ищем .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)
|
||
|
||
|
||
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 "/"
|
||
|
||
|
||
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 ""
|
||
|
||
|
||
def default_settings() -> SettingsStore:
|
||
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:
|
||
normalized = day.strip().lower()
|
||
if normalized not in parsing.DAYS:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"День должен быть одним из: {', '.join(parsing.DAYS)}",
|
||
)
|
||
return normalized
|
||
|
||
|
||
def create_app(
|
||
cache: ScheduleCache | None = None,
|
||
admin: AdminConfig | None = None,
|
||
) -> FastAPI:
|
||
cache = cache or default_cache()
|
||
admin = admin if admin is not None else AdminConfig(
|
||
path=ADMIN_PATH, login=ADMIN_LOGIN, password=ADMIN_PASSWORD
|
||
)
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
cache.warmup()
|
||
yield
|
||
|
||
app = FastAPI(
|
||
title="Расписание",
|
||
description="Расписание групп из Яндекс.Таблицы",
|
||
lifespan=lifespan,
|
||
)
|
||
app.state.cache = cache
|
||
app.state.admin = admin
|
||
|
||
if admin.enabled:
|
||
app.include_router(create_admin_router(cache, admin))
|
||
if admin.path:
|
||
log.info("Админка включена (секретный путь задан)")
|
||
else:
|
||
log.warning("Админка включена, но ADMIN_PATH не задан — страница панели недоступна")
|
||
else:
|
||
log.info("Админка выключена: не заданы ADMIN_LOGIN/ADMIN_PASSWORD")
|
||
|
||
def _data() -> tuple[dict, list[dict]]:
|
||
try:
|
||
return cache.get()
|
||
except ScheduleError as exc:
|
||
raise HTTPException(status_code=502, detail=str(exc))
|
||
|
||
def _index() -> tuple[dict, dict]:
|
||
try:
|
||
return cache.get_index()
|
||
except ScheduleError as exc:
|
||
raise HTTPException(status_code=502, detail=str(exc))
|
||
|
||
# --- обслуживание -----------------------------------------------------------
|
||
|
||
@app.get("/healthz")
|
||
def healthz() -> dict:
|
||
"""Живость процесса; не зависит от доступности таблицы."""
|
||
return {"ok": True}
|
||
|
||
@app.get("/status")
|
||
def status() -> dict:
|
||
return cache.status()
|
||
|
||
# --- справочники -------------------------------------------------------------
|
||
|
||
@app.get("/")
|
||
def index() -> dict:
|
||
return {
|
||
"title": "Расписание",
|
||
"endpoints": {
|
||
"/groups": "Список групп",
|
||
"/schedule/{group}": "Расписание группы",
|
||
"/schedule/{group}/{day}": "Расписание группы на день",
|
||
"/today?group={group}": "Расписание группы на сегодня",
|
||
"/teachers": "Список преподавателей",
|
||
"/teacher/{name}": "Расписание преподавателя",
|
||
"/rooms": "Список аудиторий",
|
||
"/room/{label}": "Занятость аудитории",
|
||
"/rooms-day?day={день}": "Занятость всех аудиторий на день",
|
||
"/rooms-free?day={день}&period={N}": "Свободные аудитории",
|
||
"/week": "Недели семестра и текущая чётность",
|
||
"/status": "Состояние кэша",
|
||
},
|
||
}
|
||
|
||
@app.get("/groups")
|
||
def groups() -> list[str]:
|
||
"""Группы в том порядке, в каком идут столбцы в таблице."""
|
||
data, _ = _data()
|
||
return list(data.keys())
|
||
|
||
@app.get("/teachers")
|
||
def teachers() -> list[str]:
|
||
teachers_map, _ = _index()
|
||
return sorted(teachers_map.keys())
|
||
|
||
@app.get("/rooms")
|
||
def rooms() -> list[str]:
|
||
_, rooms_map = _index()
|
||
return parsing.sort_rooms(set(rooms_map) | set(cache.room_info))
|
||
|
||
# --- занятия -----------------------------------------------------------------
|
||
|
||
@app.get("/schedule/{group}")
|
||
def schedule(group: str) -> dict:
|
||
data, _ = _data()
|
||
if group not in data:
|
||
raise HTTPException(status_code=404, detail=f"Группа «{group}» не найдена")
|
||
return parsing.build_group_schedule(data, group)
|
||
|
||
@app.get("/schedule/{group}/{day}")
|
||
def schedule_day(group: str, day: str) -> dict:
|
||
data, _ = _data()
|
||
if group not in data:
|
||
raise HTTPException(status_code=404, detail=f"Группа «{group}» не найдена")
|
||
normalized = _day_arg(day)
|
||
sched = parsing.build_group_schedule(data, group)
|
||
day_data = sched["days"].get(normalized, {})
|
||
return {"group": group, "day": normalized, "periods": day_data}
|
||
|
||
@app.get("/today")
|
||
def today(group: str) -> dict:
|
||
data, _ = _data()
|
||
if group not in data:
|
||
raise HTTPException(status_code=404, detail=f"Группа «{group}» не найдена")
|
||
day_name = parsing.DAYS[date.today().weekday()]
|
||
sched = parsing.build_group_schedule(data, group)
|
||
return {
|
||
"group": group,
|
||
"date": date.today().isoformat(),
|
||
"day": day_name,
|
||
"periods": sched["days"].get(day_name, {}),
|
||
}
|
||
|
||
@app.get("/teacher/{name}")
|
||
def teacher(name: str) -> dict:
|
||
teachers_map, _ = _index()
|
||
result = parsing.build_teacher_schedule(teachers_map, name)
|
||
if result is None:
|
||
raise HTTPException(status_code=404, detail=f"Преподаватель «{name}» не найден")
|
||
result["full_name"] = cache.teacher_full.get(name, name)
|
||
return result
|
||
|
||
@app.get("/room/{label}")
|
||
def room(label: str) -> dict:
|
||
_, rooms_map = _index()
|
||
result = parsing.build_room_schedule(rooms_map, label, cache.get_day_times())
|
||
if result is None:
|
||
raise HTTPException(status_code=404, detail=f"Аудитория «{label}» не найдена")
|
||
result["info"] = cache.room_info.get(label, {}).get("info", "")
|
||
return result
|
||
|
||
@app.get("/rooms-free")
|
||
def rooms_free(day: str, period: int) -> list[dict]:
|
||
normalized = _day_arg(day)
|
||
if period < 1 or period > 8:
|
||
raise HTTPException(status_code=400, detail="Пара должна быть от 1 до 8")
|
||
_, rooms_map = _index()
|
||
room_info = cache.room_info
|
||
free = []
|
||
for label in parsing.sort_rooms(set(rooms_map) | set(room_info)):
|
||
occupied = rooms_map.get(label, {}).get(normalized, {}).get(str(period), {})
|
||
busy = bool((occupied.get("ч") or []) or (occupied.get("з") or []))
|
||
if not busy:
|
||
free.append({"room": label, "info": room_info.get(label, {}).get("info", "")})
|
||
return free
|
||
|
||
@app.get("/rooms-day")
|
||
def rooms_day(day: str) -> dict:
|
||
normalized = _day_arg(day)
|
||
_, rooms_map = _index()
|
||
room_info = cache.room_info
|
||
day_times = cache.get_day_times().get(normalized, {})
|
||
rooms_out = []
|
||
for label in parsing.sort_rooms(set(rooms_map) | set(room_info)):
|
||
info = room_info.get(label, {}).get("info", "")
|
||
occupied_day = rooms_map.get(label, {}).get(normalized, {})
|
||
periods = {}
|
||
for period in sorted(set(occupied_day) | set(day_times), key=int):
|
||
slot = occupied_day.get(period, {})
|
||
periods[period] = {
|
||
"time": slot.get("time") or day_times.get(period, ""),
|
||
"ч": slot.get("ч") or [],
|
||
"з": slot.get("з") or [],
|
||
}
|
||
rooms_out.append({"room": label, "info": info, "periods": periods})
|
||
return {"day": normalized, "rooms": rooms_out}
|
||
|
||
@app.get("/week")
|
||
def week() -> dict:
|
||
_, weeks = _data()
|
||
today = date.today()
|
||
return {
|
||
"today": today.isoformat(),
|
||
"parity": parsing.current_parity(weeks, today),
|
||
"weeks": [
|
||
{"parity": w["parity"], "start": w["start"].isoformat(), "end": w["end"].isoformat()}
|
||
for w in weeks
|
||
],
|
||
}
|
||
|
||
return app
|
||
|
||
|
||
app = create_app()
|
||
|
||
if __name__ == "__main__":
|
||
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
|