Mermutef/schedule
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Бэкенд: пакет schedule/ по зонам ответственности, обобщённый разбор ячеек

Browse Source 
- xlsx.py: чтение книги (мержи, скрытые столбцы, красный текст), фикс дат до 1900
- lessons.py: общая схема ролей (предмет/преподаватель/аудитория/заметка)
  вместо ветвлений по числу строк; маркеры к.в., многострочные заметки,
  ФИО с/без отчества, организации, локации, метки групп -> заметки
- source.py: загрузка с Яндекс.Диска с понятными ошибками
- parsing.py: листы -> расписание, недели семестра, индексы преподов/аудиторий
- cache.py: lock, TTL, проверка modified, stale-while-error с retry cooldown
- main.py: create_app, /healthz, логирование; force-обновление убрано (вернётся с авторизацией)
- переменная окружения SCHEDULE_LINK
This commit is contained in:
Vladislav Kopnin
2026-09-09 11:33:03 +03:00
parent dcd9b854a0
commit 86c2a3097d
10 changed files with 1601 additions and 1057 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.env.example
View File

@@ -1,5 +1,6 @@
schedule_link=
SCHEDULE_LINK=
SCHEDULE_CACHE_TTL=900
SCHEDULE_RETRY_AFTER=60
API_PORT=8000
FRONTEND_PORT=8080
SCHEDULE_API=http://api:8000

486
src/backend/main.py
View File

@@ -1,33 +1,27 @@
"""API-приложение расписания (FastAPI)."""
from __future__ import annotations
import logging
import os
import time
import threading
from contextlib import asynccontextmanager
from datetime import datetime
from datetime import date
from pathlib import Path
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Query
from schedule_parser import (
DAYS,
ScheduleError,
build_day_times,
build_group_schedule,
build_room_schedule,
build_teacher_room_index,
build_teacher_schedule,
current_parity,
download_schedule,
extract_week_ranges,
fetch_table_modified,
parse_rooms_info,
parse_schedule,
parse_teachers_full,
parse_workbook,
sort_rooms,
)
from schedule import ScheduleCache, ScheduleError
from schedule import parsing
def _find_dotenv():
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"
@@ -43,256 +37,234 @@ _dotenv_path = _find_dotenv()
if _dotenv_path:
load_dotenv(_dotenv_path)
SCHEDULE_LINK = os.getenv("schedule_link") or os.getenv("SCHEDULE_LINK", "").strip()
CACHE_TTL = int(os.getenv("SCHEDULE_CACHE_TTL", "900"))
_cache = {
"data": None,
"weeks": None,
"modified": None,
"loaded_at": 0.0,
"error": None,
"index": None,
"teacher_full": {},
"room_info": {},
}
SCHEDULE_LINK = (os.getenv("SCHEDULE_LINK") or "").strip()
CACHE_TTL = float(os.getenv("SCHEDULE_CACHE_TTL", "900"))
RETRY_AFTER = float(os.getenv("SCHEDULE_RETRY_AFTER", "60"))
@asynccontextmanager
async def lifespan(app: FastAPI):
def _default_cache() -> ScheduleCache:
try:
_load_data()
except Exception:
pass
yield
return ScheduleCache(SCHEDULE_LINK, ttl_seconds=CACHE_TTL, retry_after=RETRY_AFTER)
except ScheduleError:
log.error("SCHEDULE_LINK не задан — API данных работать не будет")
return _BrokenCache()
app = FastAPI(
title="Расписание",
description="Расписание групп из Яндекс.Таблицы",
lifespan=lifespan,
)
class _BrokenCache(ScheduleCache):
"""Кэш без ссылки: все запросы данных отвечают понятной ошибкой."""
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:
raise ScheduleError(self._error)
def _load_data(force: bool = False):
if not SCHEDULE_LINK:
raise ScheduleError("Ссылка schedule_link не задана в .env")
now = time.time()
if (
not force
and _cache["data"] is not None
and now - _cache["loaded_at"] < CACHE_TTL
and _cache["error"] is None
):
return _cache["data"], _cache["weeks"]
modified = fetch_table_modified(SCHEDULE_LINK)
if (
not force
and modified
and modified == _cache["modified"]
and _cache["data"] is not None
and _cache["error"] is None
):
_cache["loaded_at"] = now
return _cache["data"], _cache["weeks"]
xlsx = download_schedule(SCHEDULE_LINK)
sheets = parse_workbook(xlsx)
data = parse_schedule(sheets)
weeks = extract_week_ranges(sheets)
_cache["data"] = data
_cache["weeks"] = weeks
_cache["modified"] = modified
_cache["loaded_at"] = now
_cache["error"] = None
_cache["index"] = None
_cache["teacher_full"] = parse_teachers_full(sheets)
_cache["room_info"] = parse_rooms_info(sheets)
return data, weeks
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 _get_index(force: bool = False):
data, _ = _get_data(force=force)
if _cache["index"] is None:
_cache["index"] = build_teacher_room_index(data)
return _cache["index"]
def create_app(cache: ScheduleCache | None = None) -> FastAPI:
cache = cache or _default_cache()
@asynccontextmanager
async def lifespan(app: FastAPI):
cache.warmup()
yield
def _get_data(force: bool = False):
try:
return _load_data(force=force)
except ScheduleError as exc:
_cache["error"] = str(exc)
raise HTTPException(status_code=502, detail=str(exc))
@app.get("/")
def index():
return {
"title": "Расписание",
"endpoints": {
"/groups": "Список групп",
"/schedule/{group}": "Расписание группы",
"/schedule/{group}/{day}": "Расписание группы на день",
"/today?group={group}": "Расписание группы на сегодня",
},
}
@app.get("/groups")
def groups(force: bool = Query(False)):
data, _ = _get_data(force=force)
return sorted(
data.keys(),
key=lambda g: (
0 if g.endswith("БО") else (1 if g.endswith("МО") else 2),
g,
),
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 sorted(
data.keys(),
key=lambda g: (0 if g.endswith("БО") else (1 if g.endswith("МО") else 2), g),
)
@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.get("/teachers")
def teachers(force: bool = Query(False)):
teachers_map, _ = _get_index(force=force)
return sorted(teachers_map.keys())
@app.get("/rooms")
def rooms(force: bool = Query(False)):
_, rooms_map = _get_index(force=force)
return sort_rooms(rooms_map.keys())
@app.get("/teacher/{name}")
def teacher(name: str, force: bool = Query(False)):
teachers_map, _ = _get_index(force=force)
result = build_teacher_schedule(teachers_map, name)
if result is None:
raise HTTPException(status_code=404, detail=f"Преподаватель «{name}» не найден")
result["full_name"] = _cache.get("teacher_full", {}).get(name, name)
return result
@app.get("/room/{label}")
def room(label: str, force: bool = Query(False)):
data, _ = _get_data(force=force)
_, rooms_map = _get_index(force=force)
result = build_room_schedule(rooms_map, label, build_day_times(data))
if result is None:
raise HTTPException(status_code=404, detail=f"Аудитория «{label}» не найдена")
result["info"] = _cache.get("room_info", {}).get(label, {}).get("info", "")
return result
@app.get("/rooms-free")
def rooms_free(
day: str,
period: int,
force: bool = Query(False),
):
normalized = day.strip().lower()
if normalized not in DAYS:
raise HTTPException(status_code=400, detail=f"День должен быть одним из: {', '.join(DAYS)}")
if period < 1 or period > 8:
raise HTTPException(status_code=400, detail="Пара должна быть от 1 до 8")
_, rooms_map = _get_index(force=force)
room_info = _cache.get("room_info", {})
free = []
for label in sort_rooms(room_info.keys()):
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[label].get("info", "")})
return free
@app.get("/rooms-day")
def rooms_day(day: str, force: bool = Query(False)):
normalized = day.strip().lower()
if normalized not in DAYS:
raise HTTPException(status_code=400, detail=f"День должен быть одним из: {', '.join(DAYS)}")
data, _ = _get_data(force=force)
_, rooms_map = _get_index(force=force)
room_info = _cache.get("room_info", {})
day_times = build_day_times(data).get(normalized, {})
labels = set(rooms_map.keys()) | set(room_info.keys())
rooms_out = []
for label in sort_rooms(labels):
info = room_info.get(label, {}).get("info", "")
occupied_day = rooms_map.get(label, {}).get(normalized, {})
periods = {}
keys = set(occupied_day.keys()) | set(day_times.keys())
for period in sorted(keys, key=lambda p: int(p)):
slot = occupied_day.get(period, {})
periods[str(period)] = {
"time": day_times.get(period, slot.get("time", "")),
"ч": 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(force: bool = Query(False)):
_, weeks = _get_data(force=force)
today = datetime.now().date()
return {
"today": today.isoformat(),
"parity": current_parity(weeks, today),
"weeks": [
{"parity": w["parity"], "start": w["start"].isoformat(), "end": w["end"].isoformat()}
for w in weeks
],
}
@app.get("/status")
def status(force: bool = Query(False)):
_get_data(force=force)
return {
"cache_ttl_seconds": CACHE_TTL,
"last_updated": (
datetime.fromtimestamp(_cache["loaded_at"], datetime.now().astimezone().tzinfo).isoformat() if _cache["loaded_at"] else None
),
"table_modified": _cache["modified"],
}
@app.get("/schedule/{group}")
def schedule(group: str, force: bool = Query(False)):
data, _ = _get_data(force=force)
if group not in data:
raise HTTPException(status_code=404, detail=f"Группа «{group}» не найдена")
return build_group_schedule(data, group)
@app.get("/schedule/{group}/{day}")
def schedule_day(group: str, day: str, force: bool = Query(False)):
data, _ = _get_data(force=force)
if group not in data:
raise HTTPException(status_code=404, detail=f"Группа «{group}» не найдена")
normalized = day.strip().lower()
if normalized not in DAYS:
raise HTTPException(status_code=400, detail=f"День должен быть одним из: {', '.join(DAYS)}")
sched = 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, force: bool = Query(False)):
data, _ = _get_data(force=force)
if group not in data:
raise HTTPException(status_code=404, detail=f"Группа «{group}» не найдена")
weekday = datetime.now().weekday()
day_name = DAYS[weekday]
sched = build_group_schedule(data, group)
return {"group": group, "date": datetime.now().date().isoformat(), "day": day_name, "periods": sched["days"].get(day_name, {})}
app = create_app()
if __name__ == "__main__":
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

