kadence/backend/app/sources/laundry.py

"""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 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