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

Added account_ids to the config

Browse Source
This commit is contained in:
Jonas Linter
2025-10-22 17:32:28 +02:00
parent 81074d839a
commit 90d79a71fb
9 changed files with 298 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
config/config.yaml
View File

@@ -23,20 +23,28 @@ alpine_bits_auth:
hotel_name: "Bemelmans Post"
username: "bemelman"
password: !secret BEMELMANS_PASSWORD
meta_account: null # Optional: Meta advertising account ID
google_account: null # Optional: Google advertising account ID
- hotel_id: "135"
hotel_name: "Testhotel"
username: "sebastian"
password: !secret BOB_PASSWORD
meta_account: null # Optional: Meta advertising account ID
google_account: null # Optional: Google advertising account ID
- hotel_id: "39052_001"
hotel_name: "Jagthof Kaltern"
username: "jagthof"
password: !secret JAGTHOF_PASSWORD
meta_account: null # Optional: Meta advertising account ID
google_account: null # Optional: Google advertising account ID
- hotel_id: "39040_001"
hotel_name: "Residence Erika"
username: "erika"
password: !secret ERIKA_PASSWORD
meta_account: null # Optional: Meta advertising account ID
google_account: null # Optional: Google advertising account ID
api_tokens:
- tLTI8wXF1OVEvUX7kdZRhSW3Qr5feBCz0mHo-kbnEp0

129
src/alpine_bits_python/api.py
View File

