jonas/alpinebits_python
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 4 Wiki Activity

Push support maybe?

Browse Source
This commit is contained in:
Jonas Linter
2025-10-02 16:41:38 +02:00
parent 325965bb10
commit 9fa4c23719
1 changed files with 121 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
src/alpine_bits_python/alpine_bits_helpers.py
View File

@@ -671,7 +671,7 @@ class AlpineBitsFactory:
raise ValueError(f"Unsupported object type: {type(obj)}") raise ValueError(f"Unsupported object type: {type(obj)}")
def create_xml_from_db(list: list[Tuple[Reservation, Customer]]): def create_xml_from_db(list: list[Tuple[Reservation, Customer]], message_type: OtaMessageType = OtaMessageType.RETRIEVE) -> Union[OtaResRetrieveRs, OtaHotelResNotifRq]:
"""Create RetrievedReservation XML from database entries. """Create RetrievedReservation XML from database entries.
list of pairs (Reservation, Customer) list of pairs (Reservation, Customer)
@@ -709,38 +709,65 @@ def create_xml_from_db(list: list[Tuple[Reservation, Customer]]):
) )
alpine_bits_factory = AlpineBitsFactory() alpine_bits_factory = AlpineBitsFactory()
res_guests = alpine_bits_factory.create_res_guests( res_guests = alpine_bits_factory.create_res_guests(
customer_data, OtaMessageType.RETRIEVE customer_data, message_type
) )
# Guest counts # Guest counts
children_ages = [int(a) for a in reservation.children_ages.split(",") if a] children_ages = [int(a) for a in reservation.children_ages.split(",") if a]
guest_counts = GuestCountsFactory.create_retrieve_guest_counts( if message_type == OtaMessageType.NOTIF:
reservation.num_adults, children_ages guest_counts = GuestCountsFactory.create_notif_guest_counts(
) reservation.num_adults, children_ages
)
else:
guest_counts = GuestCountsFactory.create_retrieve_guest_counts(
reservation.num_adults, children_ages
)
unique_id_string = reservation.unique_id unique_id_string = reservation.unique_id
# UniqueID # UniqueID
unique_id = OtaResRetrieveRs.ReservationsList.HotelReservation.UniqueId( if message_type == OtaMessageType.NOTIF:
type_value=UniqueIdType2.VALUE_14, id=unique_id_string unique_id = OtaHotelResNotifRq.HotelReservations.HotelReservation.UniqueId(
) type_value=UniqueIdType2.VALUE_14, id=unique_id_string
)
else:
unique_id = OtaResRetrieveRs.ReservationsList.HotelReservation.UniqueId(
type_value=UniqueIdType2.VALUE_14, id=unique_id_string
)
# TimeSpan # TimeSpan
time_span = OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay.TimeSpan( if message_type == OtaMessageType.NOTIF:
start=reservation.start_date.isoformat() time_span = OtaHotelResNotifRq.HotelReservations.HotelReservation.RoomStays.RoomStay.TimeSpan(
if reservation.start_date start=reservation.start_date.isoformat()
else None, if reservation.start_date
end=reservation.end_date.isoformat() if reservation.end_date else None, else None,
) end=reservation.end_date.isoformat() if reservation.end_date else None,
room_stay = ( )
OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay( room_stay = (
time_span=time_span, OtaHotelResNotifRq.HotelReservations.HotelReservation.RoomStays.RoomStay(
guest_counts=guest_counts, time_span=time_span,
guest_counts=guest_counts,
)
)
room_stays = OtaHotelResNotifRq.HotelReservations.HotelReservation.RoomStays(
room_stay=[room_stay],
)
else:
time_span = OtaResRetrieveRs.ReservationsList.HotelReservation.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 = (
OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay(
time_span=time_span,
guest_counts=guest_counts,
)
)
room_stays = OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays(
room_stay=[room_stay],
) )
)
room_stays = OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays(
room_stay=[room_stay],
)
res_id_source = "website" res_id_source = "website"
@@ -779,11 +806,16 @@ def create_xml_from_db(list: list[Tuple[Reservation, Customer]]):
) )
hotel_res_id = alpine_bits_factory.create( hotel_res_id = alpine_bits_factory.create(
hotel_res_id_data, OtaMessageType.RETRIEVE hotel_res_id_data, message_type
)
hotel_res_ids = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.HotelReservationIds(
hotel_reservation_id=[hotel_res_id]
) )
if message_type == OtaMessageType.NOTIF:
hotel_res_ids = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.HotelReservationIds(
hotel_reservation_id=[hotel_res_id]
)
else:
hotel_res_ids = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.HotelReservationIds(
hotel_reservation_id=[hotel_res_id]
)
if reservation.hotel_code is None: if reservation.hotel_code is None:
raise ValueError("Reservation hotel_code is None") raise ValueError("Reservation hotel_code is None")
@@ -794,10 +826,16 @@ def create_xml_from_db(list: list[Tuple[Reservation, Customer]]):
else: else:
hotel_name = str(reservation.hotel_name) hotel_name = str(reservation.hotel_name)
basic_property_info = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.BasicPropertyInfo( if message_type == OtaMessageType.NOTIF:
hotel_code=hotel_code, basic_property_info = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.BasicPropertyInfo(
hotel_name=hotel_name, hotel_code=hotel_code,
) hotel_name=hotel_name,
)
else:
basic_property_info = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.BasicPropertyInfo(
hotel_code=hotel_code,
hotel_name=hotel_name,
)
# Comments # Comments
offer_comment = None offer_comment = None
@@ -840,26 +878,45 @@ def create_xml_from_db(list: list[Tuple[Reservation, Customer]]):
comments_data = CommentsData(comments=comments) comments_data = CommentsData(comments=comments)
comments_xml = alpine_bits_factory.create( comments_xml = alpine_bits_factory.create(
comments_data, OtaMessageType.RETRIEVE comments_data, message_type
) )
res_global_info = ( if message_type == OtaMessageType.NOTIF:
OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo( res_global_info = (
hotel_reservation_ids=hotel_res_ids, OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo(
basic_property_info=basic_property_info, hotel_reservation_ids=hotel_res_ids,
comments=comments_xml, basic_property_info=basic_property_info,
comments=comments_xml,
)
) )
)
hotel_reservation = OtaResRetrieveRs.ReservationsList.HotelReservation( hotel_reservation = OtaHotelResNotifRq.HotelReservations.HotelReservation(
create_date_time=datetime.now(timezone.utc).isoformat(), create_date_time=datetime.now(timezone.utc).isoformat(),
res_status=HotelReservationResStatus.REQUESTED, res_status=HotelReservationResStatus.REQUESTED,
room_stay_reservation="true", room_stay_reservation="true",
unique_id=unique_id, unique_id=unique_id,
room_stays=room_stays, room_stays=room_stays,
res_guests=res_guests, res_guests=res_guests,
res_global_info=res_global_info, res_global_info=res_global_info,
) )
else:
res_global_info = (
OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo(
hotel_reservation_ids=hotel_res_ids,
basic_property_info=basic_property_info,
comments=comments_xml,
)
)
hotel_reservation = OtaResRetrieveRs.ReservationsList.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,
)
reservations_list.append(hotel_reservation) reservations_list.append(hotel_reservation)
@@ -868,21 +925,24 @@ def create_xml_from_db(list: list[Tuple[Reservation, Customer]]):
f"Error creating XML for reservation {reservation.unique_id} and customer {customer.given_name}: {e}" f"Error creating XML for reservation {reservation.unique_id} and customer {customer.given_name}: {e}"
) )
retrieved_reservations = OtaResRetrieveRs.ReservationsList( if message_type == OtaMessageType.NOTIF:
hotel_reservation=reservations_list hotel_reservations = OtaHotelResNotifRq.HotelReservations(hotel_reservation=reservations_list)
) ota_res_notif_rq = OtaHotelResNotifRq(version="7.000", hotel_reservations=hotel_reservations)
try:
ota_res_retrieve_rs = OtaResRetrieveRs( ota_res_notif_rq.model_validate(ota_res_notif_rq.model_dump())
version="7.000", success="", reservations_list=retrieved_reservations except Exception as e:
) _LOGGER.error(f"Validation error: {e}")
raise
try: return ota_res_notif_rq
ota_res_retrieve_rs.model_validate(ota_res_retrieve_rs.model_dump()) else:
except Exception as e: retrieved_reservations = OtaResRetrieveRs.ReservationsList(hotel_reservation=reservations_list)
_LOGGER.error(f"Validation error: {e}") ota_res_retrieve_rs = OtaResRetrieveRs(version="7.000", success="", reservations_list=retrieved_reservations)
raise try:
ota_res_retrieve_rs.model_validate(ota_res_retrieve_rs.model_dump())
return ota_res_retrieve_rs except Exception as e:
_LOGGER.error(f"Validation error: {e}")
raise
return ota_res_retrieve_rs
# Usage examples # Usage examples