"""The Kalendarz tab's own events (AV-038, pakiet 2026-10-03, zgłoszenie 3): lodging ("Nocleg") and "other / note" ("Inne") on migration 011's journey_calendar_events table. Like travel legs (travel_legs.py), an event carries no derived analysis and is not part of the append-only RouteRevision/JourneyPlan lineage: a plain mutable row, edited and deleted in place through this one repository. Transport is the existing travel legs; rest days and Day 1 are the plan's (plan_persistence.py: set_rest_days, set_day_one); the trip's own days are read-only events derived from the plan. Times are the Journey's wall clock ("YYYY-MM-DDTHH:MM", or a bare date for a whole-day event), compared as text like travel legs.""" from __future__ import annotations import re import uuid from dataclasses import dataclass from typing import Any from .persistence import JourneySqliteDatabase, _now, _transaction, _validate_identifier from .travel_legs import _optional_text CALENDAR_EVENT_TYPES = ("LODGING", "OTHER") _LOCAL = re.compile(r"\d{4}-\d{2}-\d{2}(T\d{2}:\d{2})?") _MONTH = re.compile(r"(\d{4})-(\d{2})") @dataclass(frozen=True) class CalendarEventFields: """The fields a caller may set -- shared by create and update so both validate identically.""" event_type: str title: str start_local: str end_local: str description: str | None = None place_name: str | None = None location: str | None = None cost_amount: float | None = None cost_currency: str = "PLN" def __post_init__(self) -> None: if self.event_type not in CALENDAR_EVENT_TYPES: raise ValueError(f"event_type must be one of {CALENDAR_EVENT_TYPES}") if not self.title or not self.title.strip(): raise ValueError("title must be non-empty") for name in ("start_local", "end_local"): value = getattr(self, name) if not isinstance(value, str) or not _LOCAL.fullmatch(value): raise ValueError(f"{name} must be YYYY-MM-DD or YYYY-MM-DDTHH:MM") if self.end_local < self.start_local: raise ValueError("end_local must not be before start_local") if self.cost_amount is not None and ( isinstance(self.cost_amount, bool) or self.cost_amount < 0 ): raise ValueError("cost_amount must be non-negative") if not self.cost_currency or not self.cost_currency.strip(): raise ValueError("cost_currency must be non-empty") @dataclass(frozen=True) class CalendarEvent: event_id: str journey_id: str fields: CalendarEventFields created_at: str updated_at: str def to_dict(self) -> dict[str, Any]: f = self.fields return { "event_id": self.event_id, "journey_id": self.journey_id, "event_type": f.event_type, "title": f.title, "description": f.description, "place_name": f.place_name, "location": f.location, "start_local": f.start_local, "end_local": f.end_local, "cost_amount": f.cost_amount, "cost_currency": f.cost_currency, "created_at": self.created_at, "updated_at": self.updated_at, } def overlaps_month(start_local: str, end_local: str, month: str) -> bool: """Whether an event touches calendar month ``YYYY-MM`` -- an event that starts in one month and ends in the next belongs to both.""" match = _MONTH.fullmatch(month) if not match: raise ValueError("month must be YYYY-MM") year, number = int(match[1]), int(match[2]) if not 1 <= number <= 12: raise ValueError("month must be YYYY-MM") first = f"{year:04d}-{number:02d}-01" following = f"{year + (number == 12):04d}-{number % 12 + 1:02d}-01" return start_local[:10] < following and end_local[:10] >= first def _event_from_row(row) -> CalendarEvent: return CalendarEvent( event_id=row["event_id"], journey_id=row["journey_id"], fields=CalendarEventFields( event_type=row["event_type"], title=row["title"], start_local=row["start_local"], end_local=row["end_local"], description=row["description"], place_name=row["place_name"], location=row["location"], cost_amount=row["cost_amount"], cost_currency=row["cost_currency"], ), created_at=row["created_at"], updated_at=row["updated_at"], ) class CalendarEventRepository: def __init__(self, database: JourneySqliteDatabase): self._connection = database.connect() def close(self) -> None: self._connection.close() def _require_journey(self, journey_id: str, owner_id: str) -> None: row = self._connection.execute( "SELECT 1 FROM journeys WHERE journey_id = ? AND owner_id = ?", (journey_id, owner_id), ).fetchone() if row is None: raise KeyError(journey_id) def list_for_journey( self, *, journey_id: str, owner_id: str, month: str | None = None ) -> tuple[CalendarEvent, ...]: self._require_journey(journey_id, owner_id) rows = self._connection.execute( "SELECT * FROM journey_calendar_events WHERE journey_id = ? ORDER BY start_local, created_at", (journey_id,), ).fetchall() events = tuple(_event_from_row(row) for row in rows) if month is None: return events return tuple( e for e in events if overlaps_month(e.fields.start_local, e.fields.end_local, month) ) def create( self, *, journey_id: str, owner_id: str, fields: CalendarEventFields ) -> CalendarEvent: _validate_identifier(journey_id, "journey") self._require_journey(journey_id, owner_id) now = _now() event = CalendarEvent( event_id=f"journey-calendar-event-v0_1:{uuid.uuid4()}", journey_id=journey_id, fields=fields, created_at=now, updated_at=now, ) with _transaction(self._connection): self._connection.execute( """INSERT INTO journey_calendar_events( event_id, journey_id, event_type, title, description, place_name, location, start_local, end_local, cost_amount, cost_currency, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", (event.event_id, journey_id, *self._values(fields), now, now), ) return event def update( self, *, event_id: str, journey_id: str, owner_id: str, fields: CalendarEventFields ) -> CalendarEvent: self._require_journey(journey_id, owner_id) with _transaction(self._connection): cursor = self._connection.execute( """UPDATE journey_calendar_events SET event_type = ?, title = ?, description = ?, place_name = ?, location = ?, start_local = ?, end_local = ?, cost_amount = ?, cost_currency = ?, updated_at = ? WHERE event_id = ? AND journey_id = ?""", (*self._values(fields), _now(), event_id, journey_id), ) if cursor.rowcount != 1: raise KeyError(event_id) row = self._connection.execute( "SELECT * FROM journey_calendar_events WHERE event_id = ?", (event_id,) ).fetchone() return _event_from_row(row) def delete(self, *, event_id: str, journey_id: str, owner_id: str) -> None: self._require_journey(journey_id, owner_id) with _transaction(self._connection): cursor = self._connection.execute( "DELETE FROM journey_calendar_events WHERE event_id = ? AND journey_id = ?", (event_id, journey_id), ) if cursor.rowcount != 1: raise KeyError(event_id) @staticmethod def _values(fields: CalendarEventFields) -> tuple: return ( fields.event_type, fields.title.strip(), fields.description, fields.place_name, fields.location, fields.start_local, fields.end_local, fields.cost_amount, fields.cost_currency.strip(), ) def calendar_event_fields_from_payload(payload: dict[str, Any]) -> CalendarEventFields: cost = payload.get("cost_amount") return CalendarEventFields( event_type=payload["event_type"], title=str(payload.get("title") or ""), start_local=payload["start_local"], end_local=payload.get("end_local") or payload["start_local"], description=_optional_text(payload.get("description")), place_name=_optional_text(payload.get("place_name")), location=_optional_text(payload.get("location")), cost_amount=None if cost in (None, "") else float(cost), cost_currency=str(payload.get("cost_currency") or "PLN"), )