Add registration and consultation APIs; preserve standalone grades

README.md

@@ -42,3 +42,100 @@ The docs are available [here](https://edupageapi.github.io/edupage-api/)
 
 # Discord
 https://discord.gg/fg6zBu9ZAn
+
+## Registrations, surveys and consultations (read only)
+
+The following methods use the authenticated session in its current account/child
+context. For a parent account, select the intended child with `switch_to_child`
+and restore the parent with `switch_to_parent` in a `finally` block.
+
+```python
+# School timezone; change for schools outside this timezone.
+actions = edupage.get_registrations(timezone_name="Europe/Prague")
+for action in actions:
+    print(action.name, action.closes_at, action.votable)
+    # Survey questions retain their EduPage JSON content; answers belong only
+    # to the current participant. No question JavaScript is executed.
+    print(action.targets, action.selection)
+
+history = edupage.get_registration_history()
+overview = edupage.get_consultations()
+for consultation in overview.consultations:
+    slots = edupage.get_consultation_slots(consultation.consultation_id)
+    print(consultation.teacher_name, slots)
+print(overview.reservations)
+```
+
+Assigned actions can include closed registrations. `active`, `votable` and the
+opening/closing dates must be considered together. A timeline notification's
+completion flag does not establish that a participant answered a survey.
+Selection statuses preserve EduPage's values (`ok`, `reserved`, `answered`, or
+other values). Parent confirmation and temporary reservation expiry are separate.
+
+Consultation slots preserve server availability rather than inferring it from
+the block duration. Their status is `free`, `reserved`, `temporary`, or `unknown`;
+`own` and `closed` are separate flags. Unknown states must never be considered
+available. Local timestamps are timezone aware; slot epoch timestamps use UTC.
+Different consultation blocks may overlap, so preserve the block identifier.
+
+These methods do not reserve, cancel, confirm, or submit answers. They send POST
+requests for read operations because that is how EduPage loads these views.
+HTTP failures, expired login pages and missing expected data raise exceptions;
+they do not silently become empty lists.
+
+Compatibility: HTML/bootstrap and slot shapes were inspected on one parent
+account. Tests use synthetic, anonymized examples. Nonempty own reservations,
+occupied-slot variants and actual survey question payloads still need live
+compatibility testing. Raw survey content is intentionally not interpreted as a
+universal questionnaire schema. No credentials or real school data are fixtures.
+
+### Explicit consultation booking
+
+Only the methods below perform writes. `dry_run=True` performs the same fresh
+availability/ownership checks but never calls reserve, confirm or cancel.
+
+```python
+from datetime import datetime
+
+plan = edupage.book_consultation_slot(
+    "3", datetime.fromisoformat("2026-11-10T13:30:00+01:00"), dry_run=True
+)
+# Only after the user chooses to make this real reservation:
+reservation = edupage.book_consultation_slot(
+    "3", datetime.fromisoformat("2026-11-10T13:30:00+01:00")
+)
+# Explicit cancellation; verifies ownership and refuses ambiguous blocks:
+edupage.cancel_consultation(reservation.visit_id, dry_run=True)
+```
+
+Booking verifies the server slot, creates a temporary hold, confirms its visit ID
+and reads back a confirmed own slot. No automatic retry is made after a write:
+`ConsultationUncertain` means the user must inspect EduPage before trying again.
+The server's cancellation command addresses an entire consultation block, so the
+client first verifies that the requested visit is the only own visit in it.
+`ConsultationSlotTaken`, `ConsultationClosed`, `ConsultationLimitReached`,
+`ConsultationNotOwned` and `ExpiredSessionException` distinguish known failures.
+Unrecognized localized server errors remain a generic `ConsultationError` rather
+than being guessed. Participant notes are not supported by the inspected write
+protocol; passing a nonempty `note` raises an explicit error before any write.
+
+Consultation blocks also expose `teaches_student` from EduPage's `coUci` map
+(`None` means unknown), and `is_class_teacher` for client enrichment.
+
+Write behavior is covered by synthetic tests. No real booking or cancellation
+was used during development. Real success/error responses still need supervised
+validation on a term the user actually intends to book.
+
+### Standalone grades and read completeness
+
+Some schools return marks with udalostid=0, without a paper/event. get_grades()
+and term reads retain these marks using the row's predmetid, datum, data and
+ucitelid. EduGrade.event_id is None for such marks; grade_id retains znamkaid
+when provided. Consumers must not confuse these IDs with timeline event IDs.
+Missing standalone weights use 1. Existing event metadata continues to supply
+weights for marks attached to papers/events.
+
+The latest successful get_grades() call exposes grade_read_report on Edupage:
+received, parsed, skipped and skip_reasons. Copy it before another child/term
+read if needed. Skipped rows log field names and reasons without grade values.
+Invalid event metadata fails explicitly instead of silently dropping a mark.