14
src/backend/schedule/__init__.py Normal file
View File

@@ -0,0 +1,14 @@
"""Пакет для загрузки и разбора расписания из Яндекс.Таблиц (xlsx).
Структура:
xlsx — чтение xlsx-рабочей книги (ячейки, мержи, скрытые столбцы, красный текст);
source — загрузка файла по публичной ссылке Яндекс.Диска;
lessons — разбор текста ячейки занятия (предмет/преподаватель/аудитория/заметки);
parsing — разбор листов: группы, недели, индексы преподавателей и аудиторий;
cache — кэш загруженной таблицы (TTL, перезагрузка, stale-while-error).
"""
from .cache import ScheduleCache
from .errors import ScheduleError
__all__ = ["ScheduleCache", "ScheduleError"]

170
src/backend/schedule/cache.py Normal file
View File

@@ -0,0 +1,170 @@
"""Кэш загруженного расписания.
- однопоточная перезагрузка (lock): параллельные запросы во время загрузки
ждут один download вместо N скачиваний;
- TTL, а также проверка modified таблицы: если таблица не менялась,
скачивание не повторяется;
- stale-while-error: если таблица перестала скачиваться, отдаём последнее
удачное состояние и повторяем попытку не чаще, чем раз в retry_after секунд
(вместо скачивания на каждый запрос).
"""
from __future__ import annotations
import logging
import threading
import time
from datetime import datetime
from . import parsing
from .errors import ScheduleError
from .source import download_schedule, fetch_table_modified
from .xlsx import parse_workbook
log = logging.getLogger(__name__)
class ScheduleCache:
def __init__(
self,
link: str | None,
ttl_seconds: float = 900.0,
retry_after: float = 60.0,
) -> None:
if not link:
raise ScheduleError("Не задана переменная окружения SCHEDULE_LINK")
self._link = link
self._ttl = ttl_seconds
self._retry_after = retry_after
self._lock = threading.RLock()
self._data: dict | None = None
self._weeks: list[dict] = []
self._index: tuple[dict, dict] | None = None
self._day_times: dict = {}
self._teacher_full: dict = {}
self._room_info: dict = {}
self._loaded_at: float = 0.0
self._modified: str | None = None
self._error: str | None = None
self._last_failed: float = 0.0
# --- публичное API ---------------------------------------------------------
@property
def teacher_full(self) -> dict:
return self._teacher_full
@property
def room_info(self) -> dict:
return self._room_info
def get(self) -> tuple[dict, list[dict]]:
"""(data, weeks); бросает ScheduleError, если данных нет совсем."""
with self._lock:
self._ensure_fresh()
assert self._data is not None
return self._data, self._weeks
def get_index(self) -> tuple[dict, dict]:
"""(teachers, rooms) — индексы занятий по преподавателям и аудиториям."""
with self._lock:
self._ensure_fresh()
assert self._index is not None
return self._index
def get_day_times(self) -> dict:
with self._lock:
self._ensure_fresh()
return self._day_times
def status(self) -> dict:
with self._lock:
return {
"cache_ttl_seconds": self._ttl,
"last_updated": self._iso(self._loaded_at),
"table_modified": self._modified,
"error": self._error,
}
def warmup(self) -> None:
"""Первый запуск данных на старте приложения (ошибка не фатальна)."""
try:
self.get()
except ScheduleError as exc:
log.warning("Начальная загрузка расписания не удалась: %s", exc)
# --- internals --------------------------------------------------------------
@staticmethod
def _iso(ts: float) -> str | None:
if not ts:
return None
return datetime.fromtimestamp(ts, datetime.now().astimezone().tzinfo).isoformat()
def _ensure_fresh(self) -> None:
now = time.time()
fresh = (
self._data is not None
and self._loaded_at
and self._error is None
and now - self._loaded_at < self._ttl
)
if fresh:
return
self._reload(now)
def _reload(self, now: float) -> None:
# служим старые данные, пока не истекло время повторной попытки
if (
self._data is not None
and self._error is not None
and now - self._last_failed < self._retry_after
):
return
modified = fetch_table_modified(self._link)
if (
self._data is not None
and modified is not None
and modified == self._modified
):
# таблица не менялась — источник доступен, данные актуальны
self._loaded_at = now
self._error = None
return
try:
log.info("Загружаю расписание по ссылке %s ...", self._link)
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:
self._fail(exc, now)
except Exception as exc: # непредвиденное в парсинге/сети
log.exception("Ошибка при загрузке расписания")
self._fail(ScheduleError(f"Не удалось обработать таблицу: {exc}"), now)
def _fail(self, exc: ScheduleError, now: float) -> None:
self._error = str(exc)
self._last_failed = now
if self._data is None:
raise exc
log.warning(
"Обновление не удалось (%s), отдам предыдущее состояние ещё %.0f с",
exc,
self._retry_after,
)

5
src/backend/schedule/errors.py Normal file
View File

@@ -0,0 +1,5 @@
"""Исключения пакета schedule, которые показывают пользователю как текст ошибки."""
class ScheduleError(Exception):
"""Ошибка загрузки или разбора расписания (ссылка, сеть, xlsx, лист)."""

511
src/backend/schedule/lessons.py Normal file
View File

