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

Saving to db and creating quote-request from db works

Browse Source
This commit is contained in:
Jonas Linter
2025-09-29 10:17:00 +02:00
parent 958e48b40a
commit 784ff0e5da
4 changed files with 121 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

224
src/alpine_bits_python/main.py
View File

@@ -57,23 +57,21 @@ def main():
SessionLocal = get_session_local(config)
db = SessionLocal()
# Load data from JSON file
json_path = os.path.join(os.path.dirname(__file__), '../../test_data/wix_test_data_20250928_132611.json')
with open(json_path, 'r', encoding='utf-8') as f:
wix_data = json.load(f)
data = wix_data["data"]["data"]
contact_info = data.get("contact", {})
first_name = contact_info.get("name", {}).get("first")
last_name = contact_info.get("name", {}).get("last")
email = contact_info.get("email")
phone_number = contact_info.get("phones", [{}])[0].get("e164Phone") # phone without formatting
phone_number = contact_info.get("phones", [{}])[0].get("e164Phone")
locale = contact_info.get("locale", "de-de")
contact_id = contact_info.get("contactId")
name_prefix = data.get("field:anrede")
email_newsletter = data.get("field:form_field_5a7b", "") != "Non selezionato"
@@ -93,7 +91,6 @@ def main():
num_adults = int(data.get("field:number_7cf5") or 2)
num_children = int(data.get("field:anzahl_kinder") or 0)
children_ages = []
if num_children > 0:
for k in data.keys():
if k.startswith("field:alter_kind_"):
@@ -103,13 +100,6 @@ def main():
except Exception:
pass
guest_counts = GuestCountsFactory().create_retrieve_guest_counts(num_adults, children_ages)
# print guests and kids info for debugging
print(f"Guests: {num_adults} adults, {num_children} children, ages: {children_ages}")
# UTM and offer
utm_fields = [
("utm_Source", "utm_source"),
@@ -126,9 +116,7 @@ def main():
utm_comment = " | ".join(utm_comment_text) if utm_comment_text else None
offer = data.get("field:angebot_auswaehlen")
contact_id = data.get("contact", {}).get("contactId")
# Save customer and reservation to DB
# Save all relevant data to DB (including new fields)
db_customer = DBCustomer(
given_name=first_name,
surname=last_name,
@@ -144,6 +132,8 @@ def main():
gender=gender,
birth_date=birth_date,
language=language,
address_catalog=False,
name_title=None,
)
db.add(db_customer)
db.commit()
@@ -160,124 +150,126 @@ def main():
offer=offer,
utm_comment=utm_comment,
created_at=datetime.now(timezone.utc),
utm_source=data.get("field:utm_source"),
utm_medium=data.get("field:utm_medium"),
utm_campaign=data.get("field:utm_campaign"),
utm_term=data.get("field:utm_term"),
utm_content=data.get("field:utm_content"),
user_comment=data.get("field:long_answer_3524", ""),
fbclid=data.get("field:fbclid"),
gclid=data.get("field:gclid"),
hotel_code="123",
hotel_name="Frangart Inn",
)
db.add(db_reservation)
db.commit()
db.refresh(db_reservation)
# Success - use None instead of object() for cleaner XML output
success = None
# Now read back from DB
customer = db.query(DBCustomer).filter_by(id=db_reservation.customer_id).first()
reservation = db.query(DBReservation).filter_by(id=db_reservation.id).first()
# UniqueID, we are using the formid as a stable unique id
form_id = data.get("formId")
# Generate XML from DB data
create_xml_from_db(customer, reservation)
# hardcoding the type to 14 is allowed because 15 is only for cancellations and we don't handle those
unique_id = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.UniqueId(
type_value=ab.UniqueIdType2.VALUE_14, id=form_id
)
db.close()
time_span = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay.TimeSpan(
start=start_date, end=end_date
)
# RoomStay with TimeSpan
room_stay = (
ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay(
time_span=time_span,
guest_counts=guest_counts,
)
)
room_stays = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays(
room_stay=[room_stay],
)
# CustomerData
phone_numbers = [(phone_number, PhoneTechType.MOBILE)] if phone_number else []
def create_xml_from_db(customer: DBCustomer, reservation: DBReservation):
from .simplified_access import CustomerData, GuestCountsFactory, HotelReservationIdData, AlpineBitsFactory, OtaMessageType, CommentData, CommentsData, CommentListItemData
from .generated import alpinebits as ab
from datetime import datetime, timezone
# Prepare data for XML
phone_numbers = [(customer.phone, PhoneTechType.MOBILE)] if customer.phone else []
customer_data = CustomerData(
given_name=first_name,
surname=last_name,
name_prefix=name_prefix,
given_name=customer.given_name,
surname=customer.surname,
name_prefix=customer.name_prefix,
name_title=customer.name_title,
phone_numbers=phone_numbers,
email_address=email,
email_newsletter=email_newsletter,
address_line=address_line,
city_name=city_name,
postal_code=postal_code,
country_code=country_code,
address_catalog=False,
gender=gender,
birth_date=birth_date,
language=language,
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, OtaMessageType.RETRIEVE)
# Guest counts
children_ages = [int(a) for a in reservation.children_ages.split(",") if a]
guest_counts = GuestCountsFactory.create_retrieve_guest_counts(reservation.num_adults, children_ages)
# UniqueID
unique_id = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.UniqueId(
type_value=ab.UniqueIdType2.VALUE_14, id=reservation.form_id
)
# TimeSpan
time_span = ab.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 = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay(
time_span=time_span,
guest_counts=guest_counts,
)
room_stays = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays(
room_stay=[room_stay],
)
# HotelReservationId
hotel_res_id_data = HotelReservationIdData(
res_id_type="13",
res_id_value=data.get("field:fbclid") or data.get("field:gclid"),
res_id_value=reservation.fbclid or reservation.gclid,
res_id_source=None,
res_id_source_context="99tales",
)
# Create HotelReservationId using the factory
hotel_res_id = alpine_bits_factory.create(hotel_res_id_data, OtaMessageType.RETRIEVE)
# Use the actual nested HotelReservationIds class
hotel_res_ids = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.HotelReservationIds(
hotel_reservation_id=[hotel_res_id]
)
# Basic property info (hardcoded for now)
basic_property_info = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.BasicPropertyInfo(
hotel_code="123", hotel_name="Frangart Inn"
hotel_code=reservation.hotel_code,
hotel_name=reservation.hotel_name,
)
user_comment_text = data.get("field:long_answer_3524", "")
comment = None
if user_comment_text:
# Comments from UTM fields and other info
comment = CommentData(
name= ab.CommentName2.CUSTOMER_COMMENT,
text=user_comment_text,
list_items=[CommentListItemData(
value="Landing page comment",
language=language,
list_item="1",
)],
)
# Comments
offer_comment = CommentData(
name= ab.CommentName2.ADDITIONAL_INFO,
name=ab.CommentName2.ADDITIONAL_INFO,
text="Angebot/Offerta",
list_items=[CommentListItemData(
value=offer,
language=language,
value=reservation.offer,
language=customer.language,
list_item="1",
)],
)
comments = [offer_comment, comment] if comment else [offer_comment]
comments_data = CommentsData(comments=comments)
comments = alpine_bits_factory.create(comments_data, OtaMessageType.RETRIEVE)
# ResGlobalInfo
res_global_info = (
ab.OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo(
hotel_reservation_ids=hotel_res_ids, basic_property_info=basic_property_info, comments=comments
comment = None
if reservation.user_comment:
comment = CommentData(
name=ab.CommentName2.CUSTOMER_COMMENT,
text=reservation.user_comment,
list_items=[CommentListItemData(
value="Landing page comment",
language=customer.language,
list_item="1",
)],
)
comments = [offer_comment, comment] if comment else [offer_comment]
comments_data = CommentsData(comments=comments)
comments_xml = alpine_bits_factory.create(comments_data, OtaMessageType.RETRIEVE)
res_global_info = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo(
hotel_reservation_ids=hotel_res_ids,
basic_property_info=basic_property_info,
comments=comments_xml,
)
# Hotel Reservation
hotel_reservation = ab.OtaResRetrieveRs.ReservationsList.HotelReservation(
create_date_time=datetime.now(timezone.utc).isoformat(),
res_status=ab.HotelReservationResStatus.REQUESTED,
@@ -287,62 +279,38 @@ def main():
res_guests=res_guests,
res_global_info=res_global_info,
)
reservations_list = ab.OtaResRetrieveRs.ReservationsList(
hotel_reservation=[hotel_reservation]
)
# Root element
ota_res_retrieve_rs = ab.OtaResRetrieveRs(
version="7.000", success=success, reservations_list=reservations_list
version="7.000", success=None, reservations_list=reservations_list
)
# Serialize using Pydantic's model_dump and convert to XML
# Serialize to XML
try:
# First validate the model
ota_res_retrieve_rs.model_validate(ota_res_retrieve_rs.model_dump())
print("✅ Pydantic validation successful!")
# For XML serialization with Pydantic models, we need to use xsdata-pydantic serializer
from xsdata.formats.dataclass.serializers.config import SerializerConfig
from xsdata_pydantic.bindings import XmlSerializer
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"
)
serializer = XmlSerializer(config=config)
# Use ns_map to control namespace prefixes - set default namespace
ns_map = {None: "http://www.opentravel.org/OTA/2003/05"}
xml_string = serializer.render(ota_res_retrieve_rs, ns_map=ns_map)
with open("output.xml", "w", encoding="utf-8") as outfile:
outfile.write(xml_string)
print("✅ XML serialization successful!")
print(f"Generated XML written to output.xml")
# Also print the pretty formatted XML to console
print("\n📄 Generated XML:")
print(xml_string)
# Test parsing back
from xsdata_pydantic.bindings import XmlParser
parser = XmlParser()
with open("output.xml", "r", encoding="utf-8") as infile:
xml_content = infile.read()
parsed_result = parser.from_string(xml_content, ab.OtaResRetrieveRs)
print("✅ Round-trip validation successful!")
print(
f"Parsed reservation status: {parsed_result.reservations_list.hotel_reservation[0].res_status}"
)
print(f"Parsed reservation status: {parsed_result.reservations_list.hotel_reservation[0].res_status}")
except Exception as e:
print(f"❌ Validation/Serialization failed: {e}")
finally:
db.close()