"""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 schedule import ScheduleCache, ScheduleError from schedule import parsing 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) DEFAULT_INTERVAL_MINUTES = 15 SCHEDULE_LINK = (os.getenv("SCHEDULE_LINK") or "").strip() RETRY_AFTER = float(os.getenv("SCHEDULE_RETRY_AFTER", "60")) def interval_seconds() -> float: """Интервал обновления: SCHEDULE_INTERVAL_MINUTES, иначе старый SCHEDULE_CACHE_TTL (сек).""" 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, ttl_seconds=interval_seconds(), 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) -> FastAPI: cache = cache or default_cache() @asynccontextmanager async def lifespan(app: FastAPI): cache.warmup() yield app = FastAPI( title="Расписание", description="Расписание групп из Яндекс.Таблицы", lifespan=lifespan, ) app.state.cache = cache 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)