@@ -0,0 +1,511 @@
"""Разбор текста ячейки занятия.
В одной ячейке может быть одно или несколько занятий, оформлено строго
необязательное: маркер, время, локация, примечания пишутся как придумается.
Поэтому разбор не ветвится по количеству строк, а проходит один конвейер:
1. красные фрагменты таблицы (⟪...⟫) выносятся в заметки и удаляются из текста —
они служат только примечаниями, даже если внутри «вложено» другое занятие;
2. извлекается маркер «к.в.» (своя строка или префикс первой строки);
3. каждая строка классифицируется по ролям:
- аудитория — есть «ауд.»/«каб.», URL, «онлайн», «зал...» (вместе с названием
корпуса, номерами вариантов, временем начала);
- преподаватель — фамилия + инициалы или полное ФИО, организация;
в строке с преподавателем может присутствовать аудитория;
- предмет — всё, что идёт до преподавателя/аудитории;
- заметка — время/даты/ссылки;
- организация — короткое имя капсом («НПО Криста», «ПравоТех»): преподаватель,
если преподавателя ещё нет, иначе — место проведения.
4. остаток собирается по позиции: предмет сначала, после преподавателя/аудитории
неклассифицированное без аудитории — локация (аудитория), дальше — заметки.
"""
from __future__ import annotations
import re
from dataclasses import asdict, dataclass, replace
from typing import Optional
from .xlsx import RED_CLOSE, RED_OPEN
# --- маркер недели занятия («к.в.» = курсовая/ведомостная неделя) ---------------
_MARKER_LINE_RE = re.compile(r"^\s*к\.в\.?\s*[-–—]?\s*(\d{1,2})?\s*$", re.I)
_MARKER_PREFIX_RE = re.compile(r"^\s*(к\.в\.?\s*[-–—]?\s*(\d{1,2})?)\s+(?=\S)", re.I)
# --- роли -------------------------------------------------------------------
_TIME_RE = re.compile(r"\d{1,2}:\d{2}|\d{1,2}\.\d{1,2}\.\d{2,4}")
_URL_RE = re.compile(r"https?://\S+")
_ROOM_REF_RE = re.compile(r"(?:ауд|каб)\.?\s*[\w\-]*", re.I)
_ONLINE_RE = re.compile(r"\bонлайн\b", re.I)
_ROOM_START_RE = re.compile(r"^\s*(?:зал|каб)\b", re.I)
# время в начале строки: «начало в 12:00», «с 13:45 до 15:15»
_TIME_PREFIX_RE = re.compile(
r"^(?P<time>.*?\d{1,2}:\d{2}(?:\s*(?:до|—|–|-)\s*\d{1,2}:\d{2})?)\s+(?P<rest>.*)$"
)
# Имя: «Фамилия И.И.», «Фамилия А.М.», «Фамилия Г.К» — минимум одна точка
# (чтобы «Группа Б» не принималась за преподавателя)
_NAME_INITIALS_RE = re.compile(
r"[А-ЯЁ][а-яё-]{1,}\s+(?:[А-ЯЁ]\.(?:\s*[А-ЯЁ]\.{0,1}){0,1}|[А-ЯЁ]\.{0,1}\s+[А-ЯЁ]\.)"
)
# Полное ФИО в начале строки: три и более слова, каждое с большой и не короче 3 букв
# (под предлогами «с/в/и/по» предметы не путаются)
_FULL_NAME_RE = re.compile(
r"^[А-ЯЁ][а-яё-]{2,}(?:\s+[А-ЯЁ][а-яё-]{2,}){2,}(?:\s*\([^()]*\))?"
)
# Организация/локация: «НПО Криста», «ПравоТех», «МАИС»
_ORG_WORD_RE = re.compile(r"^[A-ZА-ЯЁ][A-Za-zА-ЯЁа-яё0-9&\-]*$")
def _is_org_line(line: str) -> bool:
"""Организация или аббревиатура-локация (но не «Деловое общение»)."""
if len(line) > 30 or _TIME_RE.search(line):
return False
words = line.split()
if not 1 <= len(words) <= 3:
return False
if any(not _ORG_WORD_RE.match(w) for w in words):
return False
if len(words) == 1:
# одно слово: аббревиатура (МАИС, ДСП) или camelcase-название (ПравоТех)
return words[0].isupper() and len(words[0]) <= 12 or re.search(
r"[a-zа-яё][A-ZА-ЯЁ]", words[0]
)
# несколько слов: первое — короче 6 букв и все заглавные (НПО, НИУ)
return words[0].isupper() and len(words[0]) <= 6
_PAREN_RE = re.compile(r"\(([^()]*)\)")
_ROOM_WORD_RE = re.compile(r"(?:ауд|каб)\.?\s*", re.I)
@dataclass(frozen=True)
class Lesson:
marker: str = ""
subject: str = ""
teacher: str = ""
room: str = ""
note: str = ""
def to_dict(self) -> dict:
return asdict(self)
def _normalize_marker(num: Optional[str]) -> str:
return f"к.в. - {num}" if num else "к.в."
def _squish(text: str) -> str:
return re.sub(r"\s{2,}", " ", text).strip()
def _clean_name_parens(line: str) -> tuple[str, list[str]]:
"""Скобки в строке преподавателя: со временем/датой -> заметка, остальное остаётся."""
notes: list[str] = []
def repl(m: re.Match) -> str:
inner = m.group(1).strip()
if not inner:
return "" # пустые скобки после выноса красного фрагмента
if inner and _TIME_RE.search(inner):
notes.append(inner)
return ""
return m.group(0)
return _squish(_PAREN_RE.sub(repl, line)), notes
def _clean_room_parens(line: str) -> tuple[str, list[str]]:
"""Скобки в строке аудитории: «(221)»/«(л)» остаются, пояснения -> заметки."""
notes: list[str] = []
def repl(m: re.Match) -> str:
inner = m.group(1).strip()
keep = (
not inner
or len(inner) <= 2
or inner.isupper()
or not re.search(r"[а-яёА-ЯЁ]", inner)
)
if keep:
return m.group(0)
notes.append(inner)
return ""
cleaned = _PAREN_RE.sub(repl, line)
cleaned = _ROOM_WORD_RE.sub(
lambda m: "ауд. " if m.group(0)[:3].lower() == "ауд" else "каб. ", cleaned
)
return _squish(cleaned).strip(" ,"), notes
@dataclass
class _LineRoles:
subject: str = ""
teacher: str = ""
room: str = ""
org: str = ""
notes: tuple[str, ...] = ()
def _line_is_room(line: str) -> bool:
core = _PAREN_RE.sub("", line) # «… (онлайн)» — маркер предмета, не аудитория
return bool(
_ROOM_REF_RE.search(core)
or _URL_RE.search(line)
or _ONLINE_RE.search(core)
or _ROOM_START_RE.match(core)
)
def _classify_line(line: str) -> _LineRoles:
roles = _LineRoles()
if _line_is_room(line):
rest = line
# «начало занятия в 18:30 онлайн» -> время в заметку, аудитория отдельно
prefix_m = _TIME_PREFIX_RE.match(rest)
if prefix_m and _line_is_room(prefix_m.group("rest")):
roles.notes = (prefix_m.group("time").strip(),)
rest = prefix_m.group("rest")
initial_matches = list(_NAME_INITIALS_RE.finditer(rest))
full_m = _FULL_NAME_RE.match(rest)
# инициалы весомее полного имени: «Сети Петри Башкин В.А. ауд. 224» —
# три слова предмета, а не ФИО
if initial_matches:
first = initial_matches[0].start()
tail = initial_matches[-1].end()
teacher_text = _squish(" ".join(m.group(0) for m in initial_matches))
elif full_m is not None:
first, tail = 0, full_m.end()
teacher_text = full_m.group(0)
else:
first, tail, teacher_text = -1, 0, ""
# до аудитории без имени: предмет и (реже) организация-преподаватель
anchors = [
m.start()
for m in (
_ROOM_REF_RE.search(rest),
_URL_RE.search(rest),
_ONLINE_RE.search(rest),
_ROOM_START_RE.match(rest),
)
if m is not None
]
if anchors:
pos = min(anchors)
if pos > 0 and rest[:1].isupper():
pre = _squish(rest[:pos])
# «Методы сжатия КрасновМ.В.» — инициалы без пробела
m_unsp = re.search(
r"([А-ЯЁ][А-ЯЁа-яё\-]*\s*[А-ЯЁ]\.\s*[А-ЯЁ]?\.*)$", pre
)
if m_unsp:
teacher_text = m_unsp.group(1).strip()
pre = _squish(pre[: m_unsp.start()])
words = pre.split()
last = words[-1] if words else ""
if len(words) >= 3 and len(last) >= 4 and not last.isupper():
# «...модели принятия решения Наука ауд.221»
teacher_text = last
pre = _squish(" ".join(words[:-1]))
if pre:
roles.subject = pre
tail = pos
# иначе «юридический корпус, 3 этаж ауд. 22» — целиком локация
if first >= 0 and not roles.subject and first > 0:
roles.subject = _squish(rest[:first])
if teacher_text:
roles.teacher, name_notes = _clean_name_parens(teacher_text)
roles.notes = roles.notes + tuple(name_notes)
room_part = _squish(rest[tail:])
if room_part:
roles.room, room_notes = _clean_room_parens(room_part)
roles.notes = roles.notes + tuple(room_notes)
return roles
initial_matches = list(_NAME_INITIALS_RE.finditer(line))
if initial_matches:
first = initial_matches[0].start()
if first > 0:
roles.subject = _squish(line[:first])
roles.teacher, notes = _clean_name_parens(line[first:])
roles.notes = tuple(notes)
return roles
full_m = _FULL_NAME_RE.match(line)
if full_m is not None:
roles.teacher, notes = _clean_name_parens(full_m.group(0))
if full_m.end() < len(line):
roles.subject = _squish(line[full_m.end() :])
roles.notes = tuple(notes)
return roles
if _TIME_RE.search(line):
frag = line.strip()
if frag.startswith("(") and frag.endswith(")") and frag.count("(") == 1:
frag = frag[1:-1].strip()
roles.notes = (frag,)
return roles
if _is_org_line(line):
roles.org = line
return roles
subject = _squish(line)
# «Методы сжатия КрасновМ.В.» — инициалы без пробела
if " " in subject:
m = re.search(r"^(\S.*\s+)([А-ЯЁ][А-ЯЁа-яё\-]*\s*[А-ЯЁ]\.\s*[А-ЯЁ]?\.*)[\s,]*$", subject)
if m:
roles.subject, roles.teacher = m.group(1).strip(), m.group(2).strip()
return roles
# «Предмет Денис Верховцев (Сигма)» — ФИО без отчества в конце строки
m = re.search(
r"^(.*\s)([А-ЯЁ][а-яё]{2,}\s+[А-ЯЁ][а-яё]{2,})(\s*\([^()]*\))?\s*$", subject
)
if m and len(subject.split()) >= 4:
teacher = _squish(m.group(2) + " " + (m.group(3) or ""))
roles.subject, roles.teacher = m.group(1).strip(), teacher
return roles
roles.subject = subject
return roles
def _line_kind(line: str) -> str:
"""Вид строки только для разбиения ячейки на блоки занятий."""
if _MARKER_LINE_RE.match(line) or _MARKER_PREFIX_RE.match(line):
return "marker"
if _line_is_room(line):
return "room"
if (
_FULL_NAME_RE.match(line)
or _NAME_INITIALS_RE.search(line)
or _TIME_RE.search(line)
):
return "not-subject"
if _is_org_line(line):
return "org"
if line[:1].isupper():
return "subject"
return "other"
def _starts_new_lesson(line: str) -> bool:
"""Строка начинает новое занятие: с прописной, и не сама аудитория/ссылка/локация."""
if not line[:1].isupper():
return False
if re.match(r"(?i)^(?:ауд|каб)\.?", line):
return False
if _URL_RE.match(line) or re.match(r"(?i)^онлайн\b", line):
return False
if _ROOM_START_RE.match(line):
return False
return True
def _split_blocks(lines: list[str]) -> list[list[str]]:
"""Ячейка -> блоки занятий.
Новый блок начинается:
- после пустой строки;
- на строке с маркером «к.в.»;
- когда после аудитории или преподавателя строка начинается с названия
предмета (даже если она сама содержит аудиторию, «Методы сжатия ... ауд.231»).
Организация/локация («НПО Криста», «ИТ-11») новым занятием не считается.
"""
blocks: list[list[str]] = []
current: list[str] = []
prev_kind = ""
for raw in lines:
line = raw.strip()
if not line:
if current:
blocks.append(current)
current = []
prev_kind = ""
continue
kind = _line_kind(line)
if current and (
kind == "marker"
or (
prev_kind in ("room", "not-subject")
and kind != "org"
and _starts_new_lesson(line)
)
):
blocks.append(current)
current = []
current.append(line)
prev_kind = kind
if current:
blocks.append(current)
return blocks
def _extract_marker(lines: list[str]) -> tuple[str, list[str]]:
first = lines[0]
m = _MARKER_LINE_RE.match(first)
if m:
return _normalize_marker(m.group(1)), lines[1:]
m = _MARKER_PREFIX_RE.match(first)
if m:
rest = first[m.end(1) :].strip()
lines = lines[1:]
if rest:
lines = [rest, *lines]
return _normalize_marker(m.group(2)), lines
return "", lines
def _red_fragments(text: str) -> tuple[list[str], str]:
"""Красные фрагменты -> примечания; текст остаётся на месте (пробелом)."""
frag_re = re.escape(RED_OPEN) + r"(.*?)" + re.escape(RED_CLOSE)
frags = []
for m in re.finditer(frag_re, text, re.S):
frag = _squish(m.group(1))
if frag.startswith("(") and frag.endswith(")"):
frag = frag[1:-1].strip()
if frag:
frags.append(frag)
return frags, re.sub(frag_re, " ", text, flags=re.S)
def _balanced(text: str) -> str:
"""Убирает осиротевшие скобки (в таблице их иногда не закрывают)."""
text = text.strip()
while text and (
(text[-1] == "(" and text.count("(") > text.count(")"))
or (text[-1] == ")" and text.count(")") > text.count("("))
):
text = text[:-1].rstrip()
return text
def _parse_block(block: list[str]) -> Optional[Lesson]:
text = "\n".join(block)
red_notes, text = _red_fragments(text)
lines = [ln.strip() for ln in text.split("\n") if ln.strip()]
if not lines:
if red_notes:
# в ячейке только примечание (например, «пары не будет»)
return Lesson(note="\n".join(dict.fromkeys(red_notes)))
return None
marker, lines = _extract_marker(lines)
if not lines:
if red_notes:
# в ячейке только примечание (например, «пары не будет»)
return Lesson(marker=marker, note="\n".join(dict.fromkeys(red_notes)))
return None # в ячейке только маркер — занятия ещё нет
subjects: list[str] = []
teachers: list[str] = []
rooms: list[str] = []
notes: list[str] = list(red_notes)
person_seen = False
for line in lines:
roles = _classify_line(line)
notes.extend(roles.notes)
if roles.subject:
if not person_seen:
# однословная строка в позиции преподавателя — организация: «Хинт»
if (
subjects
and " " not in roles.subject
and not re.search(r"\d", roles.subject)
and 2 <= len(roles.subject) <= 12
):
teachers.append(roles.subject)
person_seen = True
else:
subjects.append(roles.subject)
elif not rooms:
rooms.append(roles.subject) # локация без номера: «первый этаж, ...»
else:
notes.append(roles.subject) # «ссылка в эл. университете»
if roles.teacher:
teachers.append(roles.teacher)
person_seen = True
if roles.room:
rooms.append(roles.room)
person_seen = True
if roles.org:
if not person_seen and not subjects:
subjects.append(roles.org) # «МАИС», «БЖД» — предмет-аббревиатура
elif " " not in roles.org and re.search(r"\d", roles.org):
notes.append(roles.org) # «ИТ-11» — метка параллельной группы
elif teachers:
rooms.append(roles.org) # место проведения: «НПО Криста»
else:
teachers.append(roles.org) # занятие ведёт организация: «ПравоТех»
return Lesson(
marker=marker,
subject=_squish(" ".join(subjects)),
teacher=_balanced(_squish(", ".join(dict.fromkeys(t for t in teachers if t)))),
room=_balanced(_squish(" / ".join(dict.fromkeys(r for r in rooms if r)))),
note=_balanced("\n".join(dict.fromkeys(n for n in notes if n))),
)
def _normalize_marker_of(line: str) -> Optional[str]:
m = _MARKER_LINE_RE.match(line) or _MARKER_PREFIX_RE.match(line)
if not m:
return None
num = m.group(1) if m.re is _MARKER_LINE_RE else m.group(2)
return _normalize_marker(num or None)
def parse_lesson_cell(cell_text: str) -> list[Lesson]:
"""Текст ячейки -> список занятий."""
if not cell_text or not cell_text.strip():
return []
lines = cell_text.replace("\r\n", "\n").replace("\r", "\n").split("\n")
lessons: list[Lesson] = []
last_marker = ""
for block in _split_blocks(lines):
if len(block) == 1 and not _line_is_room(block[0]):
standalone = _normalize_marker_of(block[0])
if standalone:
last_marker = standalone
continue
lesson = _parse_block(block)
if lesson is None:
continue
if lesson.marker:
last_marker = lesson.marker
elif last_marker:
lesson = replace(lesson, marker=last_marker)
lessons.append(lesson)
return lessons
# --- извлечение имён преподавателей (для индексов) -----------------------------
def is_teacher_name(text: str) -> bool:
return bool(_NAME_INITIALS_RE.search(text) or _FULL_NAME_RE.match(text.strip()))
def extract_teacher_names(teacher: str) -> list[str]:
"""Строка преподавателя -> список коротких имён «Фамилия И.О.» (ключи индекса).
«Смирнов А.В., Петров П.П.» -> оба; «Лабеев Игорь Владимирович» -> «Лабеев И.В.»;
«ПравоТех» -> как есть.
"""
if not teacher:
return []
names = [m.strip() for m in _NAME_INITIALS_RE.findall(teacher) if m.strip()]
if not names:
full = re.sub(r"\s*\([^()]*\)", "", teacher).strip()
full_m = _FULL_NAME_RE.match(full)
if full_m:
parts = full_m.group(0).split()
short = parts[0] + " " + parts[1][0].upper() + "."
if len(parts) > 2:
short += parts[2][0].upper() + "."
return [short]
if (
not re.search(r"\d|\s", teacher)
or (len(teacher) <= 30 and " " in teacher)
) and not re.search(r"[а-яё][А-ЯЁ]", teacher):
return [teacher] # организация одним словом или «НПО Криста»
return names

