diff --git a/alpinebits_capi_test.db b/alpinebits_capi_test.db new file mode 100644 index 0000000..2f72583 Binary files /dev/null and b/alpinebits_capi_test.db differ diff --git a/test_capi.py b/test_capi.py new file mode 100644 index 0000000..6eed501 --- /dev/null +++ b/test_capi.py @@ -0,0 +1,291 @@ +#!/usr/bin/env python3 +"""Test sending a test event to the Conversions Api from Meta.""" + +import asyncio +import json +import logging +import time + +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.orm import sessionmaker +from src.alpine_bits_python.customer_service import CustomerService +from src.alpine_bits_python.db import Base +from src.alpine_bits_python.reservation_service import ReservationService + +# Set up logging +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + +TEST_CODE = "TEST54726" + +# Meta CAPI configuration (placeholder values) +PIXEL_ID = "539512870322352" +ACCESS_TOKEN = "EAATsRaQOv94BPoib5XUn9ZBjNPfeZB4JlKJR1LYtiMdzbEoIa7XFDmHq3pY8UvOcHnbNYDraym107hwRd3EfzO8EpQ5ZB5C4OfF7KJ41KIrfQntkoWrCYcQReecd4vhzVk82hjm55yGDhkxzuNuzG85FZCT0nZB6VyIxZAVLR2estoUSAoQ06J742aMkZCN2AZDZD" +CAPI_ENDPOINT = f"https://graph.facebook.com/v19.0/{PIXEL_ID}/events" + + +async def load_test_data_from_db(): + """Load reservations and hashed customers from the database.""" + # Connect to the test database + db_url = "sqlite+aiosqlite:///alpinebits_capi_test.db" + engine = create_async_engine(db_url, echo=False) + + # Create tables if they don't exist + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + # Create async session + async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + async with async_session() as session: + # Initialize services + reservation_service = ReservationService(session) + customer_service = CustomerService(session) + + # Get all reservations with customers + reservations_with_customers = ( + await reservation_service.get_reservations_with_filters() + ) + + if not reservations_with_customers: + logger.warning("No reservations found in database") + return [] + + logger.info("Found %d reservations", len(reservations_with_customers)) + + # Prepare data with hashed customer info + result = [] + for reservation, customer in reservations_with_customers: + # Get hashed customer data + hashed_customer = await customer_service.get_hashed_customer(customer.id) + + result.append( + { + "reservation": reservation, + "customer": customer, + "hashed_customer": hashed_customer, + } + ) + + await engine.dispose() + return result + + +def _build_user_data(hashed_customer): + """Build user_data dict from hashed customer information. + + Args: + hashed_customer: HashedCustomer database object with SHA256 hashed PII + + Returns: + dict: User data for Meta Conversions API + + """ + user_data = {} + if not hashed_customer: + return user_data + + # Map hashed customer fields to Meta CAPI field names + field_mapping = { + "hashed_email": "em", + "hashed_phone": "ph", + "hashed_given_name": "fn", + "hashed_surname": "ln", + "hashed_city": "ct", + "hashed_postal_code": "zp", + "hashed_country_code": "country", + "hashed_gender": "ge", + "hashed_birth_date": "db", + } + + for source_field, target_field in field_mapping.items(): + value = getattr(hashed_customer, source_field, None) + if value: + user_data[target_field] = value + + return user_data + + +def _build_custom_data(reservation, booking_value): + """Build custom_data dict from reservation information. + + Args: + reservation: Reservation database object + booking_value: Booking value in EUR + + Returns: + dict: Custom data for Meta Conversions API + + """ + custom_data = { + "currency": "EUR", + "value": booking_value, + "content_type": "hotel_booking", + } + + # Add optional reservation details + optional_fields = { + "hotel_code": "hotel_code", + "hotel_name": "hotel_name", + "num_adults": "num_adults", + "num_children": "num_children", + } + + for source_field, target_field in optional_fields.items(): + value = getattr(reservation, source_field, None) + if value: + custom_data[target_field] = value + + # Add date fields with ISO format + if reservation.start_date: + custom_data["checkin_date"] = reservation.start_date.isoformat() + if reservation.end_date: + custom_data["checkout_date"] = reservation.end_date.isoformat() + + return custom_data + + +def _add_utm_parameters(custom_data, reservation): + """Add UTM parameters to custom_data if available. + + Args: + custom_data: Custom data dict to modify + reservation: Reservation database object + + """ + utm_fields = ["utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content"] + if any(getattr(reservation, field, None) for field in utm_fields): + for field in utm_fields: + custom_data[field] = getattr(reservation, field, None) + + +def _format_fbc(fbclid, timestamp): + """Format Facebook Click ID (fbclid) to fbc parameter. + + The fbc format is: fb.{subdomain_index}.{timestamp_ms}.{fbclid} + + Args: + fbclid: Facebook Click ID from the ad URL + timestamp: DateTime object from reservation creation + + Returns: + str: Formatted fbc value for Meta Conversions API + + """ + # Extract timestamp in milliseconds + timestamp_ms = int(timestamp.timestamp() * 1000) + # Subdomain index is typically 1 + subdomain_index = 1 + return f"fb.{subdomain_index}.{timestamp_ms}.{fbclid}" + + +def create_meta_capi_event(reservation, customer, hashed_customer): + """Create a Meta Conversions API event from reservation and customer data. + + Args: + reservation: Reservation database object + customer: Customer database object (currently unused) + hashed_customer: HashedCustomer database object with SHA256 hashed PII + + Returns: + dict: Formatted event for Meta Conversions API + + """ + del customer # Currently unused but kept for API consistency + # Calculate booking value (example: random value between 500-2000 EUR) + booking_value = 1250.00 # Euro + + # Current timestamp + event_time = int(time.time()) + + # Build user_data with hashed customer information + user_data = _build_user_data(hashed_customer) + + # Add tracking parameters if available + if reservation.fbclid and reservation.created_at: + # Format fbclid as fbc parameter + user_data["fbc"] = _format_fbc(reservation.fbclid, reservation.created_at) + if reservation.gclid: + user_data["gclid"] = reservation.gclid + + # Build custom_data + custom_data = _build_custom_data(reservation, booking_value) + + # Add UTM parameters to event + _add_utm_parameters(custom_data, reservation) + + # Return the event + return { + "event_name": "Purchase", + "event_time": event_time, + "event_id": reservation.unique_id, # Unique event ID for deduplication + "event_source_url": "https://example.com/booking-confirmation", + "action_source": "website", + "user_data": user_data, + "custom_data": custom_data, + } + + +async def send_test_event(): + """Load data from DB and create test Meta CAPI event.""" + logger.info("Loading test data from database...") + + # Load data + test_data = await load_test_data_from_db() + + if not test_data: + logger.error("No test data available. Please add reservations to the database.") + return + + # Use the first reservation for testing + data = test_data[0] + reservation = data["reservation"] + customer = data["customer"] + hashed_customer = data["hashed_customer"] + + logger.info("Using reservation: %s", reservation.unique_id) + logger.info("Customer: %s %s", customer.given_name, customer.surname) + + # Create the event + event = create_meta_capi_event(reservation, customer, hashed_customer) + + # Create the full payload with test_event_code at top level + payload = { + "data": [event], + "test_event_code": TEST_CODE, + } + + # Log the event (pretty print) + separator = "=" * 80 + logger.info("\n%s", separator) + logger.info("META CONVERSIONS API EVENT") + logger.info("%s", separator) + logger.info("\nEndpoint: %s", CAPI_ENDPOINT) + logger.info("\nPayload:\n%s", json.dumps(payload, indent=2)) + logger.info("\n%s", separator) + + logger.info("\nNOTE: This is a test event. To actually send it:") + logger.info("1. Set PIXEL_ID to your Meta Pixel ID") + logger.info("2. Set ACCESS_TOKEN to your Meta access token") + logger.info("3. Uncomment the httpx.post() call below") + logger.info( + "4. Test the event at: https://developers.facebook.com/tools/events_manager/" + ) + logger.info(" Use test event code: %s", TEST_CODE) + + # Uncomment to actually send the event + + # async with httpx.AsyncClient() as client: + # response = await client.post( + # CAPI_ENDPOINT, + # json=payload, + # params={"access_token": ACCESS_TOKEN}, + # ) + # logger.info("Response status: %s", response.status_code) + # logger.info("Response body: %s", response.text) + + +if __name__ == "__main__": + asyncio.run(send_test_event()) diff --git a/test_handshake.py b/test_handshake.py deleted file mode 100644 index 4d7fb15..0000000 --- a/test_handshake.py +++ /dev/null @@ -1,29 +0,0 @@ -#!/usr/bin/env python3 -"""Test the handshake functionality with the real AlpineBits sample file.""" - -import asyncio - -from alpine_bits_python.alpinebits_server import AlpineBitsServer - - -async def main(): - - # Create server instance - server = AlpineBitsServer() - - # Read the sample handshake request - with open( - "AlpineBits-HotelData-2024-10/files/samples/Handshake/Handshake-OTA_PingRQ.xml", - ) as f: - ping_request_xml = f.read() - - - # Handle the ping request - await server.handle_request( - "OTA_Ping:Handshaking", ping_request_xml, "2024-10" - ) - - - -if __name__ == "__main__": - asyncio.run(main())