@@ -38,7 +38,7 @@ from .alpinebits_server import (
)
from .auth import generate_unique_id, validate_api_key
from .config_loader import load_config
from .const import HttpStatusCode
from .const import CONF_GOOGLE_ACCOUNT, CONF_HOTEL_ID, CONF_META_ACCOUNT, HttpStatusCode
from .conversion_service import ConversionService
from .customer_service import CustomerService
from .db import Base, get_database_url
@@ -71,85 +71,47 @@ security_bearer = HTTPBearer()
# Constants for token sanitization
TOKEN_LOG_LENGTH = 10
# Country name to ISO 3166-1 alpha-2 code mapping
COUNTRY_NAME_TO_CODE = {
# English names
"germany": "DE",
"italy": "IT",
"austria": "AT",
"switzerland": "CH",
"france": "FR",
"netherlands": "NL",
"belgium": "BE",
"spain": "ES",
"portugal": "PT",
"united kingdom": "GB",
"uk": "GB",
"czech republic": "CZ",
"poland": "PL",
"hungary": "HU",
"croatia": "HR",
"slovenia": "SI",
# German names
"deutschland": "DE",
"italien": "IT",
"österreich": "AT",
"schweiz": "CH",
"frankreich": "FR",
"niederlande": "NL",
"belgien": "BE",
"spanien": "ES",
"vereinigtes königreich": "GB",
"tschechien": "CZ",
"polen": "PL",
"ungarn": "HU",
"kroatien": "HR",
"slowenien": "SI",
# Italian names
"germania": "DE",
"italia": "IT",
"svizzera": "CH",
"francia": "FR",
"paesi bassi": "NL",
"belgio": "BE",
"spagna": "ES",
"portogallo": "PT",
"regno unito": "GB",
"repubblica ceca": "CZ",
"polonia": "PL",
"ungheria": "HU",
"croazia": "HR",
}
def normalize_country_input(country_input: str | None) -> str | None:
"""Normalize country input to ISO 3166-1 alpha-2 code.
Handles:
- Country names in English, German, and Italian
- Already valid 2-letter codes (case-insensitive)
- None/empty values
def get_advertising_account_ids(
config: dict[str, Any],
hotel_code: str,
fbclid: str | None,
gclid: str | None
) -> tuple[str | None, str | None]:
"""Get advertising account IDs based on hotel config and click IDs.
Args:
country_input: Country name or code (case-insensitive)
config: Application configuration dict
hotel_code: Hotel identifier to look up in config
fbclid: Facebook click ID (if present, meta_account_id will be returned)
gclid: Google click ID (if present, google_account_id will be returned)
Returns:
2-letter ISO country code (uppercase) or None if input is None/empty
Tuple of (meta_account_id, google_account_id) based on conditional logic:
- meta_account_id is set only if fbclid is present AND hotel has
meta_account configured
- google_account_id is set only if gclid is present AND hotel has
google_account configured
"""
if not country_input:
return None
meta_account_id = None
google_account_id = None
country_input = country_input.strip()
# Look up hotel in config
alpine_bits_auth = config.get("alpine_bits_auth", [])
for hotel in alpine_bits_auth:
if hotel.get(CONF_HOTEL_ID) == hotel_code:
# Conditionally set meta_account_id if fbclid is present
if fbclid:
meta_account_id = hotel.get(CONF_META_ACCOUNT)
# If already 2 letters, assume it's a country code (ISO 3166-1 alpha-2)
iso_country_code_length = 2
if len(country_input) == iso_country_code_length and country_input.isalpha():
return country_input.upper()
# Conditionally set google_account_id if gclid is present
if gclid:
google_account_id = hotel.get(CONF_GOOGLE_ACCOUNT)
# Try to match as country name (case-insensitive)
country_lower = country_input.lower()
return COUNTRY_NAME_TO_CODE.get(country_lower, country_input)
break
return meta_account_id, google_account_id
# Pydantic models for language detection
@@ -369,7 +331,7 @@ async def lifespan(app: FastAPI):
# Run migrations after tables exist (only primary worker for race conditions)
if is_primary:
await run_all_migrations(engine)
await run_all_migrations(engine, config)
else:
_LOGGER.info("Skipping migrations (non-primary worker)")
@@ -690,6 +652,15 @@ async def process_wix_form_submission(request: Request, data: dict[str, Any], db
_LOGGER.exception("Error parsing submissionTime: %s", e)
submissionTime = None
# Extract fbclid and gclid for conditional account ID lookup
fbclid = data.get("field:fbclid")
gclid = data.get("field:gclid")
# Get advertising account IDs conditionally based on fbclid/gclid presence
meta_account_id, google_account_id = get_advertising_account_ids(
request.app.state.config, hotel_code, fbclid, gclid
)
reservation = ReservationData(
unique_id=unique_id,
start_date=date.fromisoformat(start_date),
@@ -707,8 +678,10 @@ async def process_wix_form_submission(request: Request, data: dict[str, Any], db
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"),
fbclid=fbclid,
gclid=gclid,
meta_account_id=meta_account_id,
google_account_id=google_account_id,
)
if reservation.md5_unique_id is None:
@@ -818,9 +791,6 @@ async def process_generic_webhook_submission(
city = form_data.get("stadt", "")
country = form_data.get("land", "")
# Normalize country input (convert names to codes, handle case)
country = normalize_country_input(country)
# Parse dates - handle DD.MM.YYYY format
start_date_str = form_data.get("anreise")
end_date_str = form_data.get("abreise")
@@ -912,6 +882,11 @@ async def process_generic_webhook_submission(
# Create/update customer
db_customer = await customer_service.get_or_create_customer(customer_data)
# Get advertising account IDs conditionally based on fbclid/gclid presence
meta_account_id, google_account_id = get_advertising_account_ids(
request.app.state.config, hotel_code, fbclid, gclid
)
# Create reservation
reservation_kwargs = {
"unique_id": unique_id,
@@ -931,6 +906,8 @@ async def process_generic_webhook_submission(
"user_comment": user_comment,
"fbclid": fbclid,
"gclid": gclid,
"meta_account_id": meta_account_id,
"google_account_id": google_account_id,
}
# Only include created_at if we have a valid submission_time

4
src/alpine_bits_python/config_loader.py
View File

@@ -19,11 +19,13 @@ from voluptuous import (
from alpine_bits_python.const import (
CONF_ALPINE_BITS_AUTH,
CONF_DATABASE,
CONF_GOOGLE_ACCOUNT,
CONF_HOTEL_ID,
CONF_HOTEL_NAME,
CONF_LOGGING,
CONF_LOGGING_FILE,
CONF_LOGGING_LEVEL,
CONF_META_ACCOUNT,
CONF_PASSWORD,
CONF_PUSH_ENDPOINT,
CONF_PUSH_TOKEN,
@@ -74,6 +76,8 @@ hotel_auth_schema = Schema(
Required(CONF_HOTEL_NAME): str,
Required(CONF_USERNAME): str,
Required(CONF_PASSWORD): str,
Optional(CONF_META_ACCOUNT): str,
Optional(CONF_GOOGLE_ACCOUNT): str,
Optional(CONF_PUSH_ENDPOINT): {
Required(CONF_PUSH_URL): str,
Required(CONF_PUSH_TOKEN): str,

2
src/alpine_bits_python/const.py
View File

@@ -37,6 +37,8 @@ CONF_HOTEL_ID: Final[str] = "hotel_id"
CONF_HOTEL_NAME: Final[str] = "hotel_name"
CONF_USERNAME: Final[str] = "username"
CONF_PASSWORD: Final[str] = "password"
CONF_META_ACCOUNT: Final[str] = "meta_account"
CONF_GOOGLE_ACCOUNT: Final[str] = "google_account"
CONF_PUSH_ENDPOINT: Final[str] = "push_endpoint"
CONF_PUSH_URL: Final[str] = "url"
CONF_PUSH_TOKEN: Final[str] = "token"

3
src/alpine_bits_python/db.py
View File

@@ -124,6 +124,9 @@ class Reservation(Base):
user_comment = Column(String)
fbclid = Column(String)
gclid = Column(String)
# Advertising account IDs (stored conditionally based on fbclid/gclid presence)
meta_account_id = Column(String)
google_account_id = Column(String)
# Add hotel_code and hotel_name for XML
hotel_code = Column(String)
hotel_name = Column(String)

134
src/alpine_bits_python/migrations.py
View File

@@ -4,9 +4,12 @@ This module contains migration functions that are automatically run at app start
to update existing database schemas without losing data.
"""
from typing import Any
from sqlalchemy import inspect, text
from sqlalchemy.ext.asyncio import AsyncEngine
from .const import CONF_GOOGLE_ACCOUNT, CONF_HOTEL_ID, CONF_META_ACCOUNT
from .logging_config import get_logger
_LOGGER = get_logger(__name__)
@@ -96,17 +99,146 @@ async def migrate_add_room_types(engine: AsyncEngine) -> None:
_LOGGER.info("Migration add_room_types: No changes needed (already applied)")
async def run_all_migrations(engine: AsyncEngine) -> None:
async def migrate_add_advertising_account_ids(engine: AsyncEngine, config: dict[str, Any] | None = None) -> None:
"""Migration: Add advertising account ID fields to reservations table.
This migration adds two optional fields:
- meta_account_id: String (Meta/Facebook advertising account ID)
- google_account_id: String (Google advertising account ID)
These fields are populated conditionally based on fbclid/gclid presence.
For existing reservations, backfills account IDs from config based on hotel_code and fbclid/gclid.
Safe to run multiple times - will skip if columns already exist.
Args:
engine: SQLAlchemy async engine
config: Application configuration dict containing hotel account IDs
"""
_LOGGER.info("Running migration: add_advertising_account_ids")
added_count = 0
# Add each column if it doesn't exist
if await add_column_if_not_exists(engine, "reservations", "meta_account_id", "VARCHAR"):
added_count += 1
if await add_column_if_not_exists(engine, "reservations", "google_account_id", "VARCHAR"):
added_count += 1
if added_count > 0:
_LOGGER.info("Migration add_advertising_account_ids: Added %d columns", added_count)
else:
_LOGGER.info("Migration add_advertising_account_ids: Columns already exist")
# Backfill existing reservations with account IDs based on config and fbclid/gclid presence
if config:
await _backfill_advertising_account_ids(engine, config)
else:
_LOGGER.warning("No config provided, skipping backfill of advertising account IDs")
async def _backfill_advertising_account_ids(engine: AsyncEngine, config: dict[str, Any]) -> None:
"""Backfill advertising account IDs for existing reservations.
Updates existing reservations to populate meta_account_id and google_account_id
based on the conditional logic:
- If fbclid is present, set meta_account_id from hotel config
- If gclid is present, set google_account_id from hotel config
Args:
engine: SQLAlchemy async engine
config: Application configuration dict
"""
_LOGGER.info("Backfilling advertising account IDs for existing reservations...")
# Build a mapping of hotel_id -> account IDs from config
hotel_accounts = {}
alpine_bits_auth = config.get("alpine_bits_auth", [])
for hotel in alpine_bits_auth:
hotel_id = hotel.get(CONF_HOTEL_ID)
meta_account = hotel.get(CONF_META_ACCOUNT)
google_account = hotel.get(CONF_GOOGLE_ACCOUNT)
if hotel_id:
hotel_accounts[hotel_id] = {
"meta_account": meta_account,
"google_account": google_account
}
if not hotel_accounts:
_LOGGER.info("No hotel accounts found in config, skipping backfill")
return
_LOGGER.info("Found %d hotel(s) with account configurations", len(hotel_accounts))
# Update reservations with meta_account_id where fbclid is present
meta_updated = 0
for hotel_id, accounts in hotel_accounts.items():
if accounts["meta_account"]:
async with engine.begin() as conn:
sql = text(
"UPDATE reservations "
"SET meta_account_id = :meta_account "
"WHERE hotel_code = :hotel_id "
"AND fbclid IS NOT NULL "
"AND fbclid != '' "
"AND (meta_account_id IS NULL OR meta_account_id = '')"
)
result = await conn.execute(
sql,
{"meta_account": accounts["meta_account"], "hotel_id": hotel_id}
)
count = result.rowcount
if count > 0:
_LOGGER.info("Updated %d reservations with meta_account_id for hotel %s", count, hotel_id)
meta_updated += count
# Update reservations with google_account_id where gclid is present
google_updated = 0
for hotel_id, accounts in hotel_accounts.items():
if accounts["google_account"]:
async with engine.begin() as conn:
sql = text(
"UPDATE reservations "
"SET google_account_id = :google_account "
"WHERE hotel_code = :hotel_id "
"AND gclid IS NOT NULL "
"AND gclid != '' "
"AND (google_account_id IS NULL OR google_account_id = '')"
)
result = await conn.execute(
sql,
{"google_account": accounts["google_account"], "hotel_id": hotel_id}
)
count = result.rowcount
if count > 0:
_LOGGER.info("Updated %d reservations with google_account_id for hotel %s", count, hotel_id)
google_updated += count
_LOGGER.info(
"Backfill complete: %d reservations updated with meta_account_id, %d with google_account_id",
meta_updated,
google_updated
)
async def run_all_migrations(engine: AsyncEngine, config: dict[str, Any] | None = None) -> None:
"""Run all pending migrations.
This function should be called at app startup, after Base.metadata.create_all.
Each migration function should be idempotent (safe to run multiple times).
Args:
engine: SQLAlchemy async engine
config: Application configuration dict (optional, but required for some migrations)
"""
_LOGGER.info("Starting database migrations...")
try:
# Add new migrations here in chronological order
await migrate_add_room_types(engine)
await migrate_add_advertising_account_ids(engine, config)
_LOGGER.info("Database migrations completed successfully")

