from datetime import datetime, timezone import traceback from typing import Union, Optional, Any, TypeVar from pydantic import BaseModel, ConfigDict, Field from dataclasses import dataclass from enum import Enum from typing import Tuple from alpine_bits_python.db import Customer, Reservation # Import the generated classes from .generated.alpinebits import ( HotelReservationResStatus, OtaHotelResNotifRq, OtaResRetrieveRs, CommentName2, ProfileProfileType, UniqueIdType2, ) import logging _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 RetrieveCustomer = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGuests.ResGuest.Profiles.ProfileInfo.Profile.Customer # Define type aliases for HotelReservationId types NotifHotelReservationId = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.HotelReservationIds.HotelReservationId RetrieveHotelReservationId = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.HotelReservationIds.HotelReservationId # 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) # phonetechtype enum 1,3,5 voice, fax, mobile class PhoneTechType(Enum): VOICE = "1" FAX = "3" MOBILE = "5" # 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] @dataclass class CustomerData: """Simple data class to hold customer information without nested type constraints.""" given_name: str surname: str name_prefix: None | str = None name_title: None | str = None phone_numbers: list[tuple[str, None | PhoneTechType]] = ( None # (phone_number, phone_tech_type) ) email_address: None | str = None email_newsletter: None | bool = ( None # True for "yes", False for "no", None for not specified ) address_line: None | str = None city_name: None | str = None postal_code: None | str = None country_code: None | str = None # Two-letter country code address_catalog: None | bool = ( None # True for "yes", False for "no", None for not specified ) gender: None | str = None # "Unknown", "Male", "Female" birth_date: None | str = None language: None | str = None # Two-letter language code def __post_init__(self): if self.phone_numbers is None: self.phone_numbers = [] class GuestCountsFactory: """Factory class to create GuestCounts instances for both OtaHotelResNotifRq and OtaResRetrieveRs.""" @staticmethod def create_guest_counts( adults: int, kids: Optional[list[int]] = 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) elif message_type == OtaMessageType.NOTIF: return GuestCountsFactory._create_guest_counts(adults, kids, NotifGuestCounts) else: raise ValueError(f"Unsupported message type: {message_type}") @staticmethod def _create_guest_counts( adults: int, kids: Optional[list[int]], guest_counts_class: type ) -> Any: """ Internal method to 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 OtaHotelResNotifRq and OtaResRetrieveRs.""" @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, data: CustomerData) -> Any: """Internal method to 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: """Internal method to convert any customer type to CustomerData.""" # Extract phone numbers phone_numbers = [] if customer.telephone: for tel in customer.telephone: phone_numbers.append( ( tel.phone_number, PhoneTechType(tel.phone_tech_type) if tel.phone_tech_type else None, ) ) # 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, ) @dataclass class HotelReservationIdData: """Simple data class to hold hotel reservation ID information without nested type constraints.""" res_id_type: str # Required field - pattern: [0-9]+ res_id_value: None | str = None # Max 64 characters res_id_source: None | str = None # Max 64 characters res_id_source_context: None | str = None # Max 64 characters 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: """Internal method to 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, ) @dataclass class CommentListItemData: """Simple data class to hold comment list item information.""" value: str # The text content of the list item list_item: str # Numeric identifier (pattern: [0-9]+) language: str # Two-letter language code (pattern: [a-z][a-z]) @dataclass class CommentData: """Simple data class to hold comment information without nested type constraints.""" name: CommentName2 # Required: "included services", "customer comment", "additional info" text: Optional[str] = None # Optional text content list_items: list[CommentListItemData] = None # Optional list items def __post_init__(self): if self.list_items is None: self.list_items = [] @dataclass class CommentsData: """Simple data class to hold multiple comments (1-3 max).""" comments: list[CommentData] = None # 1-3 comments maximum def __post_init__(self): if self.comments is None: self.comments = [] 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, ) ) # Extract comment data comment_data = CommentData( name=comment.name, text=comment.text, list_items=list_items_data ) comments_data_list.append(comment_data) 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, customer_class: type, customer_data: CustomerData ) -> Any: """Internal method to create 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: Union[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) else: return CustomerFactory.from_retrieve_customer(customer) class AlpineBitsFactory: """Unified factory class for creating AlpineBits objects with a simple interface.""" @staticmethod def create( data: Union[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) else: return CustomerFactory.create_retrieve_customer(data) elif isinstance(data, HotelReservationIdData): if message_type == OtaMessageType.NOTIF: return HotelReservationIdFactory.create_notif_hotel_reservation_id(data) else: return HotelReservationIdFactory.create_retrieve_hotel_reservation_id( data ) elif isinstance(data, CommentsData): if message_type == OtaMessageType.NOTIF: return CommentFactory.create_notif_comments(data) else: return CommentFactory.create_retrieve_comments(data) else: raise ValueError(f"Unsupported data type: {type(data)}") @staticmethod def create_res_guests( customer_data: CustomerData, message_type: OtaMessageType ) -> Union[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) else: return ResGuestFactory.create_retrieve_res_guests(customer_data) @staticmethod def extract_data( obj: Any, ) -> Union[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) elif 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) elif 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) elif 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)}") def create_res_retrieve_response(list: list[Tuple[Reservation, Customer]]): """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 ) unique_id_string = reservation.unique_id 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(f"Unsupported message type: {message_type}") # UniqueID unique_id = UniqueId( type_value=UniqueIdType2.VALUE_14, id=unique_id_string ) # 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" if reservation.fbclid != "": klick_id = reservation.fbclid res_id_source = "meta" elif reservation.gclid != "": klick_id = reservation.gclid res_id_source = "google" # explicitly set klick_id to None otherwise an empty string will be sent if klick_id in (None, "", "None"): klick_id = None else: # extract string from Column object klick_id = str(klick_id) 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", ) # explicitly set klick_id to None otherwise an empty string will be sent if klick_id in (None, "", "None"): klick_id = None else: # extract string from Column object klick_id = str(klick_id) utm_medium = ( str(reservation.utm_medium) if reservation.utm_medium is not None and str(reservation.utm_medium) != "" else "website" ) #shorten klick_id if longer than 64 characters if klick_id is not None and len(klick_id) > 64: klick_id = klick_id[:64] if klick_id is "": klick_id = None hotel_res_id_data = HotelReservationIdData( res_id_type="13", res_id_value=klick_id, res_id_source=utm_medium, 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") else: 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", 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(timezone.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: retrieved_reservations = OtaHotelResNotifRq.HotelReservations( hotel_reservation=reservations_list ) ota_hotel_res_notif_rq = OtaHotelResNotifRq( version="7.000", hotel_reservations=retrieved_reservations ) 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 elif type == OtaMessageType.RETRIEVE: retrieved_reservations = OtaResRetrieveRs.ReservationsList( hotel_reservation=reservations_list ) ota_res_retrieve_rs = OtaResRetrieveRs( version="7.000", success="", reservations_list=retrieved_reservations ) 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 else: 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!")