390
src/backend/schedule/parsing.py Normal file
View File

@@ -0,0 +1,390 @@
"""Разбор листов рабочей книги в структуру расписания.
Основной лист «Бакалавры и магистры» имеет сетку:
столбец 0-1 — даты недели числителей («с»/«по»),
столбец 2-3 — dates недели знаменателей,
столбец 4 — день недели (в первой строке блока дня),
столбец 5 — номер пары и время (объединён на обе строки пары),
столбец 6+ — столбцы групп (скрытые столбцы игнорируются).
Строки блока дня идут в хронологии семестра парами: первая строка пары —
неделя числителей, вторая — знаменателей (даты в левой колонке каждой пары).
Ячейки занятия разбирает lessons.parse_lesson_cell.
"""
from __future__ import annotations
import logging
import re
from datetime import date
from .errors import ScheduleError
from .lessons import extract_teacher_names, parse_lesson_cell
from .xlsx import RED_CLOSE, RED_OPEN, Worksheet, excel_serial_to_date
log = logging.getLogger(__name__)
MAIN_SHEET = "Бакалавры и магистры"
ROOMS_SHEET = "аудитории"
TEACHERS_SHEET = "Преподаватели"
DAYS = ["понедельник", "вторник", "среда", "четверг", "пятница", "суббота", "воскресенье"]
PARITIES = ("ч", "з")
_COL_DAY = 4 # день недели
_COL_PERIOD = 5 # пара и время
_FIRST_GROUP_COL = 6
_GROUP_RE = re.compile(r"^[А-ЯЁ]{2,4}-\d")
_PERIOD_RE = re.compile(r"^\s*(\d{1,2})\s*(.*)$")
_ROOM_RE = re.compile(r"ауд\.\s*([\w\-]+)", re.I)
_SEMESTER_DATE_RE = re.compile(r"(\d{1,2}\.\d{1,2}\.\d{2})")
_TIME_SPAN_RE = re.compile(r"\s*[-–—]\s*")
def _strip_marks(text: str) -> str:
return text.replace(RED_OPEN, "").replace(RED_CLOSE, "").strip()
def _normalize_header(text: str) -> str:
return _strip_marks(text).split("\n", 1)[0].strip()
def _semester_of(header_cell: str) -> str | None:
"""Дата окончания семестра из второй строки заголовка группы."""
parts = header_cell.split("\n", 1)
if len(parts) < 2:
return None
m = _SEMESTER_DATE_RE.search(_strip_marks(parts[1]))
return m.group(1) if m else None
def _parse_period_cell(text: str) -> tuple[int | None, str]:
"""'1|9:00-|10:35' / '3 13:20- 14:55' -> (номер пары, '13:20-14:55')."""
lines = [ln.strip() for ln in text.split("\n") if ln.strip()]
if not lines:
return None, ""
m = _PERIOD_RE.match(lines[0])
if not m:
return None, ""
parts = [m.group(2).strip(), *lines[1:]]
return int(m.group(1)), _TIME_SPAN_RE.sub("-", " ".join(p for p in parts if p))
# --- недели и чётность ---------------------------------------------------------
def extract_week_ranges(sheets: dict[str, Worksheet]) -> list[dict]:
"""Все циклы семестра: [{'parity', 'start', 'end'}...] в хронологии."""
sheet = sheets.get(MAIN_SHEET)
if sheet is None:
return []
ranges: list[dict] = []
seen: set[tuple[str, date]] = set()
for row in sheet.rows[1:]:
if not any(row[i] for i in range(4)):
continue
for parity, start_col, end_col in (("ч", 0, 1), ("з", 2, 3)):
start = excel_serial_to_date(row[start_col])
end = excel_serial_to_date(row[end_col])
if start is None or end is None:
continue
key = (parity, start)
if key not in seen:
seen.add(key)
ranges.append({"parity": parity, "start": start, "end": end})
return sorted(ranges, key=lambda w: w["start"])
def current_parity(week_ranges: list[dict], today: date | None = None) -> str | None:
today = today or date.today()
for week in week_ranges:
if week["start"] <= today <= week["end"]:
return week["parity"]
return None
# --- главный лист -------------------------------------------------------------
def _find_group_columns(header: list[str], hidden_cols: set[int]) -> dict[int, str]:
columns: dict[int, str] = {}
seen: set[str] = set()
for idx, cell in enumerate(header):
if idx in hidden_cols:
continue
name = _normalize_header(cell)
if name and _GROUP_RE.match(name) and name not in seen:
columns[idx] = name
seen.add(name)
return columns
def parse_schedule(sheets: dict[str, Worksheet]) -> dict:
"""Главный лист -> {группа: {'semester_until', 'days': {день: {пара: slot}}}}.
slot = {'time', 'ч': raw|None, 'з': raw|None, 'lessons': {'ч': [...], 'з': [...]}}
"""
sheet = sheets.get(MAIN_SHEET)
if sheet is None:
raise ScheduleError(f"Лист «{MAIN_SHEET}» не найден в таблице")
rows = sheet.rows
header = rows[0]
group_columns = _find_group_columns(header, sheet.hidden_cols)
groups = {
name: {"semester_until": _semester_of(header[col]), "days": {}}
for col, name in group_columns.items()
}
day_times: dict[str, dict[int, str]] = {}
cur_day: str | None = None
cur_period: int | None = None
cur_time = ""
parity: str | None = None
rows_in_day = 0
for row in rows[1:]:
if not any(row[i] for i in range(4)):
continue
day = _normalize_header(row[_COL_DAY]).lower()
if day in DAYS and day != cur_day:
if cur_day is not None and rows_in_day % 2:
log.warning(
"День «%s»: нечётное число строк (%d), недели могли сдвинуться",
cur_day,
rows_in_day,
)
cur_day = day
parity = None
rows_in_day = 0
if cur_day is None:
continue
rows_in_day += 1
period_cell = row[_COL_PERIOD]
if period_cell:
period, time = _parse_period_cell(period_cell)
if period is not None:
cur_period = period
cur_time = time
day_times.setdefault(cur_day, {}).setdefault(period, time)
if cur_period is None:
continue
parity = "з" if parity == "ч" else "ч"
for col, group_name in group_columns.items():
text = row[col]
if not text:
continue
days_map = groups[group_name]["days"].setdefault(cur_day, {})
slot = days_map.setdefault(
cur_period, {"time": cur_time, "ч": None, "з": None, "lessons": {}}
)
slot[parity] = text if slot[parity] is None else slot[parity]
# пары без занятий заполняем временем, как раньше
for group in groups.values():
for day, periods in day_times.items():
days_map = group["days"].setdefault(day, {})
for period, time in periods.items():
days_map.setdefault(
period, {"time": time, "ч": None, "з": None, "lessons": {}}
)
for group in groups.values():
for day, periods in group["days"].items():
for slot in periods.values():
for par in PARITIES:
raw = slot[par]
slot["lessons"][par] = (
[l.to_dict() for l in parse_lesson_cell(raw)] if raw else None
)
group["days"][day] = dict(sorted(periods.items()))
return groups
# --- представление для API -----------------------------------------------------
def build_group_schedule(data: dict, group_name: str) -> dict | None:
group = data.get(group_name)
if group is None:
return None
out = {"group": group_name, "semester_until": group["semester_until"], "days": {}}
for day, periods in group["days"].items():
out["days"][day] = {}
for period, slot in sorted(periods.items()):
entry = {"time": slot["time"]}
for par in PARITIES:
raw = slot[par]
entry[par] = (
None
if not raw
else {"raw": raw, "lessons": (slot["lessons"] or {}).get(par) or []}
)
out["days"][day][str(period)] = entry
return out
def iter_lessons(schedule_data: dict):
"""Все занятия: (group, day, period, time, parity, lesson_dict)."""
for group, gdata in schedule_data.items():
for day, periods in gdata["days"].items():
for period, slot in periods.items():
for par in PARITIES:
if not slot[par]:
continue
for lesson in slot["lessons"].get(par) or []:
yield group, day, period, slot["time"], par, lesson
def _merge_entry(slot_list: list, entry_key: dict, extra: dict) -> dict:
for item in slot_list:
if all(item.get(k) == v for k, v in entry_key.items()):
return item
item = dict(entry_key)
item["groups"] = []
item.update(extra)
slot_list.append(item)
return item
def build_teacher_room_index(schedule_data: dict) -> tuple[dict, dict]:
"""Расписание -> (индекс преподавателей, индекс аудиторий)."""
teachers: dict = {}
rooms: dict = {}
for group, day, period, time, parity, lesson in iter_lessons(schedule_data):
subject = lesson["subject"]
room = lesson["room"].strip()
names = extract_teacher_names(lesson["teacher"])
if names:
# в ячейке может быть несколько преподавателей — заносим в каждый
for name in names:
slot = teachers.setdefault(name, {}).setdefault(day, {}).setdefault(
str(period), {"time": time, "ч": [], "з": []}
)
item = _merge_entry(
slot[parity],
{"subject": subject, "room": room},
{"marker": lesson["marker"]},
)
if group not in item["groups"]:
item["groups"].append(group)
room_m = _ROOM_RE.match(room)
if room_m:
label = "ауд. " + room_m.group(1)
slot = rooms.setdefault(label, {}).setdefault(day, {}).setdefault(
str(period), {"time": time, "ч": [], "з": []}
)
display_teacher = ", ".join(names)
item = _merge_entry(
slot[parity],
{"subject": subject, "teacher": display_teacher},
{"marker": lesson["marker"]},
)
if group not in item["groups"]:
item["groups"].append(group)
return teachers, rooms
def sort_rooms(room_names) -> list:
def key(name: str):
m = re.search(r"(\d+)", name)
return (0, int(m.group(1))) if m else (1, name)
return sorted(room_names, key=key)
def build_day_times(schedule_data: dict) -> dict:
"""Время пар по дням: {'понедельник': {'1': '9:00-10:35', ...}}."""
day_times: dict = {}
for gdata in schedule_data.values():
for day, periods in gdata["days"].items():
for period, slot in periods.items():
day_times.setdefault(day, {}).setdefault(
str(period), slot.get("time", "")
)
return day_times
def build_teacher_schedule(teachers: dict, name: str) -> dict | None:
if name not in teachers:
return None
return {"name": name, "days": teachers[name]}
def build_room_schedule(rooms: dict, label: str, day_times: dict | None = None) -> dict | None:
if label not in rooms:
return None
data = rooms[label]
day_times = day_times or {}
out = {"room": label, "days": {}}
all_days = {*data.keys(), *day_times.keys()}
for day in sorted(all_days, key=lambda d: DAYS.index(d) if d in DAYS else 99):
times = day_times.get(day, {})
periods = {*data.get(day, {}).keys(), *times.keys()}
out["days"][day] = {}
for period in sorted(periods, key=int):
occupied = data.get(day, {}).get(period, {})
entry = {"time": occupied.get("time", times.get(period, ""))}
for par in PARITIES:
lessons = occupied.get(par) or []
entry[par] = lessons if lessons else None
out["days"][day][str(period)] = entry
return out
# --- вспомогательные листы ------------------------------------------------------
def parse_rooms_info(sheets: dict[str, Worksheet]) -> dict:
"""Лист «аудитории»: 'ауд. N' -> {'room', 'info'} (оснащение, вместимость)."""
sheet = sheets.get(ROOMS_SHEET)
if sheet is None or len(sheet.rows) < 2:
return {}
header, desc = sheet.rows[0], sheet.rows[1]
info: dict = {}
for col in range(_FIRST_GROUP_COL, max(len(header), len(desc))):
m = re.search(r"(\d+)", str(header[col]))
if not m:
continue
number = m.group(1)
text = " ".join(
ln.strip() for ln in str(desc[col]).split("\n") if ln.strip()
)
label = f"ауд. {number}"
info[label] = {"room": label, "info": _strip_marks(text)}
return info
def _initials_from_parts(first: str, patronymic: str) -> str:
initials = ""
if first:
initials += first[0].upper() + "."
if patronymic:
initials += patronymic[0].upper() + "."
return initials
def parse_teachers_full(sheets: dict[str, Worksheet]) -> dict:
"""Лист «Преподаватели»: 'Фамилия И.О.' -> полное ФИО (строка-заголовок)."""
sheet = sheets.get(TEACHERS_SHEET)
if sheet is None or len(sheet.rows) < 2:
return {}
header = sheet.rows[1]
result: dict = {}
for col in range(_FIRST_GROUP_COL, len(header)):
value = _strip_marks(str(header[col]))
parts = [p.strip() for p in re.split(r"[\s\n]+", value) if p.strip()]
if not parts:
continue
surname = parts[0]
first = parts[1] if len(parts) > 1 else ""
patronymic = parts[2] if len(parts) > 2 else ""
short = f"{surname} {_initials_from_parts(first, patronymic)}".strip()
full = " ".join(parts)
if short and full:
result.setdefault(short, full)
return result