93
src/alpine_bits_python/schemas.py
View File

@@ -16,6 +16,57 @@ from enum import Enum
from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator
# Country name to ISO 3166-1 alpha-2 code mapping
COUNTRY_NAME_TO_CODE = {
# English names
"germany": "DE",
"italy": "IT",
"austria": "AT",
"switzerland": "CH",
"france": "FR",
"netherlands": "NL",
"belgium": "BE",
"spain": "ES",
"portugal": "PT",
"united kingdom": "GB",
"uk": "GB",
"czech republic": "CZ",
"poland": "PL",
"hungary": "HU",
"croatia": "HR",
"slovenia": "SI",
# German names
"deutschland": "DE",
"italien": "IT",
"österreich": "AT",
"schweiz": "CH",
"frankreich": "FR",
"niederlande": "NL",
"belgien": "BE",
"spanien": "ES",
"vereinigtes königreich": "GB",
"tschechien": "CZ",
"polen": "PL",
"ungarn": "HU",
"kroatien": "HR",
"slowenien": "SI",
# Italian names
"germania": "DE",
"italia": "IT",
"svizzera": "CH",
"francia": "FR",
"paesi bassi": "NL",
"belgio": "BE",
"spagna": "ES",
"portogallo": "PT",
"regno unito": "GB",
"repubblica ceca": "CZ",
"polonia": "PL",
"ungheria": "HU",
"croazia": "HR",
}
# phonetechtype enum 1,3,5 voice, fax, mobile
class PhoneTechType(Enum):
VOICE = "1"
@@ -53,6 +104,9 @@ class ReservationData(BaseModel):
user_comment: str | None = Field(None, max_length=2000)
fbclid: str | None = Field(None, max_length=300)
gclid: str | None = Field(None, max_length=300)
# Advertising account IDs (populated conditionally based on fbclid/gclid)
meta_account_id: str | None = Field(None, max_length=200)
google_account_id: str | None = Field(None, max_length=200)
utm_source: str | None = Field(None, max_length=150)
utm_medium: str | None = Field(None, max_length=150)
utm_campaign: str | None = Field(None, max_length=150)
@@ -120,26 +174,41 @@ class CustomerData(BaseModel):
@field_validator("country_code", mode="before")
@classmethod
def normalize_country_code(cls, v: str | None) -> str | None:
"""Normalize country code to uppercase and validate format.
"""Normalize country input to ISO 3166-1 alpha-2 code.
Handles:
- Country names in English, German, and Italian
- Already valid 2-letter codes (case-insensitive)
- None/empty values
Runs in 'before' mode to normalize before other validations.
Accepts 2-letter country codes (case-insensitive) and normalizes
to uppercase ISO 3166-1 alpha-2 format.
This ensures that old data saved incorrectly in the database is
transformed into the correct format when retrieved, and that new
data is always normalized regardless of the source.
Args:
v: Country name or code (case-insensitive)
Returns:
2-letter ISO country code (uppercase) or None if input is None/empty
"""
if v is None or v == "":
if not v:
return None
# Convert to string and strip whitespace
v = str(v).strip().upper()
country_input = str(v).strip()
# Validate it's exactly 2 letters
if len(v) != 2 or not v.isalpha():
raise ValueError(
f"Country code must be exactly 2 letters (ISO 3166-1 alpha-2), "
f"got '{v}'"
)
if not country_input:
return None
return v
# If already 2 letters, assume it's a country code (ISO 3166-1 alpha-2)
iso_country_code_length = 2
if len(country_input) == iso_country_code_length and country_input.isalpha():
return country_input.upper()
# Try to match as country name (case-insensitive)
country_lower = country_input.lower()
return COUNTRY_NAME_TO_CODE.get(country_lower, country_input)
@field_validator("language")
@classmethod

2
tests/test_api.py
View File

@@ -131,7 +131,7 @@ def sample_wix_form_data():
"contactId": f"contact-{unique_id}",
},
"field:anrede": "Mr.",
"field:form_field_5a7b": "Checked",
"field:form_field_5a7b": True,
"field:date_picker_a7c8": "2024-12-25",
"field:date_picker_7e65": "2024-12-31",
"field:number_7cf5": "2",

23
tests/test_customer_service.py
View File

@@ -249,27 +249,30 @@ async def test_hash_existing_customers_normalizes_country_code(
hashed = result.scalar_one_or_none()
assert hashed is None
# Verify the customer has the invalid country code
# Verify the customer has the invalid country code stored in the DB
assert customer.country_code == "Italy"
# Run hash_existing_customers - this should fail to validate "Italy"
# because it's not a 2-letter code, so the customer should be skipped
# Run hash_existing_customers - this should normalize "Italy" to "IT"
# during validation and successfully create a hashed customer
service = CustomerService(async_session)
count = await service.hash_existing_customers()
# Should skip this customer due to validation error
assert count == 0
# Should successfully hash this customer (country code normalized during validation)
assert count == 1
# Verify hashed version was NOT created
# Verify hashed version was created
await async_session.refresh(customer)
result = await async_session.execute(
select(HashedCustomer).where(HashedCustomer.customer_id == customer.id)
)
hashed = result.scalar_one_or_none()
assert hashed is None
assert hashed is not None
# The customer's country_code should still be "Italy" (unchanged)
assert customer.country_code == "Italy"
# The hashed customer should have the hashed version of normalized country code "IT"
# "IT" -> lowercase "it" -> sha256 hash
expected_hash = "2ad8a7049d7c5511ac254f5f51fe70a046ebd884729056f0fe57f5160d467153"
assert hashed.hashed_country_code == expected_hash
# Note: The original customer's country_code in DB remains "Italy" unchanged
# Now let's test the case where we have a valid 2-letter code
# but in the wrong case (should be normalized to uppercase)
@@ -290,7 +293,7 @@ async def test_hash_existing_customers_normalizes_country_code(
# Run hash_existing_customers again
count = await service.hash_existing_customers()
# Should hash this customer (country code will be normalized)
# Should hash only the second customer (first one already hashed)
assert count == 1
# Verify the customer's country_code was normalized to uppercase