""" Telearn Arcade Hub — FastAPI server. Serves the project catalog, static assets, and WebSocket bridge that streams headless Pygame/OpenCV student projects to the browser. """ from __future__ import annotations import argparse import asyncio import hashlib import json import logging import os import signal import subprocess import sys import time from typing import Any, Optional from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from classroom_service import router as classroom_router from admin_service import router as admin_router # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- BASE_DIR = os.path.dirname(os.path.abspath(__file__)) PROJECTS_DIR = os.path.join(BASE_DIR, "projects") TEMPLATES_DIR = os.path.join(BASE_DIR, "templates") STATIC_DIR = os.path.join(BASE_DIR, "static") MODELS_DIR = os.path.join(BASE_DIR, "models") MAX_CONCURRENT_RUNNERS = int(os.environ.get("MAX_CONCURRENT_RUNNERS", "3")) SESSION_TIMEOUT_SEC = int(os.environ.get("SESSION_TIMEOUT_SEC", "900")) # 15 min IDLE_TIMEOUT_SEC = max(1, int(os.environ.get("IDLE_TIMEOUT_SEC", "720"))) # 12 min IDLE_WARNING_SEC = min( IDLE_TIMEOUT_SEC, max(0, int(os.environ.get("IDLE_WARNING_SEC", "600"))) ) # warn after 10 min idle HEARTBEAT_TIMEOUT_SEC = max( 30, int(os.environ.get("HEARTBEAT_TIMEOUT_SEC", "75")) ) PROJECT_CACHE_TTL_SEC = float(os.environ.get("PROJECT_CACHE_TTL_SEC", "10")) RUNNER_CONNECT_TIMEOUT_SEC = float(os.environ.get("RUNNER_CONNECT_TIMEOUT_SEC", "8")) os.makedirs(PROJECTS_DIR, exist_ok=True) os.makedirs(TEMPLATES_DIR, exist_ok=True) os.makedirs(STATIC_DIR, exist_ok=True) os.makedirs(MODELS_DIR, exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s [server] %(message)s", datefmt="%Y-%m-%dT%H:%M:%S", ) log = logging.getLogger("telearn.server") app = FastAPI(title="Student Pygame/OpenCV Arcade Hub") templates = Jinja2Templates(directory=TEMPLATES_DIR) app.mount("/classroom-assets", StaticFiles(directory=os.path.join(BASE_DIR, "classroom_static")), name="classroom-assets") app.include_router(classroom_router) app.include_router(admin_router) @app.middleware("http") async def browser_runtime_headers(request: Request, call_next): """Enable SharedArrayBuffer for interactive input inside the Python worker.""" response = await call_next(request) response.headers["Cross-Origin-Opener-Policy"] = "same-origin" response.headers["Cross-Origin-Embedder-Policy"] = "credentialless" return response # Concurrency + live session metrics runner_semaphore = asyncio.Semaphore(MAX_CONCURRENT_RUNNERS) active_sessions: dict[str, dict[str, Any]] = {} _sessions_lock = asyncio.Lock() _metrics = { "sessions_started": 0, "sessions_rejected": 0, "sessions_timed_out": 0, "sessions_idle_timed_out": 0, "sessions_heartbeat_timed_out": 0, "sessions_failed_connect": 0, } # Project tree cache _project_cache: dict[str, Any] = {"tree": None, "built_at": 0.0, "signature": None} def _session_idle_state( idle_for_sec: float, warning_sec: float = IDLE_WARNING_SEC, timeout_sec: float = IDLE_TIMEOUT_SEC, ) -> str: """Return active, warning, or timeout for a session's idle duration.""" if idle_for_sec >= timeout_sec: return "timeout" if warning_sec < timeout_sec and idle_for_sec >= warning_sec: return "warning" return "active" # --------------------------------------------------------------------------- # Project scanning (cached) # --------------------------------------------------------------------------- def _dir_signature(root: str) -> str: """Cheap signature: path + mtime + name for dirs up to depth 3.""" parts: list[str] = [] if not os.path.isdir(root): return "" try: for class_name in sorted(os.listdir(root)): class_path = os.path.join(root, class_name) if not os.path.isdir(class_path) or class_name.startswith("."): continue try: parts.append(f"{class_name}:{os.path.getmtime(class_path)}") except OSError: parts.append(class_name) for student_name in sorted(os.listdir(class_path)): student_path = os.path.join(class_path, student_name) if not os.path.isdir(student_path) or student_name.startswith("."): continue try: parts.append(f"{class_name}/{student_name}:{os.path.getmtime(student_path)}") except OSError: parts.append(f"{class_name}/{student_name}") for project_name in sorted(os.listdir(student_path)): project_path = os.path.join(student_path, project_name) if not os.path.isdir(project_path) or project_name.startswith("."): continue try: parts.append( f"{class_name}/{student_name}/{project_name}:{os.path.getmtime(project_path)}" ) except OSError: parts.append(f"{class_name}/{student_name}/{project_name}") except OSError: return "" return "|".join(parts) def _scan_projects_uncached() -> dict: """ Scan projects/ into: { class_name: { student_name: [ {name, type, url}, ... ] } } Priority: python > html > apk > external """ tree: dict = {} if not os.path.exists(PROJECTS_DIR): return tree for class_name in sorted(os.listdir(PROJECTS_DIR)): class_path = os.path.join(PROJECTS_DIR, class_name) if not os.path.isdir(class_path) or class_name.startswith("."): continue tree[class_name] = {} for student_name in sorted(os.listdir(class_path)): student_path = os.path.join(class_path, student_name) if not os.path.isdir(student_path) or student_name.startswith("."): continue projects = [] for project_name in sorted(os.listdir(student_path)): project_dir = os.path.join(student_path, project_name) if not os.path.isdir(project_dir) or project_name.startswith("."): continue main_py = os.path.join(project_dir, "main.py") index_html = os.path.join(project_dir, "index.html") link_txt = os.path.join(project_dir, "link.txt") apk_files = [] try: apk_files = [ f for f in os.listdir(project_dir) if f.endswith(".apk") and not f.startswith(".") ] except OSError: pass if os.path.exists(main_py): projects.append({"name": project_name, "type": "python", "url": ""}) elif os.path.exists(index_html): projects.append({"name": project_name, "type": "html", "url": ""}) elif apk_files: projects.append({"name": project_name, "type": "apk", "url": apk_files[0]}) elif os.path.exists(link_txt): try: with open(link_txt, "r", encoding="utf-8") as f: url = f.read().strip() if url: projects.append({"name": project_name, "type": "external", "url": url}) except OSError: pass if projects: tree[class_name][student_name] = projects if not tree[class_name]: del tree[class_name] return tree def scan_projects(force: bool = False) -> dict: """Return cached project tree; refresh when TTL expires or directory signature changes.""" now = time.time() sig = _dir_signature(PROJECTS_DIR) cache_age = now - _project_cache["built_at"] if ( not force and _project_cache["tree"] is not None and cache_age < PROJECT_CACHE_TTL_SEC and _project_cache["signature"] == sig ): return _project_cache["tree"] tree = _scan_projects_uncached() _project_cache["tree"] = tree _project_cache["built_at"] = now _project_cache["signature"] = sig return tree def get_id(name: str) -> str: """Stable 10-char hex hash for URL tokens.""" return hashlib.sha256(name.encode("utf-8")).hexdigest()[:10] def decode_class_student( class_name_or_id: str, student_name_or_id: Optional[str] = None ) -> tuple[Optional[str], Optional[str]]: """Resolve hashed or raw class/student identifiers to folder names.""" all_projects = scan_projects() if not student_name_or_id: for c in all_projects: for s in all_projects[c]: if get_id(f"{c}/{s}") == class_name_or_id: return c, s if "/" in class_name_or_id: parts = class_name_or_id.split("/", 1) c, s = parts[0], parts[1] if c in all_projects and s in all_projects[c]: return c, s real_class = None for c in all_projects: if get_id(c) == class_name_or_id: real_class = c break if not real_class and class_name_or_id in all_projects: real_class = class_name_or_id return real_class, None real_class = None for c in all_projects: if get_id(c) == class_name_or_id: real_class = c break if not real_class and class_name_or_id in all_projects: real_class = class_name_or_id real_student = None if real_class and real_class in all_projects: for s in all_projects[real_class]: if get_id(f"{real_class}/{s}") == student_name_or_id: real_student = s break if not real_student and student_name_or_id in all_projects[real_class]: real_student = student_name_or_id return real_class, real_student def _safe_project_path(*parts: str) -> Optional[str]: """Join under PROJECTS_DIR; return None if path escapes the root.""" file_path = os.path.join(PROJECTS_DIR, *parts) abs_path = os.path.abspath(file_path) root = os.path.abspath(PROJECTS_DIR) if not abs_path.startswith(root + os.sep) and abs_path != root: return None return abs_path # --------------------------------------------------------------------------- # Static / logo # --------------------------------------------------------------------------- if os.path.isdir(STATIC_DIR): app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") @app.get("/images/logo-light.png") @app.get("/logo.png") async def get_logo(): return FileResponse(os.path.join(TEMPLATES_DIR, "logo-light.png")) # Controlled project file serving only (no raw StaticFiles mount on projects/) @app.get("/static_projects/{student_token}/{project_name}/{path:path}") async def serve_static_project_file(student_token: str, project_name: str, path: str): real_class, real_student = decode_class_student(student_token) if not real_class or not real_student: # Legacy: /static_projects/{class_id}/{student_id}/{project}/{path} parts = path.split("/", 1) if len(parts) == 2: actual_project_name, actual_path = parts[0], parts[1] real_class, real_student = decode_class_student(student_token, project_name) if real_class and real_student: file_path = _safe_project_path( real_class, real_student, actual_project_name, actual_path ) if file_path and os.path.isfile(file_path): return FileResponse(file_path) return HTMLResponse("Not Found", status_code=404) file_path = _safe_project_path(real_class, real_student, project_name, path) if not file_path or not os.path.isfile(file_path): return HTMLResponse("File Not Found", status_code=404) return FileResponse(file_path) # --------------------------------------------------------------------------- # HTTP routes # --------------------------------------------------------------------------- @app.get("/health") async def health(): return { "status": "ok", "active_sessions": len(active_sessions), "max_concurrent_runners": MAX_CONCURRENT_RUNNERS, "session_timeout_sec": SESSION_TIMEOUT_SEC, "idle_warning_sec": IDLE_WARNING_SEC, "idle_timeout_sec": IDLE_TIMEOUT_SEC, "heartbeat_timeout_sec": HEARTBEAT_TIMEOUT_SEC, } @app.get("/api/metrics") async def metrics(): return { "active_sessions": len(active_sessions), "max_concurrent_runners": MAX_CONCURRENT_RUNNERS, "sessions": list(active_sessions.values()), **_metrics, } @app.get("/", response_class=HTMLResponse) async def read_index(request: Request): return templates.TemplateResponse( "index.html", {"request": request, "restrict_class": "", "restrict_student": ""}, ) @app.get("/class/{class_name}", response_class=HTMLResponse) async def read_class(request: Request, class_name: str): real_class, _ = decode_class_student(class_name) if not real_class: return HTMLResponse("Class not found", status_code=404) return templates.TemplateResponse( "index.html", {"request": request, "restrict_class": real_class, "restrict_student": ""}, ) @app.get("/student/{student_token}", response_class=HTMLResponse) async def read_student(request: Request, student_token: str): real_class, real_student = decode_class_student(student_token) if not real_class or not real_student: return HTMLResponse("Student not found", status_code=404) return templates.TemplateResponse( "index.html", { "request": request, "restrict_class": real_class, "restrict_student": real_student, }, ) @app.get("/student/{class_name}/{student_name}", response_class=HTMLResponse) async def read_student_old(request: Request, class_name: str, student_name: str): real_class, real_student = decode_class_student(class_name, student_name) if not real_class or not real_student: return HTMLResponse("Student not found", status_code=404) student_token = get_id(f"{real_class}/{real_student}") return RedirectResponse(url=f"/student/{student_token}", status_code=307) @app.get("/api/projects", response_class=JSONResponse) async def get_projects(class_name: str = None, student_name: str = None): all_projects = scan_projects() class_ids = {c: get_id(c) for c in all_projects} student_ids = { f"{c}/{s}": get_id(f"{c}/{s}") for c in all_projects for s in all_projects[c] } if class_name: real_class, real_student = decode_class_student(class_name, student_name) if real_class in all_projects: if real_student: if real_student in all_projects[real_class]: filtered = {real_class: {real_student: all_projects[real_class][real_student]}} else: filtered = {real_class: {}} else: filtered = {real_class: all_projects[real_class]} else: filtered = {} else: filtered = all_projects return {"projects": filtered, "class_ids": class_ids, "student_ids": student_ids} @app.get("/run/{student_token}/{project_name}", response_class=HTMLResponse) async def run_project_page(request: Request, student_token: str, project_name: str): real_class, real_student = decode_class_student(student_token) if not real_class or not real_student: return HTMLResponse("Project not found", status_code=404) proj_dir = os.path.join(PROJECTS_DIR, real_class, real_student, project_name) proj_type = "unknown" external_url = "" apk_name = "" apk_size = "" appetize_key = "" view_only = request.query_params.get("view") == "1" apk_files = [] if os.path.isdir(proj_dir): try: apk_files = [ f for f in os.listdir(proj_dir) if f.endswith(".apk") and not f.startswith(".") ] except OSError: pass if os.path.exists(os.path.join(proj_dir, "main.py")): proj_type = "python" elif os.path.exists(os.path.join(proj_dir, "index.html")): proj_type = "html" elif apk_files: proj_type = "apk" apk_name = apk_files[0] try: size_bytes = os.path.getsize(os.path.join(proj_dir, apk_name)) if size_bytes < 1024 * 1024: apk_size = f"{size_bytes / 1024:.1f} KB" else: apk_size = f"{size_bytes / (1024 * 1024):.1f} MB" except OSError: pass appetize_path = os.path.join(proj_dir, "appetize.txt") if os.path.exists(appetize_path): try: with open(appetize_path, "r", encoding="utf-8") as f: appetize_key = f.read().strip() except OSError: pass elif os.path.exists(os.path.join(proj_dir, "link.txt")): proj_type = "external" try: with open(os.path.join(proj_dir, "link.txt"), "r", encoding="utf-8") as f: external_url = f.read().strip() except OSError: pass return templates.TemplateResponse( "player.html", { "request": request, "class_name": real_class, "student_name": real_student, "student_token": student_token, "project_name": project_name, "project_type": proj_type, "external_url": external_url, "apk_name": apk_name, "apk_size": apk_size, "appetize_key": appetize_key, "view_only": view_only, "session_timeout_sec": SESSION_TIMEOUT_SEC, "idle_warning_sec": IDLE_WARNING_SEC, "idle_timeout_sec": IDLE_TIMEOUT_SEC, "heartbeat_interval_sec": min(20, max(10, HEARTBEAT_TIMEOUT_SEC // 3)), "max_concurrent_runners": MAX_CONCURRENT_RUNNERS, }, ) @app.get("/run/{class_name}/{student_name}/{project_name}", response_class=HTMLResponse) async def run_project_page_old( request: Request, class_name: str, student_name: str, project_name: str ): real_class, real_student = decode_class_student(class_name, student_name) if not real_class or not real_student: return HTMLResponse("Project not found", status_code=404) student_token = get_id(f"{real_class}/{real_student}") qs = f"?{request.url.query}" if request.url.query else "" return RedirectResponse( url=f"/run/{student_token}/{project_name}{qs}", status_code=307 ) # --------------------------------------------------------------------------- # Process helpers # --------------------------------------------------------------------------- def _kill_process_group(process: asyncio.subprocess.Process) -> None: """Terminate the entire process group (runner + children).""" if process.returncode is not None: return pid = process.pid try: if sys.platform == "win32": process.terminate() else: try: os.killpg(pid, signal.SIGTERM) except (ProcessLookupError, PermissionError, OSError): process.terminate() except ProcessLookupError: pass async def _cleanup_process(process: asyncio.subprocess.Process) -> None: if process is None or process.returncode is not None: return log.info("Cleaning up process group pid=%s", process.pid) _kill_process_group(process) try: await asyncio.wait_for(process.wait(), timeout=2.0) except asyncio.TimeoutError: try: if sys.platform == "win32": process.kill() else: try: os.killpg(process.pid, signal.SIGKILL) except (ProcessLookupError, PermissionError, OSError): process.kill() except ProcessLookupError: pass try: await process.wait() except Exception: pass # --------------------------------------------------------------------------- # WebSocket play session # --------------------------------------------------------------------------- async def play_project_ws_internal( websocket: WebSocket, real_class: str, real_student: str, project_name: str ): if not real_class or not real_student: await websocket.accept() await websocket.send_json( {"type": "log", "data": "錯誤: 找不到該專案的執行路徑！\n"} ) await websocket.close() return project_path = os.path.join( PROJECTS_DIR, real_class, real_student, project_name, "main.py" ) if not os.path.exists(project_path): await websocket.accept() await websocket.send_json( {"type": "log", "data": "錯誤: 找不到該專案的 main.py 檔案！\n"} ) await websocket.close() return # Concurrency gate: reject immediately if all runner slots are busy acquired = False try: await asyncio.wait_for(runner_semaphore.acquire(), timeout=0.05) acquired = True except asyncio.TimeoutError: _metrics["sessions_rejected"] += 1 await websocket.accept() await websocket.send_json( { "type": "log", "data": ( f"目前展示名額已滿（最多同時 {MAX_CONCURRENT_RUNNERS} 個專案）。" "請稍後再試。\n" ), } ) await websocket.send_json({"type": "status", "data": "busy"}) await websocket.close(code=1013) return await websocket.accept() session_id = f"{real_class}/{real_student}/{project_name}/{time.time():.0f}" started_at = time.time() started_monotonic = time.monotonic() last_activity_at = started_monotonic last_client_seen_at = started_monotonic async with _sessions_lock: active_sessions[session_id] = { "class": real_class, "student": real_student, "project": project_name, "started_at": started_at, "last_activity_at": started_at, } _metrics["sessions_started"] += 1 process = None tcp_server = None runner_writer = None try: connected_event = asyncio.Event() runner_reader = None async def handle_runner_conn(reader, writer): nonlocal runner_reader, runner_writer runner_reader = reader runner_writer = writer connected_event.set() tcp_server = await asyncio.start_server(handle_runner_conn, "127.0.0.1", 0) tcp_port = tcp_server.sockets[0].getsockname()[1] runner_script = os.path.join(BASE_DIR, "runner.py") cmd = [ sys.executable, runner_script, "--project_path", project_path, "--port", str(tcp_port), "--models_dir", MODELS_DIR, ] popen_kwargs: dict[str, Any] = { "stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL, } if sys.platform != "win32": popen_kwargs["start_new_session"] = True process = await asyncio.create_subprocess_exec(*cmd, **popen_kwargs) log.info( "Started runner pid=%s session=%s port=%s", process.pid, session_id, tcp_port, ) try: await asyncio.wait_for( connected_event.wait(), timeout=RUNNER_CONNECT_TIMEOUT_SEC ) log.info("Runner pid=%s connected via TCP", process.pid) await websocket.send_json( {"type": "status", "data": "running"} ) await websocket.send_json( { "type": "log", "data": f"[系統] 專案執行器已就緒（最長 {SESSION_TIMEOUT_SEC // 60} 分鐘）。\n", } ) except asyncio.TimeoutError: _metrics["sessions_failed_connect"] += 1 log.error("Runner pid=%s connect timeout", process.pid) await websocket.send_json( {"type": "log", "data": "伺服器錯誤: 執行模擬器連線超時！\n"} ) await websocket.send_json({"type": "status", "data": "error"}) return async def tcp_to_ws(): try: while True: type_bytes = await runner_reader.readexactly(4) msg_type = type_bytes.decode("utf-8") len_bytes = await runner_reader.readexactly(4) length = int.from_bytes(len_bytes, byteorder="big") payload = await runner_reader.readexactly(length) if msg_type == "FRME": await websocket.send_bytes(payload) elif msg_type == "LOGS": log_text = payload.decode("utf-8", errors="ignore") await websocket.send_json({"type": "log", "data": log_text}) elif msg_type == "EXIT": log.info("Runner pid=%s sent EXIT", process.pid) await websocket.send_json({"type": "status", "data": "stopped"}) break except asyncio.IncompleteReadError: log.info("Runner pid=%s TCP incomplete read", process.pid) except WebSocketDisconnect: pass except Exception as e: log.exception("tcp_to_ws error: %s", e) async def ws_to_tcp(): nonlocal last_activity_at, last_client_seen_at try: while True: data = await websocket.receive() now = time.monotonic() last_client_seen_at = now if "text" in data: message = json.loads(data["text"]) mtype = message.get("type") if mtype == "heartbeat": continue if mtype == "activity": last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() elif mtype in ("keydown", "keyup"): last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() payload = json.dumps(message).encode("utf-8") runner_writer.write( b"KEYS" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype in ("mousemove", "mousedown", "mouseup"): last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() payload = json.dumps(message).encode("utf-8") runner_writer.write( b"MOUS" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype == "cv2key": last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() payload = json.dumps( {"key_code": message.get("keyCode")} ).encode("utf-8") runner_writer.write( b"CV2K" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype == "terminal_input": last_activity_at = now active_sessions[session_id]["last_activity_at"] = time.time() text = message.get("data", "") if not text.endswith("\n"): text += "\n" payload = text.encode("utf-8") runner_writer.write( b"INPT" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() elif mtype == "pygame_quit": last_activity_at = now runner_writer.write(b"QUIT" + (0).to_bytes(4, "big") + b"") await runner_writer.drain() elif "bytes" in data: payload = data["bytes"] runner_writer.write( b"CAMF" + len(payload).to_bytes(4, "big") + payload ) await runner_writer.drain() except WebSocketDisconnect: log.info("Browser WebSocket disconnected session=%s", session_id) except Exception as e: log.exception("ws_to_tcp error: %s", e) async def session_watchdog(): warning_sent = False while True: await asyncio.sleep(1) now = time.monotonic() idle_for = now - last_activity_at session_age = now - started_monotonic heartbeat_age = now - last_client_seen_at idle_state = _session_idle_state(idle_for) if session_age >= SESSION_TIMEOUT_SEC: _metrics["sessions_timed_out"] += 1 log.warning("Session hard timeout session=%s", session_id) await websocket.send_json( { "type": "log", "data": f"\n[系統] 已達最長執行時間（{SESSION_TIMEOUT_SEC // 60} 分鐘），自動結束。\n", } ) await websocket.send_json({"type": "status", "data": "timeout"}) break if heartbeat_age >= HEARTBEAT_TIMEOUT_SEC: _metrics["sessions_heartbeat_timed_out"] += 1 log.warning("Session heartbeat timeout session=%s", session_id) await websocket.send_json( { "type": "log", "data": "\n[系統] 瀏覽器已離線，自動釋放展示名額。\n", } ) await websocket.send_json( {"type": "status", "data": "connection_timeout"} ) break if idle_state == "timeout": _metrics["sessions_idle_timed_out"] += 1 log.info("Session idle timeout session=%s", session_id) await websocket.send_json( { "type": "log", "data": f"\n[系統] 已閒置 {IDLE_TIMEOUT_SEC // 60} 分鐘，自動結束並釋放展示名額。\n", } ) await websocket.send_json( {"type": "status", "data": "idle_timeout"} ) break if idle_state == "warning" and not warning_sent: warning_sent = True await websocket.send_json( { "type": "idle_warning", "remaining_sec": max( 0, int(IDLE_TIMEOUT_SEC - idle_for) ), } ) elif idle_state == "active" and warning_sent: warning_sent = False await websocket.send_json({"type": "idle_resumed"}) try: await websocket.close(code=1000) except Exception: pass tcp_task = asyncio.create_task(tcp_to_ws()) ws_task = asyncio.create_task(ws_to_tcp()) watchdog = asyncio.create_task(session_watchdog()) try: _, pending = await asyncio.wait( {tcp_task, ws_task, watchdog}, return_when=asyncio.FIRST_COMPLETED, ) for task in pending: task.cancel() await asyncio.gather(*pending, return_exceptions=True) finally: for task in (tcp_task, ws_task, watchdog): if not task.done(): task.cancel() await asyncio.gather( tcp_task, ws_task, watchdog, return_exceptions=True ) finally: if process is not None: await _cleanup_process(process) if runner_writer is not None: try: runner_writer.close() await runner_writer.wait_closed() except Exception: pass if tcp_server is not None: tcp_server.close() await tcp_server.wait_closed() try: await websocket.close() except Exception: pass async with _sessions_lock: active_sessions.pop(session_id, None) if acquired: runner_semaphore.release() log.info( "Session ended session=%s duration=%.1fs", session_id, time.time() - started_at, ) @app.websocket("/ws/play/{student_token}/{project_name}") async def play_project_ws(websocket: WebSocket, student_token: str, project_name: str): real_class, real_student = decode_class_student(student_token) await play_project_ws_internal(websocket, real_class, real_student, project_name) @app.websocket("/ws/play/{class_name}/{student_name}/{project_name}") async def play_project_ws_old( websocket: WebSocket, class_name: str, student_name: str, project_name: str ): real_class, real_student = decode_class_student(class_name, student_name) await play_project_ws_internal(websocket, real_class, real_student, project_name) # --------------------------------------------------------------------------- # Entrypoint # --------------------------------------------------------------------------- if __name__ == "__main__": import uvicorn parser = argparse.ArgumentParser(description="Telearn Arcade Hub") parser.add_argument("--host", default="127.0.0.1", help="Server host") parser.add_argument("--port", type=int, default=8000, help="Server port") parser.add_argument( "--reload", action="store_true", default=os.environ.get("DEV_RELOAD", "").lower() in ("1", "true", "yes"), help="Enable auto-reload (dev only)", ) args = parser.parse_args() uvicorn.run( "server:app", host=args.host, port=args.port, reload=args.reload, )