edupage_api/__init__.py

@@ -25,6 +25,9 @@
     EduTeacher,
     People,
 )
+from edupage_api.registrations import (
+    Registrations, Registration, RegistrationHistory, ConsultationOverview, ConsultationSlot,
+)
 from edupage_api.ringing import RingingTime, RingingTimes
 from edupage_api.subjects import Subject, Subjects
 from edupage_api.substitution import Substitution, TimetableChange
@@ -33,6 +36,38 @@
 
 
 class Edupage(EdupageModule):
+    def book_consultation_slot(self, consultation_id, starts_at, *, dry_run=False, note=None,
+                               timezone_name="Europe/Prague"):
+        """Explicitly book and verify a confirmed consultation; dry_run only reads."""
+        from edupage_api.consultation_booking import ConsultationBooking
+        return ConsultationBooking(self, timezone_name).book_consultation_slot(
+            consultation_id, starts_at, dry_run=dry_run, note=note)
+
+    def cancel_consultation(self, visit_id, *, dry_run=False, timezone_name="Europe/Prague"):
+        """Cancel only an unambiguously owned visit; dry_run only reads."""
+        from edupage_api.consultation_booking import ConsultationBooking
+        return ConsultationBooking(self, timezone_name).cancel_consultation(visit_id, dry_run=dry_run)
+
+    def get_registrations(self, timezone_name="Europe/Prague") -> list[Registration]:
+        """Read assigned registration actions and surveys in the current context."""
+        return Registrations(self, timezone_name).get_registrations()
+
+    def get_registration(self, action_id: str, timezone_name="Europe/Prague") -> Registration:
+        """Read an action, its questions and own selections."""
+        return Registrations(self, timezone_name).get_registration(action_id)
+
+    def get_registration_history(self, timezone_name="Europe/Prague") -> list[RegistrationHistory]:
+        """Read own registration participation history."""
+        return Registrations(self, timezone_name).get_registration_history()
+
+    def get_consultations(self, timezone_name="Europe/Prague") -> ConsultationOverview:
+        """Read teacher consultation blocks and own reservations."""
+        return Registrations(self, timezone_name).get_consultations()
+
+    def get_consultation_slots(self, consultation_id: str, timezone_name="Europe/Prague") -> list[ConsultationSlot]:
+        """Read a consultation's slot availability without reserving."""
+        return Registrations(self, timezone_name).get_consultation_slots(consultation_id)
+
     def __init__(self, request_timeout=5):
         """Initialize `Edupage` object.
 

edupage_api/consultation_booking.py

@@ -0,0 +1,218 @@
+"""Explicit consultation writes. Read-only Registrations never calls this module."""
+
+import json
+import re
+from datetime import datetime, timezone
+from urllib.parse import parse_qs, urlencode, urlsplit
+
+from requests import RequestException
+
+from edupage_api.exceptions import (
+    ConsultationClosed,
+    ConsultationError,
+    ConsultationLimitReached,
+    ConsultationNotOwned,
+    ConsultationSlotNotFound,
+    ConsultationSlotTaken,
+    ConsultationUncertain,
+    ExpiredSessionException,
+)
+from edupage_api.module import ModuleHelper
+from edupage_api.registrations import _HTML, ConsultationReservation, Registrations
+
+
+class ConsultationBooking(Registrations):
+    def _detail(self, consultation_id):
+        path = "/calendar/?" + urlencode(
+            {
+                "cmd": "KonzultacneHodinyVisit",
+                "konzultaciaids": str(consultation_id),
+                "asDialog": "1",
+            }
+        )
+        text = self._request(path).text
+        # Only extract the literal transport URL. Never execute a script.
+        for script in _HTML(text).scripts:
+            match = re.search(r"\$j\.post\('((?:\\.|[^'\\])*)&akcia=rezervuj'", script)
+            if match:
+                literal = json.loads('"' + match[1].replace('"', '\\"') + '"')
+                parts = urlsplit(literal)
+                params = parse_qs(parts.query)
+                if (
+                    parts.scheme
+                    or parts.netloc
+                    or parts.path != "/calendar/"
+                    or params.get("cmd") != ["KonzultacneHodinyVisit"]
+                    or not params.get("jwid")
+                ):
+                    raise ConsultationError("Unexpected consultation form")
+                path = "/calendar/?" + urlencode(
+                    {
+                        "cmd": "KonzultacneHodinyVisit",
+                        "jwid": params["jwid"][0],
+                        "asDialog": "1",
+                    }
+                )
+                return path, self._slots(text, consultation_id)
+        raise ConsultationError("Consultation booking form missing")
+
+    def _write(self, path, action, data, *, json_required=True):
+        if action not in ("rezervuj", "confirm", "zrus"):
+            raise ValueError("Unsupported consultation write")
+        try:
+            response = self.edupage.session.post(
+                f"https://{self.edupage.subdomain}.edupage.org{path}&akcia={action}",
+                data=data,
+            )
+        except RequestException as exc:
+            # No retry: the server may already have accepted the operation.
+            raise ConsultationUncertain(
+                "Write outcome is unknown; check EduPage"
+            ) from exc
+        if "/login" in response.url or re.search(
+            r"<input[^>]+type=[\'\"]password", response.text, re.I
+        ):
+            raise ExpiredSessionException("EduPage session expired")
+        if response.status_code != 200:
+            raise ConsultationUncertain("Write outcome is unknown; check EduPage")
+        try:
+            result = response.json()
+        except ValueError as exc:
+            if not json_required:
+                return None  # Cancellation is verified by a fresh read.
+            raise ConsultationUncertain(
+                "Unexpected write response; check EduPage"
+            ) from exc
+        if not isinstance(result, dict):
+            if not json_required:
+                return None
+            raise ConsultationUncertain("Unexpected write response; check EduPage")
+        if result.get("status") == "ok":
+            return result
+        if not json_required and "status" not in result:
+            return result
+        # Only recognized machine-readable codes are classified. Do not infer
+        # server error meanings from localized free text.
+        code = result.get("code") or (result.get("data") or {}).get("code")
+        error = {
+            "slot_taken": ConsultationSlotTaken,
+            "booking_closed": ConsultationClosed,
+            "booking_limit": ConsultationLimitReached,
+        }.get(code, ConsultationError)
+        raise error("EduPage rejected the consultation request")
+
+    @ModuleHelper.logged_in
+    def book_consultation_slot(
+        self, consultation_id, starts_at, *, dry_run=False, note=None
+    ):
+        if note:
+            # The inspected booking protocol has no note field. Never discard it silently.
+            error = ConsultationError(
+                "Participant notes are not supported by this protocol"
+            )
+            error.code = "note_unsupported"
+            raise error
+        if (
+            not isinstance(starts_at, datetime)
+            or starts_at.tzinfo is None
+            or starts_at.utcoffset() is None
+        ):
+            raise ValueError("starts_at must be a timezone-aware datetime")
+        now = datetime.now(timezone.utc)
+        if starts_at <= now:
+            raise ConsultationClosed("Consultation is in the past")
+        overview = self.get_consultations()
+        block = next(
+            (
+                b
+                for b in overview.consultations
+                if b.consultation_id == str(consultation_id)
+            ),
+            None,
+        )
+        if block is None:
+            raise ConsultationClosed("Consultation no longer offered")
+        if (block.booking_opens_at and now < block.booking_opens_at) or (
+            block.booking_closes_at and now > block.booking_closes_at
+        ):
+            raise ConsultationClosed("Booking window is closed")
+        path, slots = self._detail(consultation_id)
+        if any(s.own for s in slots):
+            raise ConsultationLimitReached(
+                "An own reservation already exists in this block"
+            )
+        slot = next((s for s in slots if s.starts_at == starts_at), None)
+        if slot is None:
+            raise ConsultationSlotNotFound(
+                "This time does not exist in the consultation block"
+            )
+        if slot.status != "free":
+            raise ConsultationSlotTaken("Slot is no longer free")
+        if slot.closed:
+            raise ConsultationClosed("Slot is closed")
+        epoch = int(slot.starts_at.timestamp())
+        if starts_at.timestamp() != epoch:
+            raise ValueError("Slot start must match an exact server timeframe")
+        plan = {
+            "consultation_id": str(consultation_id),
+            "start": starts_at.isoformat(),
+            "reserve": {"konzultaciaid": str(consultation_id), "cas": str(epoch)},
+            "confirm": {"visitids": "<returned visit_id>"},
+        }
+        if dry_run:
+            return plan
+        held = self._write(path, "rezervuj", plan["reserve"])
+        visit_id = str((held.get("data") or {}).get("visitid") or "")
+        if not visit_id:
+            raise ConsultationUncertain(
+                "Missing visit ID after reservation; check EduPage"
+            )
+        try:
+            self._write(path, "confirm", {"visitids": visit_id})
+            confirmed = self.get_consultation_slots(consultation_id)
+        except ExpiredSessionException:
+            raise
+        except Exception as exc:
+            raise ConsultationUncertain(
+                f"Reservation {visit_id} may be held; check EduPage"
+            ) from exc
+        if not any(
+            s.own
+            and s.visit_id == visit_id
+            and s.status == "reserved"
+            and s.starts_at == starts_at
+            for s in confirmed
+        ):
+            raise ConsultationUncertain(
+                f"Reservation {visit_id} was not verified as confirmed"
+            )
+        return ConsultationReservation(str(consultation_id), starts_at, visit_id)
+
+    @ModuleHelper.logged_in
+    def cancel_consultation(self, visit_id, *, dry_run=False):
+        # The server cancels by block, not visit ID. Resolve own ownership first
+        # and refuse ambiguous blocks, rather than cancelling an unrelated visit.
+        overview = self.get_consultations()
+        booking = next(
+            (r for r in overview.reservations if r.visit_id == str(visit_id)), None
+        )
+        if booking is None:
+            raise ConsultationNotOwned("Own booking not found")
+        path, slots = self._detail(booking.consultation_id)
+        own = [s for s in slots if s.own]
+        if len(own) != 1 or own[0].visit_id != str(visit_id):
+            raise ConsultationNotOwned("Booking ownership is ambiguous")
+        plan = {
+            "visit_id": str(visit_id),
+            "cancel": {"konzultaciaid": booking.consultation_id},
+        }
+        if dry_run:
+            return plan
+        self._write(path, "zrus", plan["cancel"], json_required=False)
+        try:
+            remaining = self.get_consultation_slots(booking.consultation_id)
+        except Exception as exc:
+            raise ConsultationUncertain("Cancellation could not be verified") from exc
+        if any(s.own and s.visit_id == str(visit_id) for s in remaining):
+            raise ConsultationUncertain("Reservation still present after cancellation")
+        return None