init

backend/app/sources/laundry.py

@@ -0,0 +1,194 @@
+"""Campus laundry bookings (4ka.inbicst.ru) as time commitments.
+
+The site has no API and only offers Yandex SSO, so authentication *is* the
+session cookie you copy out of your browser — the approach jira-helper settled
+on. On the booking grid every slot is a `div.period`, and the ones you booked
+carry an extra `my` class, which is the only thing that makes "my laundry
+today" findable.
+
+These are not todos you tick off: they are blocks of time that will interrupt
+your pomodoros, so the UI shows them on the timer page and the daily report
+mentions them.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import date, datetime, time
+from typing import Any
+from urllib.parse import urlsplit
+
+import httpx
+from bs4 import BeautifulSoup
+
+from app.settings_schema import LaundrySourceSettings
+from app.sources.base import SourceError, TodoItem
+
+_PROFILE_PATH = "/profile/"
+_GRID_PATH = "/"
+_SESSION_COOKIE = "sessionid"
+_COOKIE_PAIR_RE = re.compile(r"^[A-Za-z0-9_.\-]+=")
+_TIME_RANGE_RE = re.compile(r"(\d{1,2}):(\d{2})\s*[-–—]\s*(\d{1,2}):(\d{2})")
+_USER_AGENT = "Mozilla/5.0 (compatible; kadence laundry reader)"
+
+_EXPIRED_HINT = (
+    "the laundry session cookie is no longer valid. Sign in at {base_url} in "
+    "your browser, copy the fresh `sessionid` cookie and paste it into "
+    "Settings → Sources → Laundry."
+)
+
+
+def parse_cookie_header(raw: str) -> dict[str, str]:
+    """Accept either a bare `sessionid` value or a whole copied cookie string."""
+    if not _COOKIE_PAIR_RE.match(raw):
+        return {_SESSION_COOKIE: raw}
+    cookies: dict[str, str] = {}
+    for part in raw.split(";"):
+        name, _, value = part.strip().partition("=")
+        if name.strip() and value.strip():
+            cookies[name.strip()] = value.strip()
+    return cookies
+
+
+@dataclass(slots=True)
+class LaundrySlot:
+    location: str
+    machine: str
+    time_range: str
+    start: time | None
+    finished: bool
+    current: bool
+    refunded: bool
+
+    @property
+    def is_upcoming(self) -> bool:
+        return not self.finished and not self.refunded
+
+
+class LaundrySource:
+    name = "laundry"
+
+    def __init__(self, config: LaundrySourceSettings, *, timeout: int = 30) -> None:
+        if not config.session_cookie:
+            raise SourceError("The laundry reader needs a `sessionid` cookie.")
+        cookies = parse_cookie_header(config.session_cookie)
+        if _SESSION_COOKIE not in cookies:
+            raise SourceError(
+                f"The laundry cookie has no {_SESSION_COOKIE!r} value in it."
+            )
+        self._base_url = config.base_url.rstrip("/")
+        host = urlsplit(self._base_url).hostname or ""
+        jar = httpx.Cookies()
+        for key, value in cookies.items():
+            jar.set(key, value, domain=host, path="/")
+        self._client = httpx.Client(
+            base_url=self._base_url,
+            cookies=jar,
+            headers={"User-Agent": _USER_AGENT},
+            timeout=timeout,
+            follow_redirects=False,
+        )
+
+    def _check_session(self) -> None:
+        """An expired cookie silently serves the anonymous page, where none of
+        the slots are marked as yours — indistinguishable from "nothing booked".
+        So verify first."""
+        try:
+            resp = self._client.get(_PROFILE_PATH)
+        except httpx.HTTPError as exc:
+            raise SourceError(f"The laundry site is unreachable: {exc}") from exc
+        if resp.status_code != 200:
+            raise SourceError(_EXPIRED_HINT.format(base_url=self._base_url))
+
+    def fetch(self) -> list[TodoItem]:
+        self._check_session()
+        try:
+            resp = self._client.get(_GRID_PATH)
+        except httpx.HTTPError as exc:
+            raise SourceError(f"The laundry site is unreachable: {exc}") from exc
+        if not resp.is_success:
+            raise SourceError(f"Laundry grid -> {resp.status_code}")
+
+        _, slots = parse_my_slots(resp.text)
+        today = date.today()
+        return [
+            TodoItem(
+                source=self.name,
+                external_id=f"{today.isoformat()}:{slot.machine}:{slot.time_range}",
+                title=f"{slot.time_range} — {slot.machine} ({slot.location})",
+                project="Laundry",
+                status="running now" if slot.current else "booked",
+                due_date=today,
+                raw={
+                    "start": slot.start.isoformat() if slot.start else None,
+                    "current": slot.current,
+                },
+            )
+            for slot in slots
+            if slot.is_upcoming
+        ]
+
+    def check(self) -> str:
+        self._check_session()
+        return f"signed in at {self._base_url}"
+
+    def complete(self, item: TodoItem) -> bool:
+        return False  # the washing machine finishes on its own
+
+    def close(self) -> None:
+        self._client.close()
+
+
+def _text(node: Any) -> str:
+    return node.get_text(strip=True) if node is not None else ""
+
+
+def _parse_start(time_range: str) -> time | None:
+    match = _TIME_RANGE_RE.search(time_range)
+    if not match:
+        return None
+    hour, minute = int(match.group(1)), int(match.group(2))
+    if not (0 <= hour <= 23 and 0 <= minute <= 59):
+        return None
+    return time(hour, minute)
+
+
+def parse_my_slots(html: str) -> tuple[str, list[LaundrySlot]]:
+    """Pull your own slots off a booking grid. Returns `(day_label, slots)`."""
+    soup = BeautifulSoup(html, "html.parser")
+    day = soup.select_one(".date-machines")
+    if day is None:
+        raise SourceError(
+            "No booking grid in the response — the layout changed, or the "
+            "session was not accepted."
+        )
+    day_label = (day.get("id") or "").strip()
+
+    slots: list[LaundrySlot] = []
+    for block in day.select(".machines"):
+        location = _text(block.select_one(".floor-title")) or "?"
+        for machine in block.select(".machine"):
+            machine_name = _text(machine.select_one(".machine-title")) or "?"
+            for period in machine.select(".period.my"):
+                classes = set(period.get("class") or [])
+                time_range = _text(period.find("div")) or "?"
+                slots.append(
+                    LaundrySlot(
+                        location=location,
+                        machine=machine_name,
+                        time_range=time_range,
+                        start=_parse_start(time_range),
+                        finished="finished" in classes,
+                        current="current" in classes,
+                        refunded="refunded" in classes,
+                    )
+                )
+
+    slots.sort(key=lambda slot: (slot.start is None, slot.start or time.min, slot.machine))
+    return day_label, slots
+
+
+def slot_is_now(slot: LaundrySlot, moment: datetime | None = None) -> bool:
+    """Whether the machine is running at `moment` (used by the timer page)."""
+    return slot.current