diff --git a/README.md b/README.md
index e3938d0..88153fc 100644
--- a/README.md
+++ b/README.md
@@ -39,6 +39,7 @@ The portal modules are referred to as applets because SPH is built on top of Moo
- `stundenplan` — timetable data
- `lerngruppen` — study groups
- `oberstufenwahl` — student-facing elections and course choices
+- `videokonferenz` — current student video rooms and safe join links
- `school_list` — school names and IDs for login and school selection
@@ -145,6 +146,7 @@ keep navigation responsive.
- **Session TTL:** 1 hour inactivity timeout per session
- **Response cache:** 10 minutes for most endpoints
+- **Always fresh:** `/videokonferenz` is never response-cached because room links may be short-lived and user-bound
- **Long cache (30 days):** `/modules`, `/apps`, `/benutzer`
- **School list cache:** 2 days with 3-day auto-refresh
- **File cache:** SHA-256 hashed, stored in `data/files/`
diff --git a/api-documentation.md b/api-documentation.md
index 187369c..896c9ed 100644
--- a/api-documentation.md
+++ b/api-documentation.md
@@ -36,6 +36,20 @@ X-Session-Token: {token}
## Endpoints
+### Video rooms (Videokonferenz)
+
+#### GET `/videokonferenz`
+
+Returns the authenticated student's current native Schulportal video rooms.
+Each room includes its learning group, teachers, `open`, `waiting`, `closed`, or
+`unknown` state, and an HTTPS join link when the portal provides one. Responses
+are intentionally never cached because join links may be short-lived and bound
+to the current user. Cookies, raw HTML, and Schulportal session tokens are not
+included in the response.
+
+The optional `refresh` query parameter is accepted for consistency with other
+module endpoints; video rooms are fetched fresh on every request regardless.
+
### Authentication
#### POST `/login`
diff --git a/api/api.py b/api/api.py
index 067378d..cd939f7 100644
--- a/api/api.py
+++ b/api/api.py
@@ -339,6 +339,7 @@ def validate_module_preferences(cls, value):
"search",
"dashboard",
"messages",
+ "videokonferenz",
"dateispeicher",
"vertretungsplan",
"dsb",
@@ -1368,6 +1369,19 @@ async def get_modules(
return result
+@app.get("/videokonferenz")
+async def get_videokonferenz(
+ response: Response,
+ refresh: bool = False,
+ auth: AuthSession = Depends(client_dependency),
+) -> Dict[str, object]:
+ """Return fresh rooms; user-bound join URLs must never enter response caches."""
+ # ``refresh`` remains accepted so clients can share a refresh-shaped API
+ # with other modules, but this endpoint intentionally always contacts SPH.
+ response.headers["Cache-Control"] = "private, no-store"
+ return await run_in_threadpool(auth.client.videokonferenz_get_rooms)
+
+
# --- Wahlen / Oberstufenwahl ---
diff --git a/api/auth_db.py b/api/auth_db.py
index dfed14c..7e3b9f6 100644
--- a/api/auth_db.py
+++ b/api/auth_db.py
@@ -95,6 +95,7 @@ def _pairing_code_hash(code: str) -> str:
"search",
"dashboard",
"messages",
+ "videokonferenz",
"dateispeicher",
"vertretungsplan",
"dsb",
diff --git a/schulportal_hessen/applets/videokonferenz/__init__.py b/schulportal_hessen/applets/videokonferenz/__init__.py
new file mode 100644
index 0000000..66370e2
--- /dev/null
+++ b/schulportal_hessen/applets/videokonferenz/__init__.py
@@ -0,0 +1,5 @@
+"""Native Videokonferenz module support."""
+
+from .api import parse_videokonferenz_overview, videokonferenz_get_rooms
+
+__all__ = ["parse_videokonferenz_overview", "videokonferenz_get_rooms"]
diff --git a/schulportal_hessen/applets/videokonferenz/api.py b/schulportal_hessen/applets/videokonferenz/api.py
new file mode 100644
index 0000000..f158341
--- /dev/null
+++ b/schulportal_hessen/applets/videokonferenz/api.py
@@ -0,0 +1,311 @@
+"""Read the student's native Schulportal video-room overview.
+
+The upstream module is a server-rendered page whose presentation has changed
+over time. Parsing is deliberately based on the German column/action labels,
+with class names used only as a fallback.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from typing import Any
+from urllib.parse import parse_qs, urljoin, urlsplit
+
+import requests
+from bs4 import BeautifulSoup
+
+_ROOM_ID_KEYS = ("room", "raum", "id", "kurs", "course", "lerngruppe", "group", "lg")
+_UPDATED_RE = re.compile(
+ r"(?:Stand|Ansicht\s+aktuell(?:\s+seit)?|aktualisiert(?:\s+am)?)\s*:?\s*"
+ r"((?:\d{1,2}\.\d{1,2}\.\d{2,4}(?:\s+um)?\s+)?"
+ r"\d{1,2}:\d{2}(?::\d{2})?\s*Uhr)",
+ re.IGNORECASE,
+)
+
+
+def _text(node: Any) -> str:
+ return " ".join(node.get_text(" ", strip=True).split()) if node else ""
+
+
+def _safe_url(value: str, source_url: str) -> str | None:
+ if not value or value.startswith(("#", "javascript:", "data:")):
+ return None
+ absolute = urljoin(source_url, value)
+ parsed = urlsplit(absolute)
+ if parsed.scheme != "https" or not parsed.netloc:
+ return None
+ return absolute
+
+
+def _room_id(join_url: str | None, name: str, position: int, node: Any) -> str:
+ for attribute in ("data-room-id", "data-id", "data-room", "data-kurs"):
+ value = str(node.get(attribute) or "").strip()
+ if value:
+ return value
+ if join_url:
+ query = parse_qs(urlsplit(join_url).query)
+ for key in _ROOM_ID_KEYS:
+ values = query.get(key)
+ if values and values[0].strip():
+ return values[0].strip()
+ seed = f"{name}|{join_url or ''}|{position}".encode()
+ return hashlib.sha256(seed).hexdigest()[:16]
+
+
+def _status(action_text: str, action_node: Any, join_url: str | None) -> str:
+ value = action_text.casefold()
+ classes = " ".join(action_node.get("class", [])).casefold() if action_node else ""
+ if "raum betreten" in value or "beitreten" in value or "btn-success" in classes:
+ return "open"
+ if (
+ "raum nicht offen" in value
+ or "noch nicht" in value
+ or "warten" in value
+ or "btn-danger" in classes
+ or "btn-warning" in classes
+ ):
+ return "waiting"
+ if not join_url or "geschlossen" in value or "beendet" in value:
+ return "closed"
+ return "unknown"
+
+
+def _teachers(cell: Any) -> list[str]:
+ if not cell:
+ return []
+ labelled = [_text(node) for node in cell.select(".label, .badge, li")]
+ values = [value for value in labelled if value]
+ if not values:
+ values = [part.strip() for part in re.split(r"[,;/|\n]+", cell.get_text("\n", strip=True))]
+ return list(dict.fromkeys(value for value in values if value))
+
+
+def _links(cell: Any, source_url: str, action_link: Any = None) -> list[dict[str, str]]:
+ result: list[dict[str, str]] = []
+ if not cell:
+ return result
+ for link in cell.select("a[href]"):
+ if action_link is not None and link is action_link:
+ continue
+ href = _safe_url(str(link.get("href") or ""), source_url)
+ if not href:
+ continue
+ result.append({"label": _text(link) or "Link öffnen", "url": href})
+ return result
+
+
+def _parse_table(table: Any, source_url: str) -> list[dict[str, Any]]:
+ header_nodes = table.select("thead th") or table.select("tr:first-child th")
+ headers = [_text(cell).casefold() for cell in header_nodes]
+
+ def column(*names: str) -> int | None:
+ for index, header in enumerate(headers):
+ if any(name in header for name in names):
+ return index
+ return None
+
+ group_index = column("lerngruppe", "kurs", "raum")
+ teacher_index = column("lehrkräft", "lehrer")
+ action_index = column("aktion", "status")
+ links_index = column("links", "material")
+ # The production page intentionally leaves the first header blank even
+ # though that column contains the Lerngruppe name.
+ if group_index is None and action_index is not None and headers:
+ group_index = 0
+ if group_index is None or action_index is None:
+ return []
+
+ rooms: list[dict[str, Any]] = []
+ rows = table.select("tbody tr")
+ if not rows:
+ rows = [row for row in table.select("tr") if not row.find("th")]
+ for position, row in enumerate(rows):
+ cells = row.find_all(["td", "th"], recursive=False)
+ if max(group_index, action_index) >= len(cells):
+ continue
+ name = _text(cells[group_index])
+ if not name:
+ continue
+ action_cell = cells[action_index]
+ action_nodes = action_cell.select("a[href], [data-href], [data-url], button")
+ action_node = next(
+ (
+ node
+ for node in action_nodes
+ if "hidden" not in (node.get("class") or [])
+ and not node.has_attr("hidden")
+ and "display:none" not in str(node.get("style") or "").replace(" ", "").casefold()
+ ),
+ action_nodes[0] if action_nodes else None,
+ )
+ action_target = ""
+ if action_node:
+ action_target = str(
+ action_node.get("href")
+ or action_node.get("data-href")
+ or action_node.get("data-url")
+ or ""
+ )
+ join_url = _safe_url(action_target, source_url)
+ action_text = _text(action_node) or _text(action_cell)
+ state = _status(action_text, action_node, join_url)
+ rooms.append(
+ {
+ "id": _room_id(join_url, name, position, row),
+ "name": name,
+ "teachers": _teachers(cells[teacher_index]) if teacher_index is not None and teacher_index < len(cells) else [],
+ "status": state,
+ "status_label": action_text or {
+ "open": "Raum betreten",
+ "waiting": "Raum nicht offen",
+ "closed": "Raum geschlossen",
+ }.get(state, "Status unbekannt"),
+ "join_url": join_url,
+ "can_join": bool(join_url and state == "open"),
+ "links": _links(cells[links_index], source_url) if links_index is not None and links_index < len(cells) else [],
+ }
+ )
+ return rooms
+
+
+def _parse_cards(soup: BeautifulSoup, source_url: str) -> list[dict[str, Any]]:
+ rooms: list[dict[str, Any]] = []
+ selectors = "[data-room-id], [data-room], .videokonferenz-room, .conference-room"
+ for position, card in enumerate(soup.select(selectors)):
+ name_node = card.select_one(".caption, .room-name, .panel-title, h2, h3, h4")
+ name = _text(name_node)
+ action_node = card.select_one(
+ "a[href].btn, a[href][role='button'], [data-href], [data-url]"
+ )
+ if not name or not action_node:
+ continue
+ join_url = _safe_url(
+ str(
+ action_node.get("href")
+ or action_node.get("data-href")
+ or action_node.get("data-url")
+ or ""
+ ),
+ source_url,
+ )
+ action_text = _text(action_node)
+ teacher_node = card.select_one(".teachers, .teacher, [data-role='teachers']")
+ rooms.append(
+ {
+ "id": _room_id(join_url, name, position, card),
+ "name": name,
+ "teachers": _teachers(teacher_node),
+ "status": _status(action_text, action_node, join_url),
+ "status_label": action_text or "Status unbekannt",
+ "join_url": join_url,
+ "can_join": bool(join_url and _status(action_text, action_node, join_url) == "open"),
+ "links": _links(card, source_url, action_node),
+ }
+ )
+ return rooms
+
+
+def parse_videokonferenz_overview(
+ html: str,
+ source_url: str = "https://start.schulportal.hessen.de/videokonferenz.php",
+) -> dict[str, Any]:
+ """Parse the video-room list without exposing session data or page markup."""
+ soup = BeautifulSoup(html or "", "html.parser")
+ source_host = (urlsplit(source_url).hostname or "").casefold()
+ if (
+ source_host == "login.schulportal.hessen.de"
+ or soup.select_one("form[action*='login']")
+ or "login.schulportal.hessen.de" in (html or "")
+ ):
+ return {"success": False, "error": "Schulportal session expired", "error_kind": "authentication"}
+
+ rooms: list[dict[str, Any]] = []
+ recognized = False
+ for table in soup.select("table"):
+ parsed = _parse_table(table, source_url)
+ if parsed:
+ recognized = True
+ rooms.extend(parsed)
+ continue
+ headers = " ".join(_text(cell).casefold() for cell in table.select("th"))
+ recognized = recognized or ("lerngruppe" in headers and "aktion" in headers)
+
+ if not rooms:
+ rooms = _parse_cards(soup, source_url)
+ recognized = recognized or bool(rooms)
+
+ page_text = _text(soup)
+ recognized = recognized or (
+ "videokonferenz" in page_text.casefold()
+ and ("ansicht aktuell" in page_text.casefold() or "lerngruppe" in page_text.casefold())
+ )
+ updated_match = _UPDATED_RE.search(page_text)
+ updated_label = updated_match.group(1).strip(" .") if updated_match and updated_match.group(1) else None
+
+ # Prevent duplicate rows when responsive markup contains the same room twice.
+ unique_rooms = list({room["id"]: room for room in rooms}.values())
+ return {
+ "success": True,
+ "available": recognized,
+ "source": "schulportal",
+ "rooms": unique_rooms,
+ "count": len(unique_rooms),
+ "open_count": sum(room["status"] == "open" for room in unique_rooms),
+ "updated_label": updated_label,
+ }
+
+
+def videokonferenz_get_rooms(self: Any) -> dict[str, Any]:
+ """Fetch the authenticated student's current video-conference rooms."""
+ if not self.logged_in:
+ return {"success": False, "error": "Not logged in", "error_kind": "authentication"}
+ try:
+ response = self.session.get(
+ f"{self.BASE_START_URL}/videokonferenz.php",
+ timeout=(10, 30),
+ )
+ response.raise_for_status()
+ result = parse_videokonferenz_overview(
+ response.text,
+ getattr(response, "url", f"{self.BASE_START_URL}/videokonferenz.php"),
+ )
+ if not result.get("success") or not result.get("rooms"):
+ result["status_live"] = bool(result.get("success"))
+ return result
+
+ try:
+ status_response = self.session.get(
+ f"{self.BASE_START_URL}/videokonferenz.php",
+ params={"a": "sus_start", "b": "update"},
+ timeout=(10, 30),
+ )
+ status_response.raise_for_status()
+ payload = json.loads(status_response.text)
+ if not isinstance(payload, list):
+ raise TypeError("Unexpected video-room status response")
+ open_room_ids = {str(value) for value in payload}
+ for room in result["rooms"]:
+ is_open = str(room["id"]) in open_room_ids
+ room["status"] = "open" if is_open else "waiting"
+ room["status_label"] = "Raum betreten" if is_open else "Raum nicht offen"
+ room["can_join"] = bool(is_open and room.get("join_url"))
+ result["open_count"] = sum(
+ room["status"] == "open" for room in result["rooms"]
+ )
+ result["status_live"] = True
+ except (requests.RequestException, ValueError, TypeError, json.JSONDecodeError):
+ # The overview remains useful, but the server-rendered buttons are
+ # only placeholders until this polling endpoint responds.
+ for room in result["rooms"]:
+ room["status"] = "unknown"
+ room["status_label"] = "Aktueller Status nicht verfügbar"
+ room["can_join"] = False
+ result["open_count"] = 0
+ result["status_live"] = False
+ return result
+ except requests.RequestException as exc:
+ return {"success": False, "error": f"Failed to fetch video rooms: {exc}", "error_kind": "upstream"}
+ except Exception as exc: # noqa: BLE001 - normalize unexpected parser failures
+ return {"success": False, "error": f"Failed to parse video rooms: {exc}", "error_kind": "parsing"}
diff --git a/schulportal_hessen/base.py b/schulportal_hessen/base.py
index 646da3f..11e64b8 100644
--- a/schulportal_hessen/base.py
+++ b/schulportal_hessen/base.py
@@ -296,6 +296,7 @@ def get_available_modules(
"wahlen_get_form",
"wahlen_submit",
],
+ "videokonferenz.php": ["videokonferenz_get_rooms"],
}
for entry in entries:
@@ -804,6 +805,11 @@ def close(self):
SchulportalHessenAPI.vertretungsplan_get_plan = vertretungsplan_get_plan
+# Import and attach the Videokonferenz methods
+from .applets.videokonferenz.api import videokonferenz_get_rooms
+
+SchulportalHessenAPI.videokonferenz_get_rooms = videokonferenz_get_rooms
+
# Import and attach the stundenplan methods
from .applets.stundenplan.api import stundenplan_get_plan
diff --git a/tests/fixtures/videokonferenz_overview.html b/tests/fixtures/videokonferenz_overview.html
new file mode 100644
index 0000000..34e99f7
--- /dev/null
+++ b/tests/fixtures/videokonferenz_overview.html
@@ -0,0 +1,37 @@
+
+
+
+ Videokonferenz
+ Ansicht aktuell seit 08:14 Uhr
+
+
+
diff --git a/tests/test_videokonferenz.py b/tests/test_videokonferenz.py
new file mode 100644
index 0000000..c41935f
--- /dev/null
+++ b/tests/test_videokonferenz.py
@@ -0,0 +1,223 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+
+import requests
+from fastapi import Response
+
+from api import api as api_module
+from schulportal_hessen.applets.videokonferenz.api import (
+ parse_videokonferenz_overview,
+ videokonferenz_get_rooms,
+)
+
+FIXTURE = Path(__file__).parent / "fixtures" / "videokonferenz_overview.html"
+
+
+def test_parse_room_overview_and_join_states() -> None:
+ result = parse_videokonferenz_overview(
+ FIXTURE.read_text(encoding="utf-8"),
+ "https://start.schulportal.hessen.de/videokonferenz.php",
+ )
+
+ assert result["success"] is True
+ assert result["available"] is True
+ assert result["count"] == 3
+ assert result["open_count"] == 1
+ assert result["updated_label"] == "08:14 Uhr"
+
+ open_room, waiting_room, closed_room = result["rooms"]
+ assert open_room == {
+ "id": "mathe-10a",
+ "name": "Mathematik 10a",
+ "teachers": ["FRA", "BEI"],
+ "status": "open",
+ "status_label": "Raum betreten",
+ "join_url": "https://start.schulportal.hessen.de/videokonferenz.php?a=join&room=42",
+ "can_join": True,
+ "links": [{"label": "Raumregeln", "url": "https://example.edu/regeln"}],
+ }
+ assert waiting_room["id"] == "91"
+ assert waiting_room["teachers"] == ["WIN", "SOM"]
+ assert waiting_room["status"] == "waiting"
+ assert waiting_room["links"] == []
+ assert closed_room["status"] == "closed"
+ assert closed_room["can_join"] is False
+
+
+def test_recognized_empty_overview_is_available() -> None:
+ result = parse_videokonferenz_overview(
+ "Videokonferenz
Ansicht aktuell
"
+ ""
+ )
+
+ assert result["success"] is True
+ assert result["available"] is True
+ assert result["rooms"] == []
+
+
+def test_login_page_is_reported_as_authentication_failure() -> None:
+ result = parse_videokonferenz_overview(
+ ''
+ )
+
+ assert result["success"] is False
+ assert result["error_kind"] == "authentication"
+
+
+def test_client_requires_login_and_uses_authenticated_page() -> None:
+ assert videokonferenz_get_rooms(SimpleNamespace(logged_in=False))["error_kind"] == "authentication"
+
+ class Response:
+ text = FIXTURE.read_text(encoding="utf-8")
+ url = "https://start.schulportal.hessen.de/videokonferenz.php"
+
+ def raise_for_status(self) -> None:
+ return None
+
+ class StatusResponse:
+ text = '["mathe-10a"]'
+
+ def raise_for_status(self) -> None:
+ return None
+
+ class Session:
+ def __init__(self) -> None:
+ self.calls = []
+
+ def get(self, url, **kwargs):
+ self.calls.append((url, kwargs))
+ if kwargs.get("params"):
+ return StatusResponse()
+ return Response()
+
+ client = SimpleNamespace(
+ logged_in=True,
+ BASE_START_URL="https://start.schulportal.hessen.de",
+ session=Session(),
+ )
+ result = videokonferenz_get_rooms(client)
+
+ assert result["success"] is True
+ assert result["status_live"] is True
+ assert result["open_count"] == 1
+ assert client.session.calls == [
+ (
+ "https://start.schulportal.hessen.de/videokonferenz.php",
+ {"timeout": (10, 30)},
+ ),
+ (
+ "https://start.schulportal.hessen.de/videokonferenz.php",
+ {"params": {"a": "sus_start", "b": "update"}, "timeout": (10, 30)},
+ ),
+ ]
+
+
+def test_failed_live_status_never_exposes_placeholder_join_links() -> None:
+ class Response:
+ text = FIXTURE.read_text(encoding="utf-8")
+ url = "https://start.schulportal.hessen.de/videokonferenz.php"
+
+ def raise_for_status(self) -> None:
+ return None
+
+ class Session:
+ def get(self, _url, **kwargs):
+ if kwargs.get("params"):
+ raise requests.Timeout("status unavailable")
+ return Response()
+
+ client = SimpleNamespace(
+ logged_in=True,
+ BASE_START_URL="https://start.schulportal.hessen.de",
+ session=Session(),
+ )
+
+ result = videokonferenz_get_rooms(client)
+
+ assert result["success"] is True
+ assert result["status_live"] is False
+ assert result["open_count"] == 0
+ assert {room["status"] for room in result["rooms"]} == {"unknown"}
+ assert all(room["can_join"] is False for room in result["rooms"])
+
+
+def test_route_never_caches_user_bound_join_urls(monkeypatch) -> None:
+ auth = SimpleNamespace(
+ user_id="5201:test",
+ client=SimpleNamespace(
+ videokonferenz_get_rooms=lambda: {
+ "success": True,
+ "available": True,
+ "rooms": [],
+ "count": 0,
+ "open_count": 0,
+ }
+ ),
+ )
+ async def run_in_threadpool(function, *args):
+ return function(*args)
+
+ monkeypatch.setattr(api_module, "run_in_threadpool", run_in_threadpool)
+
+ first_response = Response()
+ fresh_response = Response()
+ first = asyncio.run(api_module.get_videokonferenz(response=first_response, refresh=False, auth=auth))
+ fresh = asyncio.run(api_module.get_videokonferenz(response=fresh_response, refresh=True, auth=auth))
+
+ assert first["rooms"] == []
+ assert fresh["rooms"] == []
+ assert first_response.headers["cache-control"] == "private, no-store"
+ assert fresh_response.headers["cache-control"] == "private, no-store"
+
+
+def test_insecure_absolute_urls_are_not_exposed() -> None:
+ result = parse_videokonferenz_overview(
+ """
+ Videokonferenz
Ansicht aktuell
+
+ """
+ )
+
+ assert result["rooms"][0]["join_url"] is None
+ assert result["rooms"][0]["can_join"] is False
+ assert result["rooms"][0]["links"] == []
+
+
+def test_production_style_blank_header_uses_visible_room_state() -> None:
+ result = parse_videokonferenz_overview(
+ """
+ Videokonferenz
Ansicht aktuell
+
+ """
+ )
+
+ assert result["count"] == 1
+ assert result["rooms"][0]["name"] == "Mathematik"
+ assert result["rooms"][0]["status"] == "waiting"
+ assert result["rooms"][0]["can_join"] is False
+
+
+def test_module_registry_marks_videokonferenz_usable(monkeypatch) -> None:
+ client = api_module.SchulportalHessenAPI()
+ client.logged_in = True
+ monkeypatch.setattr(client, "_resolve_direct_url", lambda url: url)
+ monkeypatch.setattr(client.session, "get", lambda *args, **kwargs: SimpleNamespace(headers={}, close=lambda: None))
+
+ modules = client.get_available_modules(
+ {
+ "success": True,
+ "data": {"entrys": [{"Name": "Videokonferenz", "link": "videokonferenz.php"}]},
+ }
+ )
+
+ assert modules[0]["usable"] is True
+ assert modules[0]["usage"] == ["videokonferenz_get_rooms"]