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

Lots of refactoring and simplification in conversions_service

Browse Source
This commit is contained in:
Jonas Linter
2025-12-02 15:45:40 +01:00
parent d458f4f2c0
commit 2c1bdf6840
1 changed files with 308 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

485
src/alpine_bits_python/conversion_service.py
View File

@@ -2,7 +2,8 @@
import asyncio
import xml.etree.ElementTree as ET
from datetime import UTC, datetime
from dataclasses import dataclass, field
from datetime import UTC, date, datetime
from decimal import Decimal
from typing import Any
@@ -29,6 +30,42 @@ _LOGGER = get_logger(__name__)
MAX_CONCURRENT_RESERVATIONS = 10
@dataclass(slots=True)
class ParsedRoomReservation:
"""Typed representation of a single <roomReservation> entry."""
pms_hotel_reservation_id: str
room_number: str | None
arrival_date: date | None
departure_date: date | None
room_status: str | None
room_type: str | None
num_adults: int | None
rate_plan_code: str | None
connected_room_type: str | None
daily_sales: list[dict[str, str]] = field(default_factory=list)
total_revenue: Decimal | None = None
daily_sales_count: int = 0
@dataclass(slots=True)
class ParsedReservationData:
"""Typed representation of reservation metadata and rooms."""
hotel_id: str | None
pms_reservation_id: int
guest_id: int | None
reservation_number: str | None
reservation_date: date | None
creation_time: datetime | None
reservation_type: str | None
booking_channel: str | None
advertising_medium: str | None
advertising_partner: str | None
advertising_campagne: str | None
room_reservations: list[ParsedRoomReservation] = field(default_factory=list)
class ConversionService:
"""Service for processing and storing conversion/daily sales data.
@@ -83,6 +120,200 @@ class ConversionService:
f"session must be AsyncSession or SessionMaker, got {type(session)}"
)
@staticmethod
def _parse_required_int(value: str | None, field_name: str) -> int:
"""Parse an integer attribute that must be present."""
if value in (None, ""):
raise ValueError(f"{field_name} is required")
try:
return int(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{field_name} must be an integer (value={value})") from exc
@staticmethod
def _parse_optional_int(value: str | None, field_name: str) -> int | None:
"""Parse an optional integer attribute, logging on failure."""
if value in (None, ""):
return None
try:
return int(value)
except (TypeError, ValueError):
_LOGGER.warning("Invalid %s value: %s", field_name, value)
return None
@staticmethod
def _parse_date(value: str | None, field_name: str) -> date | None:
"""Parse a YYYY-MM-DD formatted date string."""
if not value:
return None
try:
return datetime.strptime(value, "%Y-%m-%d").date()
except ValueError:
_LOGGER.warning("Invalid %s format: %s", field_name, value)
return None
@staticmethod
def _parse_datetime(value: str | None, field_name: str) -> datetime | None:
"""Parse an ISO timestamp string."""
if not value:
return None
try:
normalized = value.replace("Z", "+00:00")
return datetime.fromisoformat(normalized)
except ValueError:
_LOGGER.warning("Invalid %s format: %s", field_name, value)
return None
def _parse_daily_sales(
self, daily_sales_elem: ET.Element | None
) -> tuple[list[dict[str, str]], Decimal | None, int]:
"""Extract the list of sale dictionaries and aggregate revenue information."""
if daily_sales_elem is None:
return [], None, 0
daily_sales_list: list[dict[str, str]] = []
total_revenue = Decimal(0)
sale_count = 0
for daily_sale in daily_sales_elem.findall("dailySale"):
sale_count += 1
sale_data: dict[str, str] = {}
sale_date_str = daily_sale.get("date")
if sale_date_str:
sale_data["date"] = sale_date_str
revenue_total_str = daily_sale.get("revenueTotal")
if revenue_total_str:
sale_data["revenueTotal"] = revenue_total_str
try:
total_revenue += Decimal(revenue_total_str)
except (ValueError, TypeError):
_LOGGER.warning(
"Invalid revenueTotal value: %s", revenue_total_str
)
# Copy the remaining optional revenue buckets if present
for field_name in (
"revenueLogis",
"revenueBoard",
"revenueFB",
"revenueSpa",
"revenueOther",
):
value = daily_sale.get(field_name)
if value:
sale_data[field_name] = value
if sale_data:
daily_sales_list.append(sale_data)
total_revenue_value = total_revenue if total_revenue > 0 else None
return daily_sales_list, total_revenue_value, sale_count
def _parse_room_reservation(
self, room_elem: ET.Element, pms_reservation_id: int, room_index: int
) -> ParsedRoomReservation:
"""Convert a <roomReservation> element into ParsedRoomReservation."""
arrival_date = self._parse_date(room_elem.get("arrival"), "arrival date")
departure_date = self._parse_date(
room_elem.get("departure"), "departure date"
)
num_adults = self._parse_optional_int(room_elem.get("adults"), "adults")
room_number = room_elem.get("roomNumber")
if room_number is None:
_LOGGER.debug(
"Room reservation %s #%d has no roomNumber", pms_reservation_id, room_index
)
daily_sales, total_revenue, sale_count = self._parse_daily_sales(
room_elem.find("dailySales")
)
return ParsedRoomReservation(
pms_hotel_reservation_id=f"{pms_reservation_id}_{room_number}",
room_number=room_number,
arrival_date=arrival_date,
departure_date=departure_date,
room_status=room_elem.get("status"),
room_type=room_elem.get("roomType"),
num_adults=num_adults,
rate_plan_code=room_elem.get("ratePlanCode"),
connected_room_type=room_elem.get("connectedRoomType"),
daily_sales=daily_sales,
total_revenue=total_revenue,
daily_sales_count=sale_count,
)
def _parse_reservation_element(
self, reservation_elem: ET.Element
) -> ParsedReservationData | None:
"""Convert a <reservation> element into a structured representation."""
try:
pms_reservation_id = self._parse_required_int(
reservation_elem.get("id"), "reservation id"
)
except ValueError as exc:
_LOGGER.error(
"Invalid reservation metadata in reservation element: %s", exc
)
return None
room_reservations_elem = reservation_elem.find("roomReservations")
if room_reservations_elem is None:
_LOGGER.debug(
"No roomReservations found for reservation %s", pms_reservation_id
)
return None
room_reservations = [
self._parse_room_reservation(room_elem, pms_reservation_id, idx)
for idx, room_elem in enumerate(
room_reservations_elem.findall("roomReservation")
)
]
if not room_reservations:
_LOGGER.debug(
"Reservation %s has no roomReservation entries", pms_reservation_id
)
return None
guest_elem = reservation_elem.find("guest")
guest_id = None
if guest_elem is not None:
guest_id = self._parse_optional_int(guest_elem.get("id"), "guest id")
return ParsedReservationData(
hotel_id=reservation_elem.get("hotelID"),
pms_reservation_id=pms_reservation_id,
guest_id=guest_id,
reservation_number=reservation_elem.get("number"),
reservation_date=self._parse_date(
reservation_elem.get("date"), "reservation date"
),
creation_time=self._parse_datetime(
reservation_elem.get("creationTime"), "creation time"
),
reservation_type=reservation_elem.get("type"),
booking_channel=reservation_elem.get("bookingChannel"),
advertising_medium=reservation_elem.get("advertisingMedium"),
advertising_partner=reservation_elem.get("advertisingPartner"),
advertising_campagne=reservation_elem.get("advertisingCampagne"),
room_reservations=room_reservations,
)
async def _extract_unique_guests_from_xml(
self, reservations: list
) -> dict[tuple[str, int], ConversionGuestData]:
@@ -651,63 +882,12 @@ class ConversionService:
stats = {
"daily_sales_count": 0,
}
hotel_id = reservation_elem.get("hotelID")
try:
# Extract reservation metadata
pms_reservation_id = int(reservation_elem.get("id"))
except ValueError as e:
_LOGGER.error("Invalid reservation metadata in reservation element: %s", e)
parsed_reservation = self._parse_reservation_element(reservation_elem)
if not parsed_reservation:
return stats
reservation_number = reservation_elem.get("number")
reservation_date_str = reservation_elem.get("date")
creation_time_str = reservation_elem.get("creationTime")
reservation_type = reservation_elem.get("type")
booking_channel = reservation_elem.get("bookingChannel")
# Extract guest information from guest element
guest_elem = reservation_elem.find("guest")
guest_birth_date_str = None
guest_id = None
if guest_elem is not None:
guest_id = guest_elem.get("id")
# Advertising/tracking data
advertising_medium = reservation_elem.get("advertisingMedium")
advertising_partner = reservation_elem.get("advertisingPartner")
advertising_campagne = reservation_elem.get("advertisingCampagne")
# Parse dates
reservation_date = None
if reservation_date_str:
try:
reservation_date = datetime.strptime(
reservation_date_str, "%Y-%m-%d"
).date()
except ValueError:
_LOGGER.warning(
"Invalid reservation date format: %s", reservation_date_str
)
creation_time = None
if creation_time_str:
try:
creation_time = datetime.fromisoformat(
creation_time_str.replace("Z", "+00:00")
)
except ValueError:
_LOGGER.warning("Invalid creation time format: %s", creation_time_str)
# Process all room reservations
room_reservations = reservation_elem.find("roomReservations")
if room_reservations is None:
_LOGGER.debug(
"No roomReservations found for reservation %s", pms_reservation_id
)
return stats
hotel_id = parsed_reservation.hotel_id
pms_reservation_id = parsed_reservation.pms_reservation_id
# ConversionGuests have already been bulk-upserted in Phase 1,
# so we can safely create/update conversions now
@@ -724,14 +904,22 @@ class ConversionService:
# Update existing conversion - only update reservation metadata and advertising data
# Guest info is stored in ConversionGuest table, not here
# Don't clear reservation/customer links (matching logic will update if needed)
existing_conversion.reservation_number = reservation_number
existing_conversion.reservation_date = reservation_date
existing_conversion.creation_time = creation_time
existing_conversion.reservation_type = reservation_type
existing_conversion.booking_channel = booking_channel
existing_conversion.advertising_medium = advertising_medium
existing_conversion.advertising_partner = advertising_partner
existing_conversion.advertising_campagne = advertising_campagne
existing_conversion.reservation_number = (
parsed_reservation.reservation_number
)
existing_conversion.reservation_date = parsed_reservation.reservation_date
existing_conversion.creation_time = parsed_reservation.creation_time
existing_conversion.reservation_type = parsed_reservation.reservation_type
existing_conversion.booking_channel = parsed_reservation.booking_channel
existing_conversion.advertising_medium = (
parsed_reservation.advertising_medium
)
existing_conversion.advertising_partner = (
parsed_reservation.advertising_partner
)
existing_conversion.advertising_campagne = (
parsed_reservation.advertising_campagne
)
existing_conversion.updated_at = datetime.now()
conversion = existing_conversion
_LOGGER.info(
@@ -746,17 +934,17 @@ class ConversionService:
# Links to existing entities (nullable, will be filled in after matching)
# Reservation metadata
hotel_id=hotel_id,
guest_id=guest_id, # Links to ConversionGuest
guest_id=parsed_reservation.guest_id, # Links to ConversionGuest
pms_reservation_id=pms_reservation_id,
reservation_number=reservation_number,
reservation_date=reservation_date,
creation_time=creation_time,
reservation_type=reservation_type,
booking_channel=booking_channel,
reservation_number=parsed_reservation.reservation_number,
reservation_date=parsed_reservation.reservation_date,
creation_time=parsed_reservation.creation_time,
reservation_type=parsed_reservation.reservation_type,
booking_channel=parsed_reservation.booking_channel,
# Advertising data
advertising_medium=advertising_medium,
advertising_partner=advertising_partner,
advertising_campagne=advertising_campagne,
advertising_medium=parsed_reservation.advertising_medium,
advertising_partner=parsed_reservation.advertising_partner,
advertising_campagne=parsed_reservation.advertising_campagne,
# Metadata
)
conversion = Conversion(**conversion_data.model_dump())
@@ -782,127 +970,70 @@ class ConversionService:
current_pms_hotel_reservation_ids = set()
# Process room reservations
for room_reservation in room_reservations.findall("roomReservation"):
# Extract room reservation details
arrival_str = room_reservation.get("arrival")
departure_str = room_reservation.get("departure")
room_status = room_reservation.get("status")
room_type = room_reservation.get("roomType")
room_number = room_reservation.get("roomNumber")
adults_str = room_reservation.get("adults")
rate_plan_code = room_reservation.get("ratePlanCode")
connected_room_type = room_reservation.get("connectedRoomType")
arrival_date = None
if arrival_str:
try:
arrival_date = datetime.strptime(arrival_str, "%Y-%m-%d").date()
except ValueError:
_LOGGER.warning("Invalid arrival date format: %s", arrival_str)
departure_date = None
if departure_str:
try:
departure_date = datetime.strptime(departure_str, "%Y-%m-%d").date()
except ValueError:
_LOGGER.warning("Invalid departure date format: %s", departure_str)
num_adults = None
if adults_str:
try:
num_adults = int(adults_str)
except ValueError:
_LOGGER.warning("Invalid adults value: %s", adults_str)
# Create composite ID for upsert: pms_reservation_id + room_number
# This allows updating the same room reservation if it appears again
pms_hotel_reservation_id = f"{pms_reservation_id}_{room_number}"
# Track this room as present in current XML
current_pms_hotel_reservation_ids.add(pms_hotel_reservation_id)
# Process daily sales and extract total revenue
daily_sales_elem = room_reservation.find("dailySales")
daily_sales_list = []
total_revenue = Decimal(0)
if daily_sales_elem is not None:
for daily_sale in daily_sales_elem.findall("dailySale"):
stats["daily_sales_count"] += 1
# Extract daily sale data
sale_date_str = daily_sale.get("date")
daily_sale_obj = {}
if sale_date_str:
daily_sale_obj["date"] = sale_date_str
# Extract all revenue fields
revenue_total_str = daily_sale.get("revenueTotal")
if revenue_total_str:
daily_sale_obj["revenueTotal"] = revenue_total_str
try:
total_revenue += Decimal(revenue_total_str)
except (ValueError, TypeError):
_LOGGER.warning(
"Invalid revenueTotal value: %s", revenue_total_str
)
# Add other revenue fields if present
if daily_sale.get("revenueLogis"):
daily_sale_obj["revenueLogis"] = daily_sale.get("revenueLogis")
if daily_sale.get("revenueBoard"):
daily_sale_obj["revenueBoard"] = daily_sale.get("revenueBoard")
if daily_sale.get("revenueFB"):
daily_sale_obj["revenueFB"] = daily_sale.get("revenueFB")
if daily_sale.get("revenueSpa"):
daily_sale_obj["revenueSpa"] = daily_sale.get("revenueSpa")
if daily_sale.get("revenueOther"):
daily_sale_obj["revenueOther"] = daily_sale.get("revenueOther")
if daily_sale_obj: # Only add if has data
daily_sales_list.append(daily_sale_obj)
for room_reservation in parsed_reservation.room_reservations:
current_pms_hotel_reservation_ids.add(
room_reservation.pms_hotel_reservation_id
)
stats["daily_sales_count"] += room_reservation.daily_sales_count
# Check if room reservation already exists using batch-loaded data
existing_room_reservation = existing_rooms.get(pms_hotel_reservation_id)
existing_room_reservation = existing_rooms.get(
room_reservation.pms_hotel_reservation_id
)
if existing_room_reservation:
# Update existing room reservation with all fields
existing_room_reservation.arrival_date = arrival_date
existing_room_reservation.departure_date = departure_date
existing_room_reservation.room_status = room_status
existing_room_reservation.room_type = room_type
existing_room_reservation.num_adults = num_adults
existing_room_reservation.rate_plan_code = rate_plan_code
existing_room_reservation.connected_room_type = connected_room_type
existing_room_reservation.arrival_date = (
room_reservation.arrival_date
)
existing_room_reservation.departure_date = (
room_reservation.departure_date
)
existing_room_reservation.room_status = (
room_reservation.room_status
)
existing_room_reservation.room_type = room_reservation.room_type
existing_room_reservation.num_adults = room_reservation.num_adults
existing_room_reservation.rate_plan_code = (
room_reservation.rate_plan_code
)
existing_room_reservation.connected_room_type = (
room_reservation.connected_room_type
)
existing_room_reservation.daily_sales = (
daily_sales_list if daily_sales_list else None
room_reservation.daily_sales
if room_reservation.daily_sales
else None
)
existing_room_reservation.total_revenue = (
total_revenue if total_revenue > 0 else None
room_reservation.total_revenue
)
existing_room_reservation.updated_at = datetime.now()
_LOGGER.debug(
"Updated room reservation %s (pms_id=%s, room=%s)",
existing_room_reservation.id,
pms_reservation_id,
room_number,
room_reservation.room_number,
)
else:
# Create new room reservation
room_reservation_record = ConversionRoom(
conversion_id=conversion.id,
pms_hotel_reservation_id=pms_hotel_reservation_id,
arrival_date=arrival_date,
departure_date=departure_date,
room_status=room_status,
room_type=room_type,
room_number=room_number,
num_adults=num_adults,
rate_plan_code=rate_plan_code,
connected_room_type=connected_room_type,
daily_sales=daily_sales_list if daily_sales_list else None,
total_revenue=total_revenue if total_revenue > 0 else None,
pms_hotel_reservation_id=room_reservation.pms_hotel_reservation_id,
arrival_date=room_reservation.arrival_date,
departure_date=room_reservation.departure_date,
room_status=room_reservation.room_status,
room_type=room_reservation.room_type,
room_number=room_reservation.room_number,
num_adults=room_reservation.num_adults,
rate_plan_code=room_reservation.rate_plan_code,
connected_room_type=room_reservation.connected_room_type,
daily_sales=(
room_reservation.daily_sales
if room_reservation.daily_sales
else None
),
total_revenue=room_reservation.total_revenue,
created_at=datetime.now(),
updated_at=datetime.now(),
)
@@ -910,8 +1041,8 @@ class ConversionService:
_LOGGER.debug(
"Created room reservation (pms_id=%s, room=%s, adults=%s)",
pms_reservation_id,
room_number,
num_adults,
room_reservation.room_number,
room_reservation.num_adults,
)
# Delete room entries that are no longer present in the current XML