"""
АНКОР · Живой класс — сервер.
Запуск:
pip install "fastapi>=0.110" "uvicorn[standard]"
export ANKOR_PASSWORD="ваш-пароль-преподавателя"
uvicorn server:app --host 0.0.0.0 --port 8000
Клиентский index.html положите в папку ./public рядом с этим файлом,
тогда интерфейс и API будут на одном домене и API_BASE можно оставить пустым.
Всё состояние хранится в памяти процесса: перезапуск стирает комнаты.
Для одного преподавателя и группы участников этого достаточно; если нужен
кластер из нескольких воркеров, запускайте строго один процесс (--workers 1),
иначе комнаты разойдутся по воркерам.
"""
from __future__ import annotations
import asyncio
import os
import re
import secrets
import time
import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import quote, unquote, urlsplit
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response
from fastapi.staticfiles import StaticFiles
from starlette.exceptions import HTTPException as StarletteHTTPException
# --------------------------------------------------------------------------
# Настройки
# --------------------------------------------------------------------------
TEACHER_PASSWORD = os.environ.get("ANKOR_PASSWORD", "")
# Домены интерфейса, если он размещён отдельно от сервера.
# Пример: ANKOR_ORIGINS="https://class.example.com,https://www.example.com"
ALLOWED_ORIGINS = [
origin.strip()
for origin in os.environ.get("ANKOR_ORIGINS", "").split(",")
if origin.strip()
]
ROOM_TTL = int(os.environ.get("ANKOR_ROOM_TTL", 12 * 3600)) # секунды
MAX_FILE_BYTES = 8 * 1024 * 1024
MAX_ROOM_FILE_BYTES = 64 * 1024 * 1024
MAX_TASKS = 60
MAX_MEMBERS = 60
MAX_OBJECTS_PER_BOARD = 400
MAX_POINTS = 1500
MAX_ANSWER_CHARS = 15000
CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # без похожих символов
CODE_LENGTH = 8 # помещается в maxlength="12" на клиенте
OBJECT_TYPES = {"stroke", "text", "card", "rect", "circle"}
TASK_KINDS = {"board", "image", "pdf", "text", "quiz", "sort", "link"}
BOARD_KINDS = {"board", "image", "sort"}
SHAREABLE_KINDS = BOARD_KINDS
ALLOWED_MIME = {
"image/png": "image",
"image/jpeg": "image",
"image/webp": "image",
"application/pdf": "pdf",
}
COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
# --------------------------------------------------------------------------
# Ошибки: клиент показывает поле "error", поэтому формат единый везде
# --------------------------------------------------------------------------
class ApiError(Exception):
def __init__(self, status: int, message: str) -> None:
self.status = status
self.message = message
class ConflictError(ApiError):
"""409: объект изменён кем-то другим. Клиент ждёт поле current."""
def __init__(self, message: str, current: dict[str, Any] | None) -> None:
super().__init__(409, message)
self.current = current
# --------------------------------------------------------------------------
# Модель данных
# --------------------------------------------------------------------------
@dataclass
class Member:
id: str
name: str
seen: float
activity: str = "В комнате"
@dataclass
class Task:
id: str
kind: str
title: str
prompt: str
mode: str
config: dict[str, Any]
@dataclass
class BoardObject:
id: str
version: int
deleted: bool
data: dict[str, Any]
op_id: str = ""
def payload(self) -> dict[str, Any]:
return {
"id": self.id,
"version": self.version,
"deleted": self.deleted,
"data": self.data,
}
@dataclass
class StoredFile:
id: str
mime: str
name: str
body: bytes
@dataclass
class Room:
code: str
title: str
created: float
touched: float
tasks: list[Task] = field(default_factory=list)
active_task: str = ""
members: dict[str, Member] = field(default_factory=dict)
# answers[task_id][member_id] = {"value": str, "done": bool}
answers: dict[str, dict[str, dict[str, Any]]] = field(default_factory=dict)
# boards[(task_id, scope)][object_id] = BoardObject
boards: dict[tuple[str, str], dict[str, BoardObject]] = field(default_factory=dict)
files: dict[str, StoredFile] = field(default_factory=dict)
file_bytes: int = 0
def task(self, task_id: str) -> Task | None:
for item in self.tasks:
if item.id == task_id:
return item
return None
def current_task(self) -> Task:
found = self.task(self.active_task)
return found or self.tasks[0]
@dataclass
class Session:
room: str
member_id: str # "teacher" для преподавателя
teacher: bool
rooms: dict[str, Room] = {}
sessions: dict[str, Session] = {}
# --------------------------------------------------------------------------
# Вспомогательное
# --------------------------------------------------------------------------
def now_ms() -> int:
return int(time.time() * 1000)
def new_code() -> str:
while True:
code = "".join(secrets.choice(CODE_ALPHABET) for _ in range(CODE_LENGTH))
if code not in rooms:
return code
def text_field(raw: Any, limit: int, label: str, required: bool = False) -> str:
if raw is None:
raw = ""
if not isinstance(raw, str):
raise ApiError(400, f"Поле «{label}» должно быть текстом.")
value = raw.strip()
if required and not value:
raise ApiError(400, f"Заполните поле «{label}».")
return value[:limit]
def lines_field(raw: Any, limit: int, item_limit: int, label: str) -> list[str]:
if not isinstance(raw, list):
raise ApiError(400, f"Поле «{label}» заполнено неверно.")
values = [
item.strip()[:item_limit]
for item in raw
if isinstance(item, str) and item.strip()
]
if not values:
raise ApiError(400, f"Заполните поле «{label}».")
return values[:limit]
def number(raw: Any, default: float, low: float, high: float) -> float:
if isinstance(raw, bool) or not isinstance(raw, (int, float)):
return default
value = float(raw)
if value != value or value in (float("inf"), float("-inf")):
return default
return max(low, min(high, value))
def clean_object_data(raw: Any) -> dict[str, Any]:
"""Приводим объект доски к безопасному виду, не доверяя клиенту."""
if not isinstance(raw, dict):
raise ApiError(400, "Объект доски передан неверно.")
kind = raw.get("type")
if kind not in OBJECT_TYPES:
raise ApiError(400, "Неизвестный тип объекта доски.")
color = raw.get("color")
if not isinstance(color, str) or not COLOR_RE.match(color):
color = "#18324b"
points: list[list[float]] = []
if kind == "stroke":
source = raw.get("points")
if not isinstance(source, list):
raise ApiError(400, "Линия передана неверно.")
for item in source[:MAX_POINTS]:
if isinstance(item, list) and len(item) == 2:
points.append(
[
number(item[0], 0, 0, 1200),
number(item[1], 0, 0, 750),
]
)
if not points:
raise ApiError(400, "Пустая линия.")
text = raw.get("text")
text = text[:500] if isinstance(text, str) else ""
return {
"type": kind,
"x": number(raw.get("x"), 0, 0, 1200),
"y": number(raw.get("y"), 0, 0, 750),
"w": number(raw.get("w"), 20, 1, 1200),
"h": number(raw.get("h"), 20, 1, 750),
"text": text,
"color": color,
"width": number(raw.get("width"), 3, 1, 12),
"points": points,
}
def external_url(raw: Any) -> str:
value = text_field(raw, 2000, "Ссылка", required=True)
parts = urlsplit(value)
if parts.scheme != "https" or not parts.netloc:
raise ApiError(400, "Нужна HTTPS-ссылка.")
return value
def resolve_scope(room: Room, task: Task, session: Session, requested: str) -> str:
"""
Общая доска — один ключ "shared" для всех.
Индивидуальная — участник видит только своё, преподаватель выбирает чьё.
"""
if task.mode == "shared" and task.kind in SHAREABLE_KINDS:
return "shared"
if not session.teacher:
return session.member_id
if requested == "teacher" or requested in room.members:
return requested
return "teacher"
def board_of(room: Room, task_id: str, scope: str) -> dict[str, BoardObject]:
return room.boards.setdefault((task_id, scope), {})
def authorize(request: Request) -> tuple[Room, Session]:
header = request.headers.get("authorization", "")
if not header.lower().startswith("bearer "):
raise ApiError(401, "Войдите в комнату заново.")
session = sessions.get(header[7:].strip())
if not session:
raise ApiError(401, "Сессия истекла. Войдите заново.")
room = rooms.get(session.room)
if not room:
raise ApiError(404, "Комната закрыта или устарела.")
room.touched = time.time()
return room, session
async def read_json(request: Request) -> dict[str, Any]:
try:
body = await request.json()
except Exception:
raise ApiError(400, "Неверный формат запроса.")
if not isinstance(body, dict):
raise ApiError(400, "Неверный формат запроса.")
return body
def check_password(body: dict[str, Any]) -> None:
if not TEACHER_PASSWORD:
raise ApiError(500, "На сервере не задан пароль преподавателя.")
supplied = body.get("password")
if not isinstance(supplied, str) or not secrets.compare_digest(
supplied, TEACHER_PASSWORD
):
raise ApiError(403, "Неверный пароль преподавателя.")
def issue_token(room_code: str, member_id: str, teacher: bool) -> str:
token = secrets.token_urlsafe(32)
sessions[token] = Session(room=room_code, member_id=member_id, teacher=teacher)
return token
def sweep() -> None:
deadline = time.time() - ROOM_TTL
stale = [code for code, room in rooms.items() if room.touched < deadline]
for code in stale:
rooms.pop(code, None)
if stale:
dead = set(stale)
for token in [t for t, s in sessions.items() if s.room in dead]:
sessions.pop(token, None)
# --------------------------------------------------------------------------
# Приложение
# --------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
async def loop() -> None:
while True:
await asyncio.sleep(300)
sweep()
task = asyncio.create_task(loop())
try:
yield
finally:
task.cancel()
app = FastAPI(title="ANKOR live class", lifespan=lifespan)
if ALLOWED_ORIGINS:
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS,
allow_methods=["GET", "POST", "OPTIONS"],
allow_headers=["Authorization", "Content-Type", "X-File-Name"],
max_age=600,
)
@app.exception_handler(ApiError)
async def api_error_handler(request: Request, error: ApiError) -> JSONResponse:
body: dict[str, Any] = {"error": error.message}
if isinstance(error, ConflictError):
body["current"] = error.current
return JSONResponse(body, status_code=error.status)
@app.exception_handler(StarletteHTTPException)
async def http_error_handler(
request: Request, error: StarletteHTTPException
) -> JSONResponse:
detail = error.detail if isinstance(error.detail, str) else "Ошибка запроса."
return JSONResponse({"error": detail}, status_code=error.status_code)
@app.exception_handler(RequestValidationError)
async def validation_error_handler(
request: Request, error: RequestValidationError
) -> JSONResponse:
return JSONResponse({"error": "Неверные данные запроса."}, status_code=400)
@app.exception_handler(Exception)
async def unexpected_error_handler(request: Request, error: Exception) -> JSONResponse:
return JSONResponse({"error": "Внутренняя ошибка сервера."}, status_code=500)
# --------------------------------------------------------------------------
# Вход в комнату
# --------------------------------------------------------------------------
@app.post("/api/create")
async def create(request: Request) -> dict[str, Any]:
body = await read_json(request)
check_password(body)
code = new_code()
moment = time.time()
first = Task(
id=uuid.uuid4().hex,
kind="board",
title="Общая доска",
prompt="Пишите и рисуйте вместе.",
mode="shared",
config={},
)
room = Room(
code=code,
title=text_field(body.get("title"), 120, "Название") or "Совместное занятие",
created=moment,
touched=moment,
tasks=[first],
active_task=first.id,
)
rooms[code] = room
return {"room": code, "token": issue_token(code, "teacher", True), "teacher": True}
@app.post("/api/resume")
async def resume(request: Request) -> dict[str, Any]:
body = await read_json(request)
check_password(body)
code = text_field(body.get("code"), 12, "Код комнаты", required=True).upper()
room = rooms.get(code)
if not room:
raise ApiError(404, "Комната не найдена. Создайте новую.")
room.touched = time.time()
return {"room": code, "token": issue_token(code, "teacher", True), "teacher": True}
@app.post("/api/join")
async def join(request: Request) -> dict[str, Any]:
body = await read_json(request)
code = text_field(body.get("code"), 12, "Код комнаты", required=True).upper()
name = text_field(body.get("name"), 50, "Имя", required=True)
room = rooms.get(code)
if not room:
raise ApiError(404, "Комната не найдена. Проверьте код.")
if len(room.members) >= MAX_MEMBERS:
raise ApiError(409, "В комнате уже максимум участников.")
member = Member(id=uuid.uuid4().hex, name=name, seen=time.time())
room.members[member.id] = member
room.touched = time.time()
return {"room": code, "token": issue_token(code, member.id, False), "teacher": False}
# --------------------------------------------------------------------------
# Снимок состояния
# --------------------------------------------------------------------------
@app.get("/api/state")
async def state(request: Request) -> dict[str, Any]:
room, session = authorize(request)
requested_scope = request.query_params.get("scope", "teacher")
activity = text_field(request.query_params.get("activity"), 60, "Активность")
member = room.members.get(session.member_id)
if member:
member.seen = time.time()
member.activity = activity or "В комнате"
elif not session.teacher:
raise ApiError(401, "Вы больше не в комнате. Присоединитесь заново.")
task = room.current_task()
scope = resolve_scope(room, task, session, requested_scope)
task_answers = room.answers.get(task.id, {})
if session.teacher:
answers = {
member_id: dict(answer) for member_id, answer in task_answers.items()
}
else:
own = task_answers.get(session.member_id)
answers = {session.member_id: dict(own)} if own else {}
snapshot: dict[str, Any] = {
"time": now_ms(),
"title": room.title,
"code": room.code,
"teacher": session.teacher,
"me": {
"id": session.member_id,
"name": member.name if member else "Преподаватель",
},
"scope": scope,
"task": {
"id": task.id,
"kind": task.kind,
"title": task.title,
"prompt": task.prompt,
"mode": task.mode,
"config": task.config,
},
"answers": answers,
"objects": [
item.payload() for item in board_of(room, task.id, scope).values()
],
}
if session.teacher:
snapshot["tasks"] = [
{"id": item.id, "kind": item.kind, "title": item.title}
for item in room.tasks
]
snapshot["members"] = [
{
"id": item.id,
"name": item.name,
"seen": int(item.seen * 1000),
"activity": item.activity,
}
for item in room.members.values()
]
else:
snapshot["tasks"] = []
snapshot["members"] = []
return snapshot
# --------------------------------------------------------------------------
# Действия
# --------------------------------------------------------------------------
@app.post("/api/action")
async def action(request: Request) -> dict[str, Any]:
room, session = authorize(request)
body = await read_json(request)
kind = body.get("action")
if kind in ("object.put", "object.delete"):
return object_action(room, session, body, delete=kind == "object.delete")
if kind == "answer":
return answer_action(room, session, body)
if kind == "task.activate":
return activate_action(room, session, body)
if kind == "task.add":
return add_task_action(room, session, body)
raise ApiError(400, "Неизвестное действие.")
def object_action(
room: Room, session: Session, body: dict[str, Any], delete: bool
) -> dict[str, Any]:
task_id = text_field(body.get("task"), 64, "Задание", required=True)
task = room.task(task_id)
if not task:
raise ApiError(404, "Задание больше не существует.")
if task.kind not in BOARD_KINDS:
raise ApiError(400, "В этом задании нет доски.")
scope = resolve_scope(room, task, session, str(body.get("scope") or "teacher"))
if not session.teacher and str(body.get("scope")) != scope:
raise ApiError(403, "Можно менять только свою доску.")
object_id = text_field(body.get("id"), 64, "Объект", required=True)
op_id = text_field(body.get("opId"), 64, "Операция")
expected = int(number(body.get("version"), 0, 0, 10**9))
board = board_of(room, task_id, scope)
existing = board.get(object_id)
# Повтор той же операции после обрыва связи — отвечаем успехом.
if existing and op_id and existing.op_id == op_id:
return {"version": existing.version}
current_version = existing.version if existing else 0
if expected != current_version:
raise ConflictError(
"Объект уже изменён другим участником.",
existing.payload() if existing else None,
)
if delete:
if not existing:
return {"version": 0}
existing.version += 1
existing.deleted = True
existing.op_id = op_id
room.touched = time.time()
return {"version": existing.version}
data = clean_object_data(body.get("data"))
if not existing:
alive = sum(1 for item in board.values() if not item.deleted)
if alive >= MAX_OBJECTS_PER_BOARD:
raise ApiError(409, "На доске слишком много объектов.")
existing = BoardObject(id=object_id, version=0, deleted=False, data=data)
board[object_id] = existing
existing.version += 1
existing.deleted = False
existing.data = data
existing.op_id = op_id
room.touched = time.time()
return {"version": existing.version}
def answer_action(
room: Room, session: Session, body: dict[str, Any]
) -> dict[str, Any]:
if session.teacher:
raise ApiError(403, "Преподаватель не отправляет ответы.")
task_id = text_field(body.get("task"), 64, "Задание", required=True)
task = room.task(task_id)
if not task:
raise ApiError(404, "Задание больше не существует.")
value = body.get("value")
value = value[:MAX_ANSWER_CHARS] if isinstance(value, str) else ""
room.answers.setdefault(task_id, {})[session.member_id] = {
"value": value,
"done": bool(body.get("done")),
"updated": now_ms(),
}
room.touched = time.time()
return {"ok": True}
def activate_action(
room: Room, session: Session, body: dict[str, Any]
) -> dict[str, Any]:
if not session.teacher:
raise ApiError(403, "Только преподаватель меняет задание.")
task_id = text_field(body.get("task"), 64, "Задание", required=True)
if not room.task(task_id):
raise ApiError(404, "Задание больше не существует.")
room.active_task = task_id
room.touched = time.time()
return {"ok": True}
def add_task_action(
room: Room, session: Session, body: dict[str, Any]
) -> dict[str, Any]:
if not session.teacher:
raise ApiError(403, "Только преподаватель добавляет задания.")
if len(room.tasks) >= MAX_TASKS:
raise ApiError(409, "Слишком много заданий в комнате.")
kind = body.get("kind")
if kind not in TASK_KINDS:
raise ApiError(400, "Неизвестный тип задания.")
raw_config = body.get("config")
if not isinstance(raw_config, dict):
raw_config = {}
config: dict[str, Any] = {}
if kind == "link":
config["url"] = external_url(raw_config.get("url"))
if kind == "quiz":
config["options"] = lines_field(
raw_config.get("options"), 12, 200, "Варианты"
)
if kind == "sort":
config["labels"] = lines_field(raw_config.get("labels"), 30, 120, "Карточки")
config["groups"] = lines_field(raw_config.get("groups"), 6, 60, "Группы")
if kind in ("image", "pdf"):
file_id = text_field(raw_config.get("fileId"), 64, "Файл", required=True)
if file_id not in room.files:
raise ApiError(404, "Файл не найден. Загрузите его заново.")
config["fileId"] = file_id
mode = body.get("mode")
if kind in SHAREABLE_KINDS and mode == "shared":
mode = "shared"
else:
mode = "individual"
task = Task(
id=uuid.uuid4().hex,
kind=kind,
title=text_field(body.get("title"), 150, "Название", required=True),
prompt=text_field(body.get("prompt"), 5000, "Инструкция"),
mode=mode,
config=config,
)
room.tasks.append(task)
room.active_task = task.id
room.touched = time.time()
return {"ok": True, "task": task.id}
# --------------------------------------------------------------------------
# Файлы
# --------------------------------------------------------------------------
@app.post("/api/upload")
async def upload(request: Request) -> dict[str, Any]:
room, session = authorize(request)
if not session.teacher:
raise ApiError(403, "Файлы загружает преподаватель.")
mime = (request.headers.get("content-type") or "").split(";")[0].strip().lower()
if mime not in ALLOWED_MIME:
raise ApiError(415, "Допустимы PNG, JPEG, WebP или PDF.")
declared = request.headers.get("content-length")
if declared and declared.isdigit() and int(declared) > MAX_FILE_BYTES:
raise ApiError(413, "Файл больше 8 МБ.")
chunks: list[bytes] = []
size = 0
async for chunk in request.stream():
size += len(chunk)
if size > MAX_FILE_BYTES:
raise ApiError(413, "Файл больше 8 МБ.")
chunks.append(chunk)
body = b"".join(chunks)
if not body:
raise ApiError(400, "Пустой файл.")
if room.file_bytes + size > MAX_ROOM_FILE_BYTES:
raise ApiError(413, "В комнате закончилось место для файлов.")
try:
name = unquote(request.headers.get("x-file-name", "") or "")[:120]
except Exception:
name = ""
stored = StoredFile(
id=uuid.uuid4().hex, mime=mime, name=name or "material", body=body
)
room.files[stored.id] = stored
room.file_bytes += size
room.touched = time.time()
return {"id": stored.id, "mime": stored.mime, "name": stored.name}
@app.get("/api/file")
async def file(request: Request) -> Response:
room, _session = authorize(request)
stored = room.files.get(request.query_params.get("id", ""))
if not stored:
raise ApiError(404, "Файл не найден.")
return Response(
content=stored.body,
media_type=stored.mime,
headers={
"Content-Disposition": (
"inline; filename*=UTF-8''" + quote(stored.name)
),
"Cache-Control": "private, max-age=600",
},
)
@app.get("/api/health")
async def health() -> dict[str, Any]:
return {"ok": True, "rooms": len(rooms), "time": now_ms()}
# --------------------------------------------------------------------------
# Интерфейс (public/index.html) — монтируется последним
# --------------------------------------------------------------------------
PUBLIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "public")
if os.path.isdir(PUBLIC_DIR):
app.mount("/", StaticFiles(directory=PUBLIC_DIR, html=True), name="public")
