1052 lines
38 KiB
Python
1052 lines
38 KiB
Python
import logging
|
|
import traceback
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from alpine_bits_python.db import Customer, Reservation
|
|
from alpine_bits_python.schemas import (
|
|
CommentData,
|
|
CommentListItemData,
|
|
CommentsData,
|
|
CustomerData,
|
|
HotelReservationIdData,
|
|
PhoneTechType,
|
|
)
|
|
|
|
# Import the generated classes
|
|
from .generated.alpinebits import (
|
|
CommentName2,
|
|
HotelReservationResStatus,
|
|
OtaHotelResNotifRq,
|
|
OtaResRetrieveRs,
|
|
ProfileProfileType,
|
|
UniqueIdType2,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
_LOGGER.setLevel(logging.INFO)
|
|
|
|
# Define type aliases for the two Customer types
|
|
NotifCustomer = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGuests.ResGuest.Profiles.ProfileInfo.Profile.Customer # noqa: E501
|
|
RetrieveCustomer = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGuests.ResGuest.Profiles.ProfileInfo.Profile.Customer # noqa: E501
|
|
|
|
# Define type aliases for HotelReservationId types
|
|
NotifHotelReservationId = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.HotelReservationIds.HotelReservationId # noqa: E501
|
|
RetrieveHotelReservationId = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.HotelReservationIds.HotelReservationId # noqa: E501
|
|
|
|
# Define type aliases for Comments types
|
|
NotifComments = (
|
|
OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.Comments
|
|
)
|
|
RetrieveComments = (
|
|
OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.Comments
|
|
)
|
|
NotifComment = (
|
|
OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.Comments.Comment
|
|
)
|
|
RetrieveComment = (
|
|
OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.Comments.Comment
|
|
)
|
|
|
|
# type aliases for GuestCounts
|
|
NotifGuestCounts = (
|
|
OtaHotelResNotifRq.HotelReservations.HotelReservation.RoomStays.RoomStay.GuestCounts
|
|
)
|
|
RetrieveGuestCounts = (
|
|
OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay.GuestCounts
|
|
)
|
|
|
|
NotifUniqueId = OtaHotelResNotifRq.HotelReservations.HotelReservation.UniqueId
|
|
RetrieveUniqueId = OtaResRetrieveRs.ReservationsList.HotelReservation.UniqueId
|
|
|
|
NotifTimeSpan = (
|
|
OtaHotelResNotifRq.HotelReservations.HotelReservation.RoomStays.RoomStay.TimeSpan
|
|
)
|
|
RetrieveTimeSpan = (
|
|
OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay.TimeSpan
|
|
)
|
|
|
|
NotifRoomStays = OtaHotelResNotifRq.HotelReservations.HotelReservation.RoomStays
|
|
RetrieveRoomStays = OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays
|
|
|
|
NotifHotelReservation = OtaHotelResNotifRq.HotelReservations.HotelReservation
|
|
RetrieveHotelReservation = OtaResRetrieveRs.ReservationsList.HotelReservation
|
|
|
|
|
|
# Enum to specify which OTA message type to use
|
|
class OtaMessageType(Enum):
|
|
NOTIF = "notification" # For OtaHotelResNotifRq
|
|
RETRIEVE = "retrieve" # For OtaResRetrieveRs
|
|
|
|
|
|
@dataclass
|
|
class KidsAgeData:
|
|
"""Data class to hold information about children's ages."""
|
|
|
|
ages: list[int]
|
|
|
|
|
|
class GuestCountsFactory:
|
|
"""Factory class to create GuestCounts instances for both OtaHotelResNotifRq and OtaResRetrieveRs."""
|
|
|
|
@staticmethod
|
|
def create_guest_counts(
|
|
adults: int,
|
|
kids: list[int] | None = None,
|
|
message_type: OtaMessageType = OtaMessageType.RETRIEVE,
|
|
) -> NotifGuestCounts:
|
|
"""Create a GuestCounts object for OtaHotelResNotifRq or OtaResRetrieveRs.
|
|
|
|
:param adults: Number of adults
|
|
:param kids: List of ages for each kid (optional)
|
|
:return: GuestCounts instance
|
|
"""
|
|
if message_type == OtaMessageType.RETRIEVE:
|
|
return GuestCountsFactory._create_guest_counts(
|
|
adults, kids, RetrieveGuestCounts
|
|
)
|
|
if message_type == OtaMessageType.NOTIF:
|
|
return GuestCountsFactory._create_guest_counts(
|
|
adults, kids, NotifGuestCounts
|
|
)
|
|
raise ValueError(f"Unsupported message type: {message_type}")
|
|
|
|
@staticmethod
|
|
def _create_guest_counts(
|
|
adults: int, kids: list[int] | None, guest_counts_class: type
|
|
) -> Any:
|
|
"""Create a GuestCounts object of the specified type.
|
|
|
|
:param adults: Number of adults
|
|
:param kids: List of ages for each kid (optional)
|
|
:param guest_counts_class: The GuestCounts class to instantiate
|
|
:return: GuestCounts instance
|
|
"""
|
|
GuestCount = guest_counts_class.GuestCount
|
|
guest_count_list = []
|
|
if adults > 0:
|
|
guest_count_list.append(GuestCount(count=str(adults)))
|
|
if kids:
|
|
# create a dict with amount of kids for each age
|
|
age_count = {}
|
|
|
|
for age in kids:
|
|
if age in age_count:
|
|
age_count[age] += 1
|
|
else:
|
|
age_count[age] = 1
|
|
|
|
for age, count in age_count.items():
|
|
guest_count_list.append(GuestCount(count=str(count), age=str(age)))
|
|
return guest_counts_class(guest_count=guest_count_list)
|
|
|
|
|
|
class CustomerFactory:
|
|
"""Factory class to create Customer instances for both Retrieve and Notif."""
|
|
|
|
@staticmethod
|
|
def create_notif_customer(data: CustomerData) -> NotifCustomer:
|
|
"""Create a Customer for OtaHotelResNotifRq."""
|
|
return CustomerFactory._create_customer(NotifCustomer, data)
|
|
|
|
@staticmethod
|
|
def create_retrieve_customer(data: CustomerData) -> RetrieveCustomer:
|
|
"""Create a Customer for OtaResRetrieveRs."""
|
|
return CustomerFactory._create_customer(RetrieveCustomer, data)
|
|
|
|
@staticmethod
|
|
def _create_customer(
|
|
customer_class: type[RetrieveCustomer | NotifCustomer], data: CustomerData
|
|
) -> Any:
|
|
"""Create a customer of the specified type."""
|
|
# Create PersonName
|
|
person_name = customer_class.PersonName(
|
|
given_name=data.given_name,
|
|
surname=data.surname,
|
|
name_prefix=data.name_prefix,
|
|
name_title=data.name_title,
|
|
)
|
|
|
|
# Create telephone list
|
|
telephones = []
|
|
for phone_number, phone_tech_type in data.phone_numbers:
|
|
telephone = customer_class.Telephone(
|
|
phone_number=phone_number,
|
|
phone_tech_type=phone_tech_type.value if phone_tech_type else None,
|
|
)
|
|
telephones.append(telephone)
|
|
|
|
# Create email if provided
|
|
email = None
|
|
if data.email_address:
|
|
remark = None
|
|
if data.email_newsletter is not None:
|
|
remark = f"newsletter:{'yes' if data.email_newsletter else 'no'}"
|
|
|
|
email = customer_class.Email(value=data.email_address, remark=remark)
|
|
|
|
# Create address if any address fields are provided
|
|
address = None
|
|
if any(
|
|
[data.address_line, data.city_name, data.postal_code, data.country_code]
|
|
):
|
|
country_name = None
|
|
if data.country_code:
|
|
country_name = customer_class.Address.CountryName(
|
|
code=data.country_code
|
|
)
|
|
|
|
address_remark = None
|
|
if data.address_catalog is not None:
|
|
address_remark = f"catalog:{'yes' if data.address_catalog else 'no'}"
|
|
|
|
address = customer_class.Address(
|
|
address_line=data.address_line,
|
|
city_name=data.city_name,
|
|
postal_code=data.postal_code,
|
|
country_name=country_name,
|
|
remark=address_remark,
|
|
)
|
|
|
|
# Create the customer
|
|
return customer_class(
|
|
person_name=person_name,
|
|
telephone=telephones,
|
|
email=email,
|
|
address=address,
|
|
gender=data.gender,
|
|
birth_date=data.birth_date,
|
|
language=data.language,
|
|
)
|
|
|
|
@staticmethod
|
|
def from_notif_customer(customer: NotifCustomer) -> CustomerData:
|
|
"""Convert a NotifCustomer back to CustomerData."""
|
|
return CustomerFactory._customer_to_data(customer)
|
|
|
|
@staticmethod
|
|
def from_retrieve_customer(customer: RetrieveCustomer) -> CustomerData:
|
|
"""Convert a RetrieveCustomer back to CustomerData."""
|
|
return CustomerFactory._customer_to_data(customer)
|
|
|
|
@staticmethod
|
|
def _customer_to_data(customer: Any) -> CustomerData:
|
|
"""Convert any customer type to CustomerData."""
|
|
# Extract phone numbers
|
|
phone_numbers = []
|
|
if customer.telephone:
|
|
phone_numbers.extend(
|
|
[
|
|
(
|
|
tel.phone_number,
|
|
PhoneTechType(tel.phone_tech_type)
|
|
if tel.phone_tech_type
|
|
else None,
|
|
)
|
|
for tel in customer.telephone
|
|
]
|
|
)
|
|
|
|
# Extract email info
|
|
email_address = None
|
|
email_newsletter = None
|
|
if customer.email:
|
|
email_address = customer.email.value
|
|
if customer.email.remark:
|
|
if "newsletter:yes" in customer.email.remark:
|
|
email_newsletter = True
|
|
elif "newsletter:no" in customer.email.remark:
|
|
email_newsletter = False
|
|
|
|
# Extract address info
|
|
address_line = None
|
|
city_name = None
|
|
postal_code = None
|
|
country_code = None
|
|
address_catalog = None
|
|
|
|
if customer.address:
|
|
address_line = customer.address.address_line
|
|
city_name = customer.address.city_name
|
|
postal_code = customer.address.postal_code
|
|
|
|
if customer.address.country_name:
|
|
country_code = customer.address.country_name.code
|
|
|
|
if customer.address.remark:
|
|
if "catalog:yes" in customer.address.remark:
|
|
address_catalog = True
|
|
elif "catalog:no" in customer.address.remark:
|
|
address_catalog = False
|
|
|
|
return CustomerData(
|
|
given_name=customer.person_name.given_name,
|
|
surname=customer.person_name.surname,
|
|
name_prefix=customer.person_name.name_prefix,
|
|
name_title=customer.person_name.name_title,
|
|
phone_numbers=phone_numbers,
|
|
email_address=email_address,
|
|
email_newsletter=email_newsletter,
|
|
address_line=address_line,
|
|
city_name=city_name,
|
|
postal_code=postal_code,
|
|
country_code=country_code,
|
|
address_catalog=address_catalog,
|
|
gender=customer.gender,
|
|
birth_date=customer.birth_date,
|
|
language=customer.language,
|
|
)
|
|
|
|
|
|
class HotelReservationIdFactory:
|
|
"""Factory class to create HotelReservationId instances for both OtaHotelResNotifRq and OtaResRetrieveRs."""
|
|
|
|
@staticmethod
|
|
def create_notif_hotel_reservation_id(
|
|
data: HotelReservationIdData,
|
|
) -> NotifHotelReservationId:
|
|
"""Create a HotelReservationId for OtaHotelResNotifRq."""
|
|
return HotelReservationIdFactory._create_hotel_reservation_id(
|
|
NotifHotelReservationId, data
|
|
)
|
|
|
|
@staticmethod
|
|
def create_retrieve_hotel_reservation_id(
|
|
data: HotelReservationIdData,
|
|
) -> RetrieveHotelReservationId:
|
|
"""Create a HotelReservationId for OtaResRetrieveRs."""
|
|
return HotelReservationIdFactory._create_hotel_reservation_id(
|
|
RetrieveHotelReservationId, data
|
|
)
|
|
|
|
@staticmethod
|
|
def _create_hotel_reservation_id(
|
|
hotel_reservation_id_class: type, data: HotelReservationIdData
|
|
) -> Any:
|
|
"""Create a hotel reservation id of the specified type."""
|
|
return hotel_reservation_id_class(
|
|
res_id_type=data.res_id_type,
|
|
res_id_value=data.res_id_value,
|
|
res_id_source=data.res_id_source,
|
|
res_id_source_context=data.res_id_source_context,
|
|
)
|
|
|
|
@staticmethod
|
|
def from_notif_hotel_reservation_id(
|
|
hotel_reservation_id: NotifHotelReservationId,
|
|
) -> HotelReservationIdData:
|
|
"""Convert a NotifHotelReservationId back to HotelReservationIdData."""
|
|
return HotelReservationIdFactory._hotel_reservation_id_to_data(
|
|
hotel_reservation_id
|
|
)
|
|
|
|
@staticmethod
|
|
def from_retrieve_hotel_reservation_id(
|
|
hotel_reservation_id: RetrieveHotelReservationId,
|
|
) -> HotelReservationIdData:
|
|
"""Convert a RetrieveHotelReservationId back to HotelReservationIdData."""
|
|
return HotelReservationIdFactory._hotel_reservation_id_to_data(
|
|
hotel_reservation_id
|
|
)
|
|
|
|
@staticmethod
|
|
def _hotel_reservation_id_to_data(
|
|
hotel_reservation_id: Any,
|
|
) -> HotelReservationIdData:
|
|
"""Internal method to convert any hotel reservation id type to HotelReservationIdData."""
|
|
return HotelReservationIdData(
|
|
res_id_type=hotel_reservation_id.res_id_type,
|
|
res_id_value=hotel_reservation_id.res_id_value,
|
|
res_id_source=hotel_reservation_id.res_id_source,
|
|
res_id_source_context=hotel_reservation_id.res_id_source_context,
|
|
)
|
|
|
|
|
|
class CommentFactory:
|
|
"""Factory class to create Comment instances for both OtaHotelResNotifRq and OtaResRetrieveRs."""
|
|
|
|
@staticmethod
|
|
def create_notif_comments(data: CommentsData) -> NotifComments:
|
|
"""Create Comments for OtaHotelResNotifRq."""
|
|
return CommentFactory._create_comments(NotifComments, NotifComment, data)
|
|
|
|
@staticmethod
|
|
def create_retrieve_comments(data: CommentsData) -> RetrieveComments:
|
|
"""Create Comments for OtaResRetrieveRs."""
|
|
return CommentFactory._create_comments(RetrieveComments, RetrieveComment, data)
|
|
|
|
@staticmethod
|
|
def _create_comments(
|
|
comments_class: type[RetrieveComments] | type[NotifComments],
|
|
comment_class: type[RetrieveComment] | type[NotifComment],
|
|
data: CommentsData,
|
|
) -> Any:
|
|
"""Internal method to create comments of the specified type."""
|
|
comments_list = []
|
|
for comment_data in data.comments:
|
|
# Create list items
|
|
list_items = []
|
|
for item_data in comment_data.list_items:
|
|
_LOGGER.info(
|
|
f"Creating list item: value={item_data.value}, list_item={item_data.list_item}, language={item_data.language}"
|
|
)
|
|
|
|
list_item = comment_class.ListItem(
|
|
value=item_data.value,
|
|
list_item=item_data.list_item,
|
|
language=item_data.language,
|
|
)
|
|
list_items.append(list_item)
|
|
|
|
# Create comment
|
|
comment = comment_class(
|
|
name=comment_data.name, text=comment_data.text, list_item=list_items
|
|
)
|
|
comments_list.append(comment)
|
|
|
|
# Create comments container
|
|
return comments_class(comment=comments_list)
|
|
|
|
@staticmethod
|
|
def from_notif_comments(comments: NotifComments) -> CommentsData:
|
|
"""Convert NotifComments back to CommentsData."""
|
|
return CommentFactory._comments_to_data(comments)
|
|
|
|
@staticmethod
|
|
def from_retrieve_comments(comments: RetrieveComments) -> CommentsData:
|
|
"""Convert RetrieveComments back to CommentsData."""
|
|
return CommentFactory._comments_to_data(comments)
|
|
|
|
@staticmethod
|
|
def _comments_to_data(comments: Any) -> CommentsData:
|
|
"""Internal method to convert any comments type to CommentsData."""
|
|
comments_data_list = []
|
|
for comment in comments.comment:
|
|
# Extract list items
|
|
list_items_data = []
|
|
if comment.list_item:
|
|
for list_item in comment.list_item:
|
|
list_items_data.append(
|
|
CommentListItemData(
|
|
value=list_item.value,
|
|
list_item=list_item.list_item,
|
|
language=list_item.language,
|
|
)
|
|
)
|
|
|
|
comments_data_list.append(comment)
|
|
|
|
return CommentsData(comments=comments_data_list)
|
|
|
|
|
|
# Define type aliases for ResGuests types
|
|
NotifResGuests = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGuests
|
|
RetrieveResGuests = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGuests
|
|
|
|
|
|
class ResGuestFactory:
|
|
"""Factory class to create complete ResGuests structures with a primary customer."""
|
|
|
|
@staticmethod
|
|
def create_notif_res_guests(customer_data: CustomerData) -> NotifResGuests:
|
|
"""Create a complete ResGuests structure for OtaHotelResNotifRq with primary customer."""
|
|
return ResGuestFactory._create_res_guests(
|
|
NotifResGuests, NotifCustomer, customer_data
|
|
)
|
|
|
|
@staticmethod
|
|
def create_retrieve_res_guests(customer_data: CustomerData) -> RetrieveResGuests:
|
|
"""Create a complete ResGuests structure for OtaResRetrieveRs with primary customer."""
|
|
return ResGuestFactory._create_res_guests(
|
|
RetrieveResGuests, RetrieveCustomer, customer_data
|
|
)
|
|
|
|
@staticmethod
|
|
def _create_res_guests(
|
|
res_guests_class: type[RetrieveResGuests] | type[NotifResGuests],
|
|
customer_class: type[NotifCustomer | RetrieveCustomer],
|
|
customer_data: CustomerData,
|
|
) -> Any:
|
|
"""Create the complete ResGuests structure."""
|
|
# Create the customer using the existing CustomerFactory
|
|
customer = CustomerFactory._create_customer(customer_class, customer_data)
|
|
|
|
# Create Profile with the customer
|
|
profile = res_guests_class.ResGuest.Profiles.ProfileInfo.Profile(
|
|
customer=customer
|
|
)
|
|
|
|
# Create ProfileInfo with the profile
|
|
profile_info = res_guests_class.ResGuest.Profiles.ProfileInfo(profile=profile)
|
|
|
|
# Create Profiles with the profile_info
|
|
profiles = res_guests_class.ResGuest.Profiles(profile_info=profile_info)
|
|
|
|
# Create ResGuest with the profiles
|
|
res_guest = res_guests_class.ResGuest(profiles=profiles)
|
|
|
|
# Create ResGuests with the res_guest
|
|
return res_guests_class(res_guest=res_guest)
|
|
|
|
@staticmethod
|
|
def extract_primary_customer(
|
|
res_guests: NotifResGuests | RetrieveResGuests,
|
|
) -> CustomerData:
|
|
"""Extract the primary customer data from a ResGuests structure."""
|
|
# Navigate down the nested structure to get the customer
|
|
customer = res_guests.res_guest.profiles.profile_info.profile.customer
|
|
|
|
# Use the existing CustomerFactory conversion method
|
|
if isinstance(res_guests, NotifResGuests):
|
|
return CustomerFactory.from_notif_customer(customer)
|
|
return CustomerFactory.from_retrieve_customer(customer)
|
|
|
|
|
|
class AlpineBitsFactory:
|
|
"""Unified factory class for creating AlpineBits objects with a simple interface."""
|
|
|
|
@staticmethod
|
|
def create(
|
|
data: CustomerData | HotelReservationIdData | CommentsData,
|
|
message_type: OtaMessageType,
|
|
) -> Any:
|
|
"""Create an AlpineBits object based on the data type and message type.
|
|
|
|
Args:
|
|
data: The data object (CustomerData, HotelReservationIdData, CommentsData, etc.)
|
|
message_type: Whether to create for NOTIF or RETRIEVE message types
|
|
|
|
Returns:
|
|
The appropriate AlpineBits object based on the data type and message type
|
|
|
|
"""
|
|
if isinstance(data, CustomerData):
|
|
if message_type == OtaMessageType.NOTIF:
|
|
return CustomerFactory.create_notif_customer(data)
|
|
return CustomerFactory.create_retrieve_customer(data)
|
|
|
|
if isinstance(data, HotelReservationIdData):
|
|
if message_type == OtaMessageType.NOTIF:
|
|
return HotelReservationIdFactory.create_notif_hotel_reservation_id(data)
|
|
return HotelReservationIdFactory.create_retrieve_hotel_reservation_id(data)
|
|
|
|
if isinstance(data, CommentsData):
|
|
if message_type == OtaMessageType.NOTIF:
|
|
return CommentFactory.create_notif_comments(data)
|
|
return CommentFactory.create_retrieve_comments(data)
|
|
|
|
raise ValueError(f"Unsupported data type: {type(data)}")
|
|
|
|
@staticmethod
|
|
def create_res_guests(
|
|
customer_data: CustomerData, message_type: OtaMessageType
|
|
) -> NotifResGuests | RetrieveResGuests:
|
|
"""Create a complete ResGuests structure with a primary customer.
|
|
|
|
Args:
|
|
customer_data: The customer data
|
|
message_type: Whether to create for NOTIF or RETRIEVE message types
|
|
|
|
Returns:
|
|
The appropriate ResGuests object
|
|
|
|
"""
|
|
if message_type == OtaMessageType.NOTIF:
|
|
return ResGuestFactory.create_notif_res_guests(customer_data)
|
|
return ResGuestFactory.create_retrieve_res_guests(customer_data)
|
|
|
|
@staticmethod
|
|
def extract_data(
|
|
obj: Any,
|
|
) -> CustomerData | HotelReservationIdData | CommentsData:
|
|
"""Extract data from an AlpineBits object back to a simple data class.
|
|
|
|
Args:
|
|
obj: The AlpineBits object to extract data from
|
|
|
|
Returns:
|
|
The appropriate data object
|
|
|
|
"""
|
|
# Check if it's a Customer object
|
|
if hasattr(obj, "person_name") and hasattr(obj.person_name, "given_name"):
|
|
if isinstance(obj, NotifCustomer):
|
|
return CustomerFactory.from_notif_customer(obj)
|
|
if isinstance(obj, RetrieveCustomer):
|
|
return CustomerFactory.from_retrieve_customer(obj)
|
|
|
|
# Check if it's a HotelReservationId object
|
|
elif hasattr(obj, "res_id_type"):
|
|
if isinstance(obj, NotifHotelReservationId):
|
|
return HotelReservationIdFactory.from_notif_hotel_reservation_id(obj)
|
|
if isinstance(obj, RetrieveHotelReservationId):
|
|
return HotelReservationIdFactory.from_retrieve_hotel_reservation_id(obj)
|
|
|
|
# Check if it's a Comments object
|
|
elif hasattr(obj, "comment"):
|
|
if isinstance(obj, NotifComments):
|
|
return CommentFactory.from_notif_comments(obj)
|
|
if isinstance(obj, RetrieveComments):
|
|
return CommentFactory.from_retrieve_comments(obj)
|
|
|
|
# Check if it's a ResGuests object
|
|
elif hasattr(obj, "res_guest"):
|
|
return ResGuestFactory.extract_primary_customer(obj)
|
|
|
|
else:
|
|
raise ValueError(f"Unsupported object type: {type(obj)}")
|
|
return None
|
|
|
|
|
|
def create_res_retrieve_response(
|
|
list: list[tuple[Reservation, Customer]],
|
|
) -> OtaResRetrieveRs:
|
|
"""Create RetrievedReservation XML from database entries."""
|
|
return _create_xml_from_db(list, OtaMessageType.RETRIEVE)
|
|
|
|
|
|
def create_res_notif_push_message(list: tuple[Reservation, Customer]):
|
|
"""Create Reservation Notification XML from database entries."""
|
|
return _create_xml_from_db(list, OtaMessageType.NOTIF)
|
|
|
|
|
|
def _process_single_reservation(
|
|
reservation: Reservation, customer: Customer, message_type: OtaMessageType
|
|
):
|
|
phone_numbers = (
|
|
[(customer.phone, PhoneTechType.MOBILE)] if customer.phone is not None else []
|
|
)
|
|
|
|
customer_data = CustomerData(
|
|
given_name=customer.given_name,
|
|
surname=customer.surname,
|
|
name_prefix=customer.name_prefix,
|
|
name_title=customer.name_title,
|
|
phone_numbers=phone_numbers,
|
|
email_address=customer.email_address,
|
|
email_newsletter=customer.email_newsletter,
|
|
address_line=customer.address_line,
|
|
city_name=customer.city_name,
|
|
postal_code=customer.postal_code,
|
|
country_code=customer.country_code,
|
|
address_catalog=customer.address_catalog,
|
|
gender=customer.gender,
|
|
birth_date=customer.birth_date,
|
|
language=customer.language,
|
|
)
|
|
alpine_bits_factory = AlpineBitsFactory()
|
|
res_guests = alpine_bits_factory.create_res_guests(customer_data, message_type)
|
|
|
|
# Guest counts
|
|
children_ages = [int(a) for a in reservation.children_ages.split(",") if a]
|
|
guest_counts = GuestCountsFactory.create_guest_counts(
|
|
reservation.num_adults, children_ages, message_type
|
|
)
|
|
|
|
if message_type == OtaMessageType.NOTIF:
|
|
UniqueId = NotifUniqueId
|
|
RoomStays = NotifRoomStays
|
|
HotelReservation = NotifHotelReservation
|
|
Profile = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.Profiles.ProfileInfo.Profile
|
|
elif message_type == OtaMessageType.RETRIEVE:
|
|
UniqueId = RetrieveUniqueId
|
|
RoomStays = RetrieveRoomStays
|
|
HotelReservation = RetrieveHotelReservation
|
|
Profile = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.Profiles.ProfileInfo.Profile
|
|
else:
|
|
raise ValueError("Unsupported message type: %s", message_type.value)
|
|
|
|
unique_id_str = reservation.md5_unique_id
|
|
|
|
# UniqueID
|
|
unique_id = UniqueId(type_value=UniqueIdType2.VALUE_14, id=unique_id_str)
|
|
|
|
# TimeSpan
|
|
time_span = RoomStays.RoomStay.TimeSpan(
|
|
start=reservation.start_date.isoformat() if reservation.start_date else None,
|
|
end=reservation.end_date.isoformat() if reservation.end_date else None,
|
|
)
|
|
room_stay = RoomStays.RoomStay(
|
|
time_span=time_span,
|
|
guest_counts=guest_counts,
|
|
)
|
|
room_stays = RoomStays(
|
|
room_stay=[room_stay],
|
|
)
|
|
|
|
res_id_source = "website"
|
|
klick_id = None
|
|
|
|
if reservation.fbclid != "":
|
|
klick_id = str(reservation.fbclid)
|
|
res_id_source = "meta"
|
|
elif reservation.gclid != "":
|
|
klick_id = str(reservation.gclid)
|
|
res_id_source = "google"
|
|
|
|
# Get utm_medium if available, otherwise use source
|
|
if reservation.utm_medium is not None and str(reservation.utm_medium) != "":
|
|
res_id_source = str(reservation.utm_medium)
|
|
|
|
# Use Pydantic model for automatic validation and truncation
|
|
# It will automatically:
|
|
# - Trim whitespace
|
|
# - Truncate to 64 characters if needed
|
|
# - Convert empty strings to None
|
|
hotel_res_id_data = HotelReservationIdData(
|
|
res_id_type="13",
|
|
res_id_value=klick_id,
|
|
res_id_source=res_id_source,
|
|
res_id_source_context="99tales",
|
|
)
|
|
|
|
hotel_res_id = alpine_bits_factory.create(hotel_res_id_data, message_type)
|
|
hotel_res_ids = HotelReservation.ResGlobalInfo.HotelReservationIds(
|
|
hotel_reservation_id=[hotel_res_id]
|
|
)
|
|
|
|
if reservation.hotel_code is None:
|
|
raise ValueError("Reservation hotel_code is None")
|
|
hotel_code = str(reservation.hotel_code)
|
|
if reservation.hotel_name is None:
|
|
hotel_name = None
|
|
else:
|
|
hotel_name = str(reservation.hotel_name)
|
|
|
|
basic_property_info = HotelReservation.ResGlobalInfo.BasicPropertyInfo(
|
|
hotel_code=hotel_code,
|
|
hotel_name=hotel_name,
|
|
)
|
|
# Comments
|
|
|
|
offer_comment = None
|
|
if reservation.offer is not None:
|
|
offer_comment = CommentData(
|
|
name=CommentName2.ADDITIONAL_INFO,
|
|
text="Angebot/Offerta: " + reservation.offer,
|
|
# list_items=[
|
|
# CommentListItemData(
|
|
# value=reservation.offer,
|
|
# language=customer.language,
|
|
# list_item="1",
|
|
# )
|
|
# ],
|
|
)
|
|
comment = None
|
|
if reservation.user_comment:
|
|
comment = CommentData(
|
|
name=CommentName2.CUSTOMER_COMMENT,
|
|
text=reservation.user_comment,
|
|
# list_items=[
|
|
# CommentListItemData(
|
|
# value="Landing page comment",
|
|
# language=customer.language,
|
|
# list_item="1",
|
|
# )
|
|
# ],
|
|
)
|
|
comments = [offer_comment, comment]
|
|
|
|
# filter out None comments
|
|
comments = [c for c in comments if c is not None]
|
|
|
|
comments_xml = None
|
|
if comments:
|
|
for c in comments:
|
|
_LOGGER.info(
|
|
f"Creating comment: name={c.name}, text={c.text}, list_items={len(c.list_items)}"
|
|
)
|
|
|
|
comments_data = CommentsData(comments=comments)
|
|
comments_xml = alpine_bits_factory.create(comments_data, message_type)
|
|
|
|
company_name = Profile.CompanyInfo.CompanyName(
|
|
value="99tales GmbH", code="who knows?", code_context="who knows?"
|
|
)
|
|
|
|
company_info = Profile.CompanyInfo(company_name=company_name)
|
|
|
|
profile = Profile(
|
|
company_info=company_info, profile_type=ProfileProfileType.VALUE_4
|
|
)
|
|
|
|
profile_info = HotelReservation.ResGlobalInfo.Profiles.ProfileInfo(profile=profile)
|
|
|
|
_LOGGER.info(f"Type of profile_info: {type(profile_info)}")
|
|
|
|
profiles = HotelReservation.ResGlobalInfo.Profiles(profile_info=profile_info)
|
|
|
|
res_global_info = HotelReservation.ResGlobalInfo(
|
|
hotel_reservation_ids=hotel_res_ids,
|
|
basic_property_info=basic_property_info,
|
|
comments=comments_xml,
|
|
profiles=profiles,
|
|
)
|
|
|
|
hotel_reservation = HotelReservation(
|
|
create_date_time=datetime.now(UTC).isoformat(),
|
|
res_status=HotelReservationResStatus.REQUESTED,
|
|
room_stay_reservation="true",
|
|
unique_id=unique_id,
|
|
room_stays=room_stays,
|
|
res_guests=res_guests,
|
|
res_global_info=res_global_info,
|
|
)
|
|
|
|
return hotel_reservation
|
|
|
|
|
|
def _create_xml_from_db(
|
|
entries: list[tuple[Reservation, Customer]] | tuple[Reservation, Customer],
|
|
type: OtaMessageType,
|
|
):
|
|
"""Create RetrievedReservation XML from database entries.
|
|
|
|
list of pairs (Reservation, Customer)
|
|
"""
|
|
reservations_list = []
|
|
|
|
# if entries isn't a list wrap the element in a list
|
|
|
|
if not isinstance(entries, list):
|
|
entries = [entries]
|
|
|
|
for reservation, customer in entries:
|
|
_LOGGER.info(
|
|
f"Creating XML for reservation {reservation.unique_id} and customer {customer.given_name}"
|
|
)
|
|
|
|
try:
|
|
hotel_reservation = _process_single_reservation(reservation, customer, type)
|
|
|
|
reservations_list.append(hotel_reservation)
|
|
|
|
except Exception as e:
|
|
_LOGGER.error(
|
|
f"Error creating XML for reservation {reservation.unique_id} and customer {customer.given_name}: {e}"
|
|
)
|
|
_LOGGER.debug(traceback.format_exc())
|
|
|
|
if type == OtaMessageType.NOTIF:
|
|
res_list_obj = OtaHotelResNotifRq.HotelReservations(
|
|
hotel_reservation=reservations_list
|
|
)
|
|
|
|
ota_hotel_res_notif_rq = OtaHotelResNotifRq(
|
|
version="7.000", hotel_reservations=res_list_obj
|
|
)
|
|
|
|
try:
|
|
ota_hotel_res_notif_rq.model_validate(ota_hotel_res_notif_rq.model_dump())
|
|
except Exception as e:
|
|
_LOGGER.error(f"Validation error: {e}")
|
|
raise
|
|
|
|
return ota_hotel_res_notif_rq
|
|
if type == OtaMessageType.RETRIEVE:
|
|
res_list_obj = OtaResRetrieveRs.ReservationsList(
|
|
hotel_reservation=reservations_list
|
|
)
|
|
|
|
ota_res_retrieve_rs = OtaResRetrieveRs(
|
|
version="7.000", success="", reservations_list=res_list_obj
|
|
)
|
|
|
|
try:
|
|
ota_res_retrieve_rs.model_validate(ota_res_retrieve_rs.model_dump())
|
|
except Exception as e:
|
|
_LOGGER.error(f"Validation error: {e}")
|
|
raise
|
|
|
|
return ota_res_retrieve_rs
|
|
|
|
raise ValueError(f"Unsupported message type: {type}")
|
|
|
|
|
|
# Usage examples
|
|
if __name__ == "__main__":
|
|
# Create customer data using simple data class
|
|
customer_data = CustomerData(
|
|
given_name="John",
|
|
surname="Doe",
|
|
name_prefix="Mr.",
|
|
phone_numbers=[
|
|
("+1234567890", PhoneTechType.MOBILE), # Phone number with type
|
|
("+0987654321", None), # Phone number without type
|
|
],
|
|
email_address="john.doe@example.com",
|
|
email_newsletter=True,
|
|
address_line="123 Main Street",
|
|
city_name="Anytown",
|
|
postal_code="12345",
|
|
country_code="US",
|
|
address_catalog=False,
|
|
gender="Male",
|
|
birth_date="1980-01-01",
|
|
language="en",
|
|
)
|
|
|
|
# Create customer for OtaHotelResNotifRq
|
|
notif_customer = CustomerFactory.create_notif_customer(customer_data)
|
|
print(
|
|
"Created NotifCustomer:",
|
|
notif_customer.person_name.given_name,
|
|
notif_customer.person_name.surname,
|
|
)
|
|
|
|
# Create customer for OtaResRetrieveRs
|
|
retrieve_customer = CustomerFactory.create_retrieve_customer(customer_data)
|
|
print(
|
|
"Created RetrieveCustomer:",
|
|
retrieve_customer.person_name.given_name,
|
|
retrieve_customer.person_name.surname,
|
|
)
|
|
|
|
# Convert back to data class
|
|
converted_data = CustomerFactory.from_notif_customer(notif_customer)
|
|
print("Converted back to data:", converted_data.given_name, converted_data.surname)
|
|
|
|
# Verify they contain the same information
|
|
print("Original and converted data match:", customer_data == converted_data)
|
|
|
|
print("\n--- HotelReservationIdFactory Examples ---")
|
|
|
|
# Create hotel reservation ID data
|
|
reservation_id_data = HotelReservationIdData(
|
|
res_id_type="123",
|
|
res_id_value="RESERVATION-456",
|
|
res_id_source="HOTEL_SYSTEM",
|
|
res_id_source_context="BOOKING_ENGINE",
|
|
)
|
|
|
|
# Create HotelReservationId for both types
|
|
notif_res_id = HotelReservationIdFactory.create_notif_hotel_reservation_id(
|
|
reservation_id_data
|
|
)
|
|
retrieve_res_id = HotelReservationIdFactory.create_retrieve_hotel_reservation_id(
|
|
reservation_id_data
|
|
)
|
|
|
|
print(
|
|
"Created NotifHotelReservationId:",
|
|
notif_res_id.res_id_type,
|
|
notif_res_id.res_id_value,
|
|
)
|
|
print(
|
|
"Created RetrieveHotelReservationId:",
|
|
retrieve_res_id.res_id_type,
|
|
retrieve_res_id.res_id_value,
|
|
)
|
|
|
|
# Convert back to data class
|
|
converted_res_id_data = HotelReservationIdFactory.from_notif_hotel_reservation_id(
|
|
notif_res_id
|
|
)
|
|
print(
|
|
"Converted back to reservation ID data:",
|
|
converted_res_id_data.res_id_type,
|
|
converted_res_id_data.res_id_value,
|
|
)
|
|
|
|
# Verify they contain the same information
|
|
print(
|
|
"Original and converted reservation ID data match:",
|
|
reservation_id_data == converted_res_id_data,
|
|
)
|
|
|
|
print("\n--- ResGuestFactory Examples ---")
|
|
|
|
# Create complete ResGuests structure for OtaHotelResNotifRq - much simpler!
|
|
notif_res_guests = ResGuestFactory.create_notif_res_guests(customer_data)
|
|
print(
|
|
"Created NotifResGuests with customer:",
|
|
notif_res_guests.res_guest.profiles.profile_info.profile.customer.person_name.given_name,
|
|
)
|
|
|
|
# Create complete ResGuests structure for OtaResRetrieveRs - much simpler!
|
|
retrieve_res_guests = ResGuestFactory.create_retrieve_res_guests(customer_data)
|
|
print(
|
|
"Created RetrieveResGuests with customer:",
|
|
retrieve_res_guests.res_guest.profiles.profile_info.profile.customer.person_name.given_name,
|
|
)
|
|
|
|
# Extract primary customer data back from ResGuests structure
|
|
extracted_data = ResGuestFactory.extract_primary_customer(retrieve_res_guests)
|
|
print("Extracted customer data:", extracted_data.given_name, extracted_data.surname)
|
|
|
|
# Verify roundtrip conversion
|
|
print("Roundtrip conversion successful:", customer_data == extracted_data)
|
|
|
|
print("\n--- Unified AlpineBitsFactory Examples ---")
|
|
|
|
# Much simpler approach - single factory with enum parameter!
|
|
print("=== Customer Creation ===")
|
|
notif_customer = AlpineBitsFactory.create(customer_data, OtaMessageType.NOTIF)
|
|
retrieve_customer = AlpineBitsFactory.create(customer_data, OtaMessageType.RETRIEVE)
|
|
print("Created customers using unified factory")
|
|
|
|
print("=== HotelReservationId Creation ===")
|
|
reservation_id_data = HotelReservationIdData(
|
|
res_id_type="123", res_id_value="RESERVATION-456", res_id_source="HOTEL_SYSTEM"
|
|
)
|
|
notif_res_id = AlpineBitsFactory.create(reservation_id_data, OtaMessageType.NOTIF)
|
|
retrieve_res_id = AlpineBitsFactory.create(
|
|
reservation_id_data, OtaMessageType.RETRIEVE
|
|
)
|
|
print("Created reservation IDs using unified factory")
|
|
|
|
print("=== Comments Creation ===")
|
|
comments_data = CommentsData(
|
|
comments=[
|
|
CommentData(
|
|
name=CommentName2.CUSTOMER_COMMENT,
|
|
text="This is a customer comment about the reservation",
|
|
list_items=[
|
|
CommentListItemData(
|
|
value="Special dietary requirements: vegetarian",
|
|
list_item="1",
|
|
language="en",
|
|
),
|
|
CommentListItemData(
|
|
value="Late arrival expected", list_item="2", language="en"
|
|
),
|
|
],
|
|
),
|
|
CommentData(
|
|
name=CommentName2.ADDITIONAL_INFO,
|
|
text="Additional information about the stay",
|
|
),
|
|
]
|
|
)
|
|
notif_comments = AlpineBitsFactory.create(comments_data, OtaMessageType.NOTIF)
|
|
retrieve_comments = AlpineBitsFactory.create(comments_data, OtaMessageType.RETRIEVE)
|
|
print("Created comments using unified factory")
|
|
|
|
print("=== ResGuests Creation ===")
|
|
notif_res_guests = AlpineBitsFactory.create_res_guests(
|
|
customer_data, OtaMessageType.NOTIF
|
|
)
|
|
retrieve_res_guests = AlpineBitsFactory.create_res_guests(
|
|
customer_data, OtaMessageType.RETRIEVE
|
|
)
|
|
print("Created ResGuests using unified factory")
|
|
|
|
print("=== Data Extraction ===")
|
|
# Extract data back using unified interface
|
|
extracted_customer_data = AlpineBitsFactory.extract_data(notif_customer)
|
|
extracted_res_id_data = AlpineBitsFactory.extract_data(notif_res_id)
|
|
extracted_comments_data = AlpineBitsFactory.extract_data(retrieve_comments)
|
|
extracted_from_res_guests = AlpineBitsFactory.extract_data(retrieve_res_guests)
|
|
|
|
print("Data extraction successful:")
|
|
print("- Customer roundtrip:", customer_data == extracted_customer_data)
|
|
print("- ReservationId roundtrip:", reservation_id_data == extracted_res_id_data)
|
|
print("- Comments roundtrip:", comments_data == extracted_comments_data)
|
|
print("- ResGuests roundtrip:", customer_data == extracted_from_res_guests)
|
|
|
|
print("\n--- Comparison with old approach ---")
|
|
print("Old way required multiple imports and knowing specific factory methods")
|
|
print("New way: single import, single factory, enum parameter to specify type!")
|