Added capi test. Data is received in facebook
This commit is contained in:
Jonas Linter
291
test_capi.py
Normal file
291
test_capi.py
Normal file
@@ -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())
|
||||
Reference in New Issue
Block a user