45
src/backend/schedule/source.py Normal file
View File

@@ -0,0 +1,45 @@
"""Загрузка файла расписания по публичной ссылке Яндекс.Диска."""
from __future__ import annotations
import logging
import requests
from .errors import ScheduleError
log = logging.getLogger(__name__)
CLOUD_API = "https://cloud-api.yandex.net/v1/disk/public/resources"
def download_schedule(public_url: str, timeout: float = 60.0) -> bytes:
"""Скачивает файл с диска; при любой ошибке бросает ScheduleError."""
try:
resp = requests.get(
f"{CLOUD_API}/download",
params={"public_key": public_url},
timeout=timeout,
)
resp.raise_for_status()
href = resp.json()["href"]
file_resp = requests.get(href, timeout=timeout)
file_resp.raise_for_status()
return file_resp.content
except (requests.RequestException, KeyError, ValueError) as exc:
raise ScheduleError(f"Не удалось скачать таблицу: {exc}") from exc
def fetch_table_modified(public_url: str, timeout: float = 20.0) -> str | None:
"""Время последнего изменения файла на диске или None, если не удалось узнать."""
try:
resp = requests.get(
CLOUD_API,
params={"public_key": public_url},
timeout=timeout,
)
resp.raise_for_status()
return resp.json().get("modified")
except (requests.RequestException, ValueError) as exc:
log.debug("Не удалось получить modified таблицы: %s", exc)
return None

235
src/backend/schedule/xlsx.py Normal file
View File

