"""Zero-dependency Python client for the public TotemClock Time Tools API. Example: from totemclock_time_tools import TotemClockClient client = TotemClockClient() result = client.calculate_worked_hours( start="22:00", end="06:30", break_minutes=30 ) print(result["formatted"]) """ from __future__ import annotations import json from typing import Any, Callable, Iterable, Mapping, Optional, Sequence, Tuple from urllib.error import HTTPError from urllib.parse import urlencode from urllib.request import Request, urlopen DEFAULT_BASE_URL = "https://totemclock.com" Parameter = Tuple[str, Any] Opener = Callable[..., Any] class TotemClockApiError(RuntimeError): """An HTTP error returned by the TotemClock API.""" def __init__(self, status: int, body: Any) -> None: message = body.get("message") if isinstance(body, Mapping) else None super().__init__(message or f"TotemClock API request failed with HTTP {status}") self.status = status self.body = body class TotemClockClient: """Small synchronous client that uses only the Python standard library.""" def __init__( self, base_url: str = DEFAULT_BASE_URL, *, timeout: float = 10, opener: Optional[Opener] = None, ) -> None: if timeout <= 0: raise ValueError("timeout must be greater than zero") self.base_url = str(base_url).rstrip("/") self.timeout = timeout self._opener = opener or urlopen @staticmethod def _parameters(values: Iterable[Parameter]) -> list[Parameter]: return [ (key, value) for key, value in values if value is not None and value != "" ] @staticmethod def _decode_body(response: Any) -> Any: raw = response.read() if not raw: return None try: return json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return raw.decode("utf-8", errors="replace") def _get( self, path: str, parameters: Iterable[Parameter], *, accept: str = "application/json", ) -> Any: query = urlencode(self._parameters(parameters)) url = f"{self.base_url}{path}" if query: url = f"{url}?{query}" request = Request(url, headers={"Accept": accept}, method="GET") try: with self._opener(request, timeout=self.timeout) as response: return self._decode_body(response) except HTTPError as error: body = self._decode_body(error) raise TotemClockApiError(error.code, body) from error def _post( self, path: str, body: Mapping[str, Any], *, accept: str = "application/json", ) -> Any: request = Request( f"{self.base_url}{path}", data=json.dumps(body, separators=(",", ":")).encode("utf-8"), headers={"Accept": accept, "Content-Type": "application/json"}, method="POST", ) try: with self._opener(request, timeout=self.timeout) as response: return self._decode_body(response) except HTTPError as error: response_body = self._decode_body(error) raise TotemClockApiError(error.code, response_body) from error def convert_decimal_hours( self, *, time: Optional[str] = None, decimal: Optional[float] = None, ) -> Any: """Convert HH:MM to decimal hours, or decimal hours to HH:MM.""" return self._get( "/api/tools/decimal-hours", (("time", time), ("decimal", decimal)), ) def calculate_worked_hours( self, *, start: str, end: str, break_minutes: Optional[int] = None, ) -> Any: """Calculate worked time for a shift, including overnight shifts.""" return self._get( "/api/tools/worked-hours", (("start", start), ("end", end), ("breakMinutes", break_minutes)), ) def summarize_timesheet( self, entries: Sequence[str], *, regular_minutes_limit: Optional[int] = None, hourly_rate: Optional[float] = None, overtime_multiplier: Optional[float] = None, currency: Optional[str] = None, ) -> Any: """Total durations and optionally apply caller-defined pay inputs.""" parameters: list[Parameter] = [("entry", entry) for entry in entries] parameters.extend( [ ("regularMinutesLimit", regular_minutes_limit), ("hourlyRate", hourly_rate), ("overtimeMultiplier", overtime_multiplier), ("currency", currency), ] ) return self._get("/api/tools/timesheet-summary", parameters) def summarize_timesheet_json( self, entries: Sequence[str], *, regular_minutes_limit: Optional[int] = None, hourly_rate: Optional[float] = None, overtime_multiplier: Optional[float] = None, currency: Optional[str] = None, ) -> Any: """POST timesheet durations as JSON for larger server-side batches.""" body = { key: value for key, value in { "entries": list(entries), "regularMinutesLimit": regular_minutes_limit, "hourlyRate": hourly_rate, "overtimeMultiplier": overtime_multiplier, "currency": currency, }.items() if value is not None } return self._post("/api/tools/timesheet-summary", body) def summarize_worked_shifts( self, shifts: Sequence[Sequence[Any]], *, regular_minutes_limit: Optional[int] = None, hourly_rate: Optional[float] = None, overtime_multiplier: Optional[float] = None, currency: Optional[str] = None, ) -> Any: """Total start/end shifts, with an optional third break-minutes value.""" parameters: list[Parameter] = [ ("shift", ",".join(str(value) for value in shift)) for shift in shifts ] parameters.extend( [ ("regularMinutesLimit", regular_minutes_limit), ("hourlyRate", hourly_rate), ("overtimeMultiplier", overtime_multiplier), ("currency", currency), ] ) return self._get("/api/tools/worked-shifts-summary", parameters) def summarize_worked_shifts_json( self, shifts: Sequence[Mapping[str, Any]], *, regular_minutes_limit: Optional[int] = None, hourly_rate: Optional[float] = None, overtime_multiplier: Optional[float] = None, currency: Optional[str] = None, ) -> Any: """POST structured shifts as JSON for larger server-side batches.""" body = { key: value for key, value in { "shifts": list(shifts), "regularMinutesLimit": regular_minutes_limit, "hourlyRate": hourly_rate, "overtimeMultiplier": overtime_multiplier, "currency": currency, }.items() if value is not None } return self._post("/api/tools/worked-shifts-summary", body) def export_worked_shifts_calendar( self, shifts: Sequence[Mapping[str, Any]], ) -> str: """Download RFC 5545 calendar text from RFC 3339 shift instants.""" result = self._post( "/api/tools/worked-shifts-calendar.ics", {"shifts": list(shifts)}, accept="text/calendar", ) if not isinstance(result, str): raise TypeError("TotemClock API returned a non-text calendar response") return result def export_timesheet_csv( self, entries: Sequence[str], *, regular_minutes_limit: Optional[int] = None, hourly_rate: Optional[float] = None, overtime_multiplier: Optional[float] = None, currency: Optional[str] = None, ) -> str: """Download the caller-defined timesheet summary as CSV text.""" parameters: list[Parameter] = [("entry", entry) for entry in entries] parameters.extend( [ ("regularMinutesLimit", regular_minutes_limit), ("hourlyRate", hourly_rate), ("overtimeMultiplier", overtime_multiplier), ("currency", currency), ] ) result = self._get( "/api/tools/timesheet-summary/csv", parameters, accept="text/csv", ) if not isinstance(result, str): raise TypeError("TotemClock API returned a non-text CSV response") return result time_tools = TotemClockClient()