"""Booking model. Sprint 14: Import bookings from WordPress kurs-booking plugin. """ import contextlib from datetime import UTC, datetime from decimal import Decimal from typing import Any from sqlalchemy import ( Column, Date, DateTime, ForeignKey, Index, Integer, Numeric, String, ) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import relationship from customer_portal.models import Base class Booking(Base): """Booking synchronized from WordPress kurs-booking. Each booking is linked to a customer via email address matching. The wp_booking_id is the unique identifier from WordPress. """ __tablename__ = "bookings" id = Column(Integer, primary_key=True) customer_id = Column( Integer, ForeignKey("customers.id", ondelete="CASCADE"), nullable=False ) # WordPress identifiers wp_booking_id = Column(Integer, unique=True, nullable=False, index=True) wp_kurs_id = Column(Integer, index=True) # Booking data booking_number = Column(String(50), index=True) kurs_title = Column(String(255)) kurs_date = Column(Date) kurs_time = Column(String(10)) kurs_end_time = Column(String(10)) kurs_location = Column(String(255)) # Status & pricing status = Column(String(50), default="pending", index=True) total_price = Column(Numeric(10, 2)) ticket_type = Column(String(100)) ticket_count = Column(Integer, default=1) # Customer data snapshot (at time of booking) customer_name = Column(String(255)) customer_email = Column(String(255)) customer_phone = Column(String(100)) # sevDesk integration sevdesk_invoice_id = Column(Integer) sevdesk_invoice_number = Column(String(50)) # Custom fields from booking (JSON) custom_fields = Column(JSONB, default=dict) extra_fields = Column(JSONB, default=dict) # Timestamps wp_created_at = Column(DateTime) # Original WordPress creation date synced_at = Column(DateTime) # Last sync timestamp created_at = Column(DateTime, default=lambda: datetime.now(UTC)) updated_at = Column( DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC), ) # Relationship customer = relationship("Customer", back_populates="bookings") # Indexes __table_args__ = ( Index("ix_bookings_customer_status", customer_id, status), Index("ix_bookings_kurs_date", kurs_date), ) def __repr__(self) -> str: return f"" @classmethod def get_by_wp_id(cls, db, wp_booking_id: int): """Get booking by WordPress ID. Args: db: Database session wp_booking_id: WordPress booking post ID Returns: Booking instance or None """ return db.query(cls).filter(cls.wp_booking_id == wp_booking_id).first() @classmethod def get_by_customer(cls, db, customer_id: int, status: str | None = None): """Get all bookings for a customer. Args: db: Database session customer_id: Customer ID status: Optional status filter Returns: List of Booking instances """ query = db.query(cls).filter(cls.customer_id == customer_id) if status: query = query.filter(cls.status == status) return query.order_by(cls.kurs_date.desc()).all() @classmethod def create_or_update_from_wp( cls, db, customer_id: int, wp_data: dict ) -> tuple["Booking", bool]: """Create or update booking from WordPress API data. Args: db: Database session customer_id: Customer ID to link booking to wp_data: Data from WordPress REST API Returns: Tuple of (Booking instance, is_new) """ wp_booking_id = wp_data.get("id") if not wp_booking_id: raise ValueError("WordPress booking ID required") existing = cls.get_by_wp_id(db, wp_booking_id) is_new = existing is None if is_new: booking = cls( customer_id=customer_id, wp_booking_id=wp_booking_id, created_at=datetime.now(UTC), ) db.add(booking) else: booking = existing # Update fields from WordPress data booking.wp_kurs_id = wp_data.get("kurs_id") booking.booking_number = wp_data.get("number", "") booking.kurs_title = wp_data.get("kurs_title", "") booking.kurs_location = wp_data.get("kurs_location", "") booking.status = wp_data.get("status", "pending") booking.ticket_type = wp_data.get("ticket_type", "") booking.ticket_count = wp_data.get("ticket_count", 1) # Parse date kurs_date = wp_data.get("kurs_date") if kurs_date: with contextlib.suppress(ValueError): booking.kurs_date = datetime.strptime(kurs_date, "%Y-%m-%d").date() # Parse times booking.kurs_time = wp_data.get("kurs_time", "") booking.kurs_end_time = wp_data.get("kurs_end_time", "") # Parse price price = wp_data.get("price") if price is not None: booking.total_price = Decimal(str(price)) # Customer snapshot customer_data = wp_data.get("customer", {}) if customer_data: booking.customer_name = customer_data.get("name", "") booking.customer_email = customer_data.get("email", "") booking.customer_phone = customer_data.get("phone", "") # sevDesk sevdesk = wp_data.get("sevdesk", {}) if sevdesk: if sevdesk.get("invoice_id"): booking.sevdesk_invoice_id = int(sevdesk["invoice_id"]) booking.sevdesk_invoice_number = sevdesk.get("invoice_number", "") # Custom fields if wp_data.get("custom_fields"): booking.custom_fields = wp_data["custom_fields"] if wp_data.get("extra_fields"): booking.extra_fields = wp_data["extra_fields"] # Parse WordPress created date wp_created = wp_data.get("created_at") if wp_created: with contextlib.suppress(ValueError): booking.wp_created_at = datetime.strptime( wp_created, "%Y-%m-%d %H:%M:%S" ) # Update sync timestamp booking.synced_at = datetime.now(UTC) booking.updated_at = datetime.now(UTC) return booking, is_new @property def status_display(self) -> str: """Get human-readable status.""" status_map = { "pending": "Ausstehend", "confirmed": "Bestaetigt", "cancelled": "Storniert", "cancel_requested": "Stornierung angefragt", } return status_map.get(self.status, self.status or "Unbekannt") @property def status_color(self) -> str: """Get Bootstrap color class for status.""" color_map = { "pending": "warning", "confirmed": "success", "cancelled": "danger", "cancel_requested": "info", } return color_map.get(self.status, "secondary") @property def formatted_price(self) -> str: """Get formatted price string.""" if self.total_price is None: return "-" return f"{self.total_price:.2f} EUR" @property def formatted_date(self) -> str: """Get formatted date string.""" if not self.kurs_date: return "-" return self.kurs_date.strftime("%d.%m.%Y") @property def formatted_time(self) -> str: """Get formatted time range.""" if not self.kurs_time: return "-" if self.kurs_end_time: return f"{self.kurs_time} - {self.kurs_end_time}" return self.kurs_time def get_custom_field(self, key: str, default: Any = None) -> Any: """Get a custom field value.""" if not self.custom_fields: return default return self.custom_fields.get(key, default)