@@ -0,0 +1,235 @@
"""Чтение xlsx-рабочей книги без сторонних библиотек.
Извлекает только то, что нужно для расписания:
- строки каждого листа как матрицу строк (объединённые ячейки раскатываются
до верхне-левого значения, пустые ячейки — ""),
- номера скрытых столбцов (в таблице ими помечены служебные данные),
- красные фрагменты текста, которые в таблице используются как примечания:
они оборачиваются служебными знаками RED_OPEN/RED_CLOSE.
"""
from __future__ import annotations
import io
import re
import zipfile
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from .errors import ScheduleError
XLSX_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
RED_OPEN = "\u27EA" # ⟪ — начало красного фрагмента (примечание)
RED_CLOSE = "\u27EB" # ⟫ — конец красного фрагмента
_REF_RE = re.compile(r"^([A-Z]+)(\d+)$")
@dataclass
class Worksheet:
rows: list[list[str]]
hidden_cols: set[int]
def cell(self, row_idx: int, col_idx: int) -> str:
"""Значение ячейки, "" вне границ матрицы."""
if row_idx < 0 or row_idx >= len(self.rows):
return ""
row = self.rows[row_idx]
if col_idx < 0 or col_idx >= len(row):
return ""
return row[col_idx]
def col_index(col_letter: str) -> int:
"""'A' -> 0, 'Z' -> 25, 'AA' -> 26."""
idx = 0
for ch in col_letter:
idx = idx * 26 + (ord(ch) - ord("A") + 1)
return idx - 1
def _parse_ref(ref: str) -> tuple[int, int]:
m = _REF_RE.match(ref)
if not m:
raise ScheduleError(f"Некорректная ссылка на ячейку: {ref!r}")
return col_index(m.group(1)), int(m.group(2)) - 1
def excel_serial_to_date(serial: object) -> date | None:
"""Серийная дата Excel (число или строка) в date; None, если не похоже на дату."""
try:
value = float(str(serial).replace(",", "."))
except (TypeError, ValueError):
return None
if value <= 0:
return None
# Excel считает 1.01.1900 = 1 и знает о несуществующем 29.02.1900 (серия 60).
if value <= 59:
value += 1
return (datetime(1899, 12, 30) + timedelta(days=value)).date()
def _is_red_rgb(rgb: str | None) -> bool:
"""Красный цвет ячейки вида FFRRGGBB."""
if not rgb or len(rgb) < 8 or not rgb.startswith("FF"):
return False
try:
r = int(rgb[2:4], 16)
g = int(rgb[4:6], 16)
b = int(rgb[6:8], 16)
except ValueError:
return False
return r > 0x90 and g < 0x70 and b < 0x70
def _parse_shared_strings(xml_bytes: bytes) -> list[str]:
"""sharedStrings.xml -> список значений.
Красные ранки оборачиваются RED_OPEN/RED_CLOSE, чтобы на этапе разбора
занятий примечания отделить от основного текста.
"""
root = ET.fromstring(xml_bytes)
out: list[str] = []
for si in root.iter(f"{{{XLSX_NS}}}si"):
parts: list[tuple[str, bool]] = []
for child in si:
if child.tag == f"{{{XLSX_NS}}}t":
parts.append((child.text or "", False))
continue
if child.tag != f"{{{XLSX_NS}}}r":
continue
t = child.find(f"{{{XLSX_NS}}}t")
red = False
rpr = child.find(f"{{{XLSX_NS}}}rPr")
if rpr is not None:
color = rpr.find(f"{{{XLSX_NS}}}color")
if color is not None:
red = _is_red_rgb(color.get("rgb"))
parts.append((t.text or "", red))
out.append(
"".join(
f"{RED_OPEN}{text}{RED_CLOSE}" if red else text for text, red in parts
)
)
return out
def _parse_sheet(xml_bytes: bytes, shared: list[str]) -> Worksheet:
root = ET.fromstring(xml_bytes)
cells: dict[tuple[int, int], str] = {}
max_row = 0
max_col = 0
hidden_cols: set[int] = set()
cols_node = root.find(f".//{{{XLSX_NS}}}cols")
if cols_node is not None:
for col in cols_node.iter(f"{{{XLSX_NS}}}col"):
try:
c_min = int(col.get("min"))
c_max = int(col.get("max"))
except (TypeError, ValueError):
continue
if col.get("hidden") == "1":
hidden_cols.update(range(c_min - 1, c_max))
for row in root.iter(f"{{{XLSX_NS}}}row"):
row_num = row.get("r")
if not row_num:
continue
r = int(row_num) - 1
for cell in row.iter(f"{{{XLSX_NS}}}c"):
ref = cell.get("r")
if not ref:
continue
c = col_index("".join(ch for ch in ref if ch.isalpha()))
t = cell.get("t")
v = cell.find(f"{{{XLSX_NS}}}v")
if t == "s" and v is not None and v.text is not None:
try:
idx = int(v.text)
except ValueError:
continue
value = shared[idx] if idx < len(shared) else ""
elif v is not None and v.text is not None:
value = v.text
else:
is_node = cell.find(f"{{{XLSX_NS}}}is")
value = "".join(_iter_text(is_node)) if is_node is not None else ""
if value != "":
cells[(r, c)] = value
max_row = max(max_row, r)
max_col = max(max_col, c)
merge_node = root.find(f".//{{{XLSX_NS}}}mergeCells")
if merge_node is not None:
for mc in merge_node.iter(f"{{{XLSX_NS}}}mergeCell"):
ref = mc.get("ref")
if not ref or ":" not in ref:
continue
a, b = ref.split(":")
c1, r1 = _parse_ref(a)
c2, r2 = _parse_ref(b)
max_row = max(max_row, r2)
max_col = max(max_col, c2)
top = cells.get((r1, c1), "")
if not top:
continue
for rr in range(r1, r2 + 1):
for cc in range(c1, c2 + 1):
if (rr, cc) not in cells:
cells[(rr, cc)] = top
rows = [
[cells.get((r, c), "") for c in range(max_col + 1)]
for r in range(max_row + 1)
]
return Worksheet(rows=rows, hidden_cols=hidden_cols)
def _iter_text(node: ET.Element):
for el in node.iter(f"{{{XLSX_NS}}}t"):
if el.text:
yield el.text
def parse_workbook(xlsx_bytes: bytes) -> dict[str, Worksheet]:
"""xlsx -> {имя листа: Worksheet}."""
try:
zf = zipfile.ZipFile(io.BytesIO(xlsx_bytes))
except zipfile.BadZipFile as exc:
raise ScheduleError("Скачанный файл не является xlsx") from exc
try:
shared = _parse_shared_strings(zf.read("xl/sharedStrings.xml"))
except KeyError:
shared = []
rels_root = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels"))
rel_map = {
rel.get("Id"): rel.get("Target")
for rel in rels_root
if rel.get("Id") and rel.get("Target")
}
sheets: dict[str, Worksheet] = {}
wb_root = ET.fromstring(zf.read("xl/workbook.xml"))
for sheet in wb_root.iter(f"{{{XLSX_NS}}}sheet"):
name = sheet.get("name")
rid = sheet.get(f"{{{REL_NS}}}id")
target = rel_map.get(rid or "")
if not (name and target):
continue
target = target.lstrip("/")
if target.startswith("xl/"):
target = target[3:]
if target.startswith("worksheets/"):
try:
sheets[name] = _parse_sheet(zf.read("xl/" + target), shared)
except ScheduleError:
raise
except ET.ParseError as exc:
raise ScheduleError(f"Не удалось разобрать лист «{name}»") from exc
return sheets

799
src/backend/schedule_parser.py
View File

@@ -1,799 +0,0 @@
import io
import re
import zipfile
import xml.etree.ElementTree as ET
from datetime import date, datetime, timedelta
import requests
XLSX_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
MAIN_SHEET = "Бакалавры и магистры"
DAYS = ["понедельник", "вторник", "среда", "четверг", "пятница", "суббота", "воскресенье"]
_PERIOD_RE = re.compile(r"^\s*(\d+)\s*(.*)$")
_REF_RE = re.compile(r"^([A-Z]+)(\d+)$")
class ScheduleError(Exception):
pass
def _iter_text(node):
for el in node.iter(f"{{{XLSX_NS}}}t"):
if el.text:
yield el.text
def _col_index(col_letter):
idx = 0
for ch in col_letter:
idx = idx * 26 + (ord(ch) - ord("A") + 1)
return idx - 1
def _parse_ref(ref):
m = _REF_RE.match(ref)
if not m:
raise ValueError(f"Некорректная ссылка на ячейку: {ref}")
return _col_index(m.group(1)), int(m.group(2)) - 1
RED_OPEN = "\u27EA"
RED_CLOSE = "\u27EB"
def _is_red_rgb(rgb):
try:
if not rgb or len(rgb) < 8 or not rgb.startswith("FF"):
return False
r = int(rgb[2:4], 16)
g = int(rgb[4:6], 16)
b = int(rgb[6:8], 16)
except (ValueError, TypeError):
return False
return r > 0x90 and g < 0x70 and b < 0x70
def _parse_shared_strings(xml_bytes):
root = ET.fromstring(xml_bytes)
out = []
for si in root.iter(f"{{{XLSX_NS}}}si"):
parts = []
for child in si:
if child.tag == f"{{{XLSX_NS}}}t":
parts.append((child.text or "", False))
continue
if child.tag != f"{{{XLSX_NS}}}r":
continue
t = child.find(f"{{{XLSX_NS}}}t")
text = t.text if t is not None else ""
red = False
rpr = child.find(f"{{{XLSX_NS}}}rPr")
if rpr is not None:
color = rpr.find(f"{{{XLSX_NS}}}color")
if color is not None:
red = _is_red_rgb(color.get("rgb"))
parts.append((text or "", red))
chunks = []
for text, red in parts:
if red:
chunks.append(RED_OPEN + text + RED_CLOSE)
else:
chunks.append(text)
out.append("".join(chunks))
return out
def _parse_sheet(xml_bytes, shared):
root = ET.fromstring(xml_bytes)
cells = {}
max_row = 0
max_col = 0
hidden_cols = set()
cols_node = root.find(f".//{{{XLSX_NS}}}cols")
if cols_node is not None:
for col in cols_node.iter(f"{{{XLSX_NS}}}col"):
if col.get("hidden") == "1":
try:
c_min = int(col.get("min"))
c_max = int(col.get("max"))
except (TypeError, ValueError):
continue
hidden_cols.update(range(c_min - 1, c_max))
for row in root.iter(f"{{{XLSX_NS}}}row"):
r = int(row.get("r")) - 1
for cell in row.iter(f"{{{XLSX_NS}}}c"):
ref = cell.get("r")
if not ref:
continue
c = _col_index("".join(ch for ch in ref if ch.isalpha()))
t = cell.get("t")
v = cell.find(f"{{{XLSX_NS}}}v")
if t == "s" and v is not None and v.text:
idx = int(v.text)
value = shared[idx] if idx < len(shared) else ""
elif v is not None and v.text is not None:
value = v.text
else:
is_node = cell.find(f"{{{XLSX_NS}}}is")
value = "".join(_iter_text(is_node)) if is_node is not None else ""
if value != "":
cells[(r, c)] = value
max_row = max(max_row, r)
max_col = max(max_col, c)
merge_node = root.find(f".//{{{XLSX_NS}}}mergeCells")
if merge_node is not None:
for mc in merge_node.iter(f"{{{XLSX_NS}}}mergeCell"):
ref = mc.get("ref")
if not ref or ":" not in ref:
continue
a, b = ref.split(":")
c1, r1 = _parse_ref(a)
c2, r2 = _parse_ref(b)
max_row = max(max_row, r2)
max_col = max(max_col, c2)
top = cells.get((r1, c1), "")
if not top:
continue
for rr in range(r1, r2 + 1):
for cc in range(c1, c2 + 1):
if (rr, cc) not in cells:
cells[(rr, cc)] = top
matrix = [
[cells.get((r, c), "") for c in range(max_col + 1)]
for r in range(max_row + 1)
]
return matrix, hidden_cols
def _read_cell(row, col_idx):
if col_idx < len(row):
return row[col_idx]
return ""
def download_schedule(public_url):
try:
resp = requests.get(
"https://cloud-api.yandex.net/v1/disk/public/resources/download",
params={"public_key": public_url},
timeout=30,
)
resp.raise_for_status()
href = resp.json()["href"]
file_resp = requests.get(href, timeout=60)
file_resp.raise_for_status()
return file_resp.content
except Exception as exc:
raise ScheduleError(f"Не удалось скачать таблицу: {exc}") from exc
def fetch_table_modified(public_url):
try:
resp = requests.get(
"https://cloud-api.yandex.net/v1/disk/public/resources",
params={"public_key": public_url},
timeout=30,
)
resp.raise_for_status()
return resp.json().get("modified")
except Exception:
return None
def parse_workbook(xlsx_bytes):
try:
zf = zipfile.ZipFile(io.BytesIO(xlsx_bytes))
except zipfile.BadZipFile as exc:
raise ScheduleError("Скачанный файл не является xlsx") from exc
shared = []
try:
shared = _parse_shared_strings(zf.read("xl/sharedStrings.xml"))
except KeyError:
pass
rels_root = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels"))
rel_map = {}
for rel in rels_root:
rid = rel.get("Id")
target = rel.get("Target")
if rid and target:
rel_map[rid] = target
wb_root = ET.fromstring(zf.read("xl/workbook.xml"))
sheets = {}
for sheet in wb_root.iter(f"{{{XLSX_NS}}}sheet"):
name = sheet.get("name")
rid = sheet.get(f"{{{REL_NS}}}id")
if name and rid and rid in rel_map:
target = rel_map[rid]
if target.startswith("worksheets/"):
matrix, hidden_cols = _parse_sheet(zf.read("xl/" + target), shared)
sheets[name] = {"rows": matrix, "hidden_cols": hidden_cols}
return sheets
def _normalize_header(text):
return text.split("\n", 1)[0].strip()
def _excel_serial_to_date(serial):
try:
value = float(str(serial).replace(",", "."))
except (TypeError, ValueError):
return None
if value <= 0:
return None
if value <= 60:
value -= 1
return (datetime(1899, 12, 30) + timedelta(days=value)).date()
def extract_week_ranges(sheets):
if MAIN_SHEET not in sheets:
return []
matrix = sheets[MAIN_SHEET]["rows"]
ranges = []
seen = set()
for row in matrix[1:]:
if not any(_read_cell(row, i) for i in range(4)):
continue
for parity, start_col, end_col in (("ч", 0, 1), ("з", 2, 3)):
start = _excel_serial_to_date(_read_cell(row, start_col))
end = _excel_serial_to_date(_read_cell(row, end_col))
if start and end:
key = (parity, start)
if key not in seen:
seen.add(key)
ranges.append({"parity": parity, "start": start, "end": end})
return sorted(ranges, key=lambda w: w["start"])
def current_parity(week_ranges, today=None):
today = today or date.today()
for week in week_ranges:
if week["start"] <= today <= week["end"]:
return week["parity"]
return None
def parse_schedule(sheets):
if MAIN_SHEET not in sheets:
raise ScheduleError(f"Лист «{MAIN_SHEET}» не найден в таблице")
sheet = sheets[MAIN_SHEET]
matrix = sheet["rows"]
hidden_cols = sheet["hidden_cols"]
header = matrix[0]
group_columns = {}
seen_names = set()
group_re = re.compile(r"^[А-ЯЁ]{2,4}-\d")
for idx, cell in enumerate(header):
if idx in hidden_cols:
continue
name = _normalize_header(cell)
if name and group_re.match(name) and name not in seen_names:
group_columns[idx] = name
seen_names.add(name)
groups = {}
for col, name in group_columns.items():
header_text = _read_cell(header, col)
semester = None
if "\n" in header_text:
second = header_text.split("\n", 1)[1].strip()
m = re.search(r"(\d{1,2}\.\d{2}\.\d{2})", second)
if m:
semester = m.group(1)
groups[name] = {"semester_until": semester, "days": {}}
day_periods = {}
cur_day = None
period = None
cur_time = ""
parity = None
for row in matrix[1:]:
has_dates = any(_read_cell(row, i) for i in range(4))
if not has_dates:
continue
day = _read_cell(row, 4)
if day:
normalized = day.strip().lower()
if normalized in DAYS:
if normalized != cur_day:
cur_day = normalized
parity = None
if cur_day is None:
continue
period_cell = _read_cell(row, 5)
if period_cell:
lines = [ln.strip() for ln in period_cell.split("\n") if ln.strip()]
m = _PERIOD_RE.match(lines[0])
if m:
period = int(m.group(1))
time_parts = [m.group(2).strip()]
time_parts += lines[1:]
cur_time = " ".join(p for p in time_parts if p)
cur_time = re.sub(r"\s*[-–—]\s*", "-", cur_time)
day_periods.setdefault(cur_day, {}).setdefault(period, cur_time)
if period is None:
continue
parity = "з" if parity == "ч" else "ч"
for col, group_name in group_columns.items():
cell_text = _read_cell(row, col)
if cell_text:
days_map = groups[group_name]["days"].setdefault(cur_day, {})
slot = days_map.setdefault(period, {"time": cur_time, "ч": None, "з": None})
slot[parity] = cell_text
for group_name in groups:
for day, periods in day_periods.items():
days_map = groups[group_name]["days"]
for period, time in periods.items():
days_map.setdefault(day, {})[period] = days_map.get(day, {}).get(
period, {"time": time, "ч": None, "з": None}
)
days_map[day] = dict(sorted(days_map[day].items()))
return groups
_MARKER_RE = re.compile(r"к\.в\.?\s*-?\s*\d*\s*", re.IGNORECASE)
_ROOM_LINE_RE = re.compile(r"ауд\.\s*[\w\-]+", re.IGNORECASE)
_SUBJECT_START_RE = re.compile(r"^[А-ЯЁA-Z]")
def _split_lesson_blocks(cell_text):
lines = cell_text.replace("\r\n", "\n").replace("\r", "\n").split("\n")
blocks = []
current = []
prev_blank = True
prev_room_line = False
for raw in lines:
line = raw.strip()
if not line:
prev_blank = True
continue
starts_marker = bool(_MARKER_RE.match(line))
room_line = bool(_ROOM_LINE_RE.search(line))
subject_start = bool(_SUBJECT_START_RE.match(line)) and not any(
w in line for w in ("не будет", "отменяется", "переносится")
)
if current and (prev_blank or starts_marker or (prev_room_line and subject_start)):
blocks.append(current)
current = []
current.append(line)
prev_blank = False
prev_room_line = room_line
if current:
blocks.append(current)
return blocks
def _lesson(marker, subject, teacher, room):
room = re.sub(r"ауд\.?\s*", "ауд. ", room, flags=re.IGNORECASE)
room = re.sub(r"\s{2,}", " ", room).strip()
return {
"marker": marker,
"subject": subject,
"teacher": teacher,
"room": room,
}
def _split_subject_teacher(head):
head = head.strip()
name_re = re.compile(
r"([А-ЯЁ][а-яё-]+(?:\s+[А-ЯЁ]\.[А-ЯЁ]\.?)+)(\s*\([^)]*\))*$"
)
name_m = name_re.search(head)
if name_m:
teacher = name_m.group(1).strip()
subject = (head[: name_m.start()] + (name_m.group(2) or "")).strip()
return subject, teacher
words = head.split()
if len(words) >= 2 and words[-1][:1].isupper():
return " ".join(words[:-1]), words[-1]
return head, ""
def _parse_inline(text, marker):
text = text.strip()
m = None
for match in re.finditer(r"ауд\.?\s*[\d\-]+\w*", text, re.IGNORECASE):
m = match
if not m:
return _lesson(marker, text, "", "")
room = m.group(0).strip()
head = text[: m.start()].strip()
subject, teacher = _split_subject_teacher(head)
return _lesson(marker, subject, teacher, room)
def _parse_lesson(lines):
lines = [
ln.replace(RED_OPEN, "").replace(RED_CLOSE, "").strip()
for ln in lines
if ln.strip()
]
if not lines:
return None
marker = None
m = re.match(r"^(к\.в\.?\s*-?\s*\d*)\s*(.*)$", lines[0], re.IGNORECASE)
if m:
marker = m.group(1).strip()
rest_head = m.group(2).strip()
if rest_head:
lines[0] = rest_head
else:
lines = lines[1:]
if not lines:
return None
if len(lines) == 1:
return _parse_inline(lines[0], marker)
if len(lines) == 2:
second = lines[1]
if re.search(r"(ауд\.|http|зал|каб\.|онлайн)", second, re.IGNORECASE):
subject, teacher = _split_subject_teacher(lines[0])
return _lesson(marker, subject, teacher, second)
if re.search(r"\b[А-ЯЁ]\.[А-ЯЁ]\.?$", second):
return _lesson(marker, lines[0], second, "")
return _lesson(marker, lines[0], "", second)
# >=3 строк: ищем строку аудитории, преподаватель — строка перед ней;
# строки после аудитории (примечания) не ломают разбор
room_idx = None
for i in range(len(lines) - 1, -1, -1):
if _ROOM_LINE_RE.match(lines[i]):
room_idx = i
break
if room_idx is None:
room = lines[-1]
teacher = lines[-2]
subject = " ".join(lines[:-2])
return _lesson(marker, subject, teacher, room)
room = lines[room_idx]
before = lines[:room_idx]
teacher = ""
teacher_idx = None
for i in range(len(before) - 1, -1, -1):
cand = before[i]
if _looks_like_teacher_line(cand):
teacher_idx = i
teacher = cand
break
if teacher_idx is not None:
subject = " ".join(before[:teacher_idx])
else:
subject = " ".join(before)
return _lesson(marker, subject, teacher, room)
def _unwrap_parens(text):
text = text.strip()
if text.startswith("(") and text.endswith(")"):
inner = text[1:-1].strip()
if inner:
return inner
return text
def _collect_notes(text):
notes = []
for m in re.findall(r"⟪([^⟫]*)⟫", text):
frag = m.strip()
if frag:
notes.append((_unwrap_parens(frag), True))
plain = re.sub(r"⟪[^⟫]*⟫", "", text)
for m in re.finditer(r"\(([^()]*)\)", plain):
inner = m.group(1).strip()
if not inner:
continue
if re.search(r"[А-ЯЁ]\.[А-ЯЁ]\.?", inner):
continue
if not re.search(r"[а-яёА-ЯЁa-zA-Z]", inner):
continue
notes.append((_unwrap_parens(inner), False))
return notes
def _strip_note_markers(value):
return value.replace(RED_OPEN, "").replace(RED_CLOSE, "")
def _split_lessons(cell_text):
lessons = []
last_marker = None
for block in _split_lesson_blocks(cell_text):
block_text = "\n".join(block)
lesson = _parse_lesson(block)
if lesson is None:
continue
for key in ("marker", "subject", "teacher", "room"):
if lesson[key]:
lesson[key] = _strip_note_markers(lesson[key])
fields = " ".join(
x for x in (lesson["subject"], lesson["teacher"], lesson["room"]) if x
)
keep = []
seen_notes = set()
for note_text, is_red in _collect_notes(block_text):
if not note_text or note_text in seen_notes:
continue
if not is_red and note_text in fields:
continue
seen_notes.add(note_text)
keep.append(note_text)
lesson["note"] = "\n".join(keep)
if lesson["marker"]:
last_marker = lesson["marker"]
elif last_marker:
lesson["marker"] = last_marker
lessons.append(lesson)
return lessons
def build_group_schedule(data, group_name):
group = data.get(group_name)
if group is None:
return None
out = {
"group": group_name,
"semester_until": group.get("semester_until"),
"days": {},
}
for day, periods in group["days"].items():
out["days"][day] = {}
for period, slot in periods.items():
entry = {"time": slot.get("time", "")}
for parity in ("ч", "з"):
raw = slot.get(parity)
if raw:
entry[parity] = {"raw": raw, "lessons": _split_lessons(raw)}
else:
entry[parity] = None
out["days"][day][str(period)] = entry
return out
_ROOM_RE = re.compile(r"ауд\.\s*([\w\-]+)", re.IGNORECASE)
def _iter_lessons(schedule_data):
for group, gdata in schedule_data.items():
for day, periods in gdata.get("days", {}).items():
for period, slot in periods.items():
time = slot.get("time", "")
for parity in ("ч", "з"):
raw = slot.get(parity)
if not raw:
continue
for lesson in _split_lessons(raw):
yield group, day, period, time, parity, lesson
def _is_name(text):
if not text:
return False
if text.startswith(("(", "—", "–", "-", "«", '"')):
return False
if any(w in text for w in ("не будет", "отменяется", "переносится", "пара будет")):
return False
if not re.search(r"[А-ЯЁа-яё]", text):
return False
if re.search(r"\d", text) and not re.search(r"[А-ЯЁ]\.[А-ЯЁ]", text):
return False
return True
_NAME_RE = re.compile(r"[А-ЯЁ][а-яё-]+(?:\s+[А-ЯЁ]\.[А-ЯЁ]\.?)+")
def _extract_teachers(text):
return [m.strip() for m in _NAME_RE.findall(text) if m.strip()]
def _is_org(text):
if not _is_name(text):
return False
if any(ch in text for ch in "()"):
return False
if re.search(r"[А-ЯЁ]\.[А-ЯЁ]", text):
return False
return True
_ORG_LINE_RE = re.compile(
r"[А-ЯЁ][A-Za-zА-ЯЁа-яё-]*(?:\s+[А-ЯЁ][A-Za-zА-ЯЁа-яё-]+)?"
)
def _looks_like_teacher_line(text):
if re.search(r"[А-ЯЁ]\.[А-ЯЁ]\.?", text):
return True
return bool(_ORG_LINE_RE.fullmatch(text)) and len(text) <= 40
def _merge_entry(slot_list, entry_key, fields):
for item in slot_list:
if all(item.get(k) == v for k, v in entry_key.items()):
return item
item = dict(entry_key)
item["groups"] = []
item.update(fields)
slot_list.append(item)
return item
def build_teacher_room_index(schedule_data):
teachers = {}
rooms = {}
for group, day, period, time, parity, lesson in _iter_lessons(schedule_data):
subject = lesson["subject"]
room = lesson["room"].strip()
if lesson["teacher"]:
teachers_found = _extract_teachers(lesson["teacher"])
if not teachers_found and _is_org(lesson["teacher"]):
teachers_found = [lesson["teacher"].strip()]
for teacher in teachers_found:
days_map = teachers.setdefault(teacher, {}).setdefault(day, {})
slot = days_map.setdefault(
str(period), {"time": time, "ч": [], "з": []}
)
item = _merge_entry(
slot[parity],
{"subject": subject, "room": room},
{"marker": lesson["marker"]},
)
if group not in item["groups"]:
item["groups"].append(group)
display_teacher = ", ".join(teachers_found) if teachers_found else ""
else:
display_teacher = ""
room_m = _ROOM_RE.match(room)
if room_m:
label = "ауд. " + room_m.group(1)
days_map = rooms.setdefault(label, {}).setdefault(day, {})
slot = days_map.setdefault(
str(period), {"time": time, "ч": [], "з": []}
)
item = _merge_entry(
slot[parity],
{"subject": subject, "teacher": display_teacher},
{"marker": lesson["marker"]},
)
if group not in item["groups"]:
item["groups"].append(group)
return teachers, rooms
def sort_rooms(room_names):
def key(name):
m = re.search(r"(\d+)", name)
return (0, int(m.group(1))) if m else (1, name)
return sorted(room_names, key=key)
def parse_rooms_info(sheets):
"""Аудитории: номер (заголовок) -> оснащение/вместимость (2-я строка)."""
name = "аудитории"
if name not in sheets:
return {}
rows = sheets[name]["rows"]
if len(rows) < 2:
return {}
header = rows[0]
desc = rows[1]
info = {}
for col in range(6, max(len(header), len(desc))):
raw = _read_cell(header, col)
m = re.search(r"(\d+)", str(raw))
if not m:
continue
number = int(m.group(1))
text = " ".join(
line.strip()
for line in _read_cell(desc, col).split("\n")
if line.strip()
)
label = f"ауд. {number}"
info[label] = {"room": label, "info": text}
return info
def _initials_from_parts(first, patronymic):
initials = ""
if first:
initials += first[0].upper() + "."
if patronymic:
initials += patronymic[0].upper() + "."
return initials
def parse_teachers_full(sheets):
"""Преподаватели: короткое «Фамилия И.О.» -> полное ФИО (3-я строка-заголовок)."""
name = "Преподаватели"
if name not in sheets:
return {}
rows = sheets[name]["rows"]
if len(rows) < 2:
return {}
header = rows[1]
result = {}
for col in range(6, len(header)):
value = _read_cell(header, col)
parts = [p.strip() for p in re.split(r"[\s\n]+", value) if p.strip()]
if not parts:
continue
surname = parts[0]
first = parts[1] if len(parts) > 1 else ""
patronymic = parts[2] if len(parts) > 2 else ""
short = f"{surname} {_initials_from_parts(first, patronymic)}".strip()
full = " ".join(parts)
if short and full:
result.setdefault(short, full)
return result
def build_teacher_schedule(teachers, name):
if name not in teachers:
return None
return {"name": name, "days": teachers[name]}
def build_day_times(schedule_data):
day_times = {}
for gdata in schedule_data.values():
for day, periods in gdata.get("days", {}).items():
target = day_times.setdefault(day, {})
for period, slot in periods.items():
target.setdefault(str(period), slot.get("time", ""))
return day_times
def build_room_schedule(rooms, label, day_times=None):
if label not in rooms:
return None
data = rooms[label]
out = {"room": label, "days": {}}
day_times = day_times or {}
all_days = {*data.keys(), *day_times.keys()}
for day in sorted(all_days, key=lambda d: DAYS.index(d) if d in DAYS else 99):
out["days"][day] = {}
times = day_times.get(day, {})
periods = {*data.get(day, {}).keys(), *times.keys()}
for period in sorted(periods, key=lambda p: int(p)):
entry = {"time": times.get(period, data.get(day, {}).get(period, {}).get("time", ""))}
occupied = data.get(day, {}).get(period, {})
for parity in ("ч", "з"):
lessons = occupied.get(parity) or []
entry[parity] = lessons if lessons else None
out["days"][day][str(period)] = entry
return out