alteryx_smarty_batch_standardization.py

from ayx import Alteryx
import requests
import pandas as pd
import json
from pandas import json_normalize

# =============================================================================
# CONFIGURATION
# =============================================================================
SMARTY_API_URL = "https://us-street.api.smarty.com/street-address"
BATCH_SIZE = 100  # Maximum records per API request

# SSL Certificate Configuration
# Set to None to use default certificates
# Set to path string to use a custom certificate file (e.g., 'C:/certs/my_cert.pem')
# Set to False to disable SSL verification (not recommended for production)
CUSTOM_CERT_PATH = None

# Map your input field names here (adjust as needed)
INPUT_ID_FIELD = 'input_id'  # Unique identifier for tracking records
STREET_FIELD = 'street'
CITY_FIELD = 'city'
STATE_FIELD = 'state'
ZIP_FIELD = 'zipcode'
AUTH_ID_FIELD = 'auth_id'  # Field name containing Smarty Auth ID
AUTH_TOKEN_FIELD = 'auth_token'  # Field name containing Smarty Auth Token

# =============================================================================
# Main Processing Function
# =============================================================================

def standardize_addresses_batch(addresses, auth_id, auth_token):
    """
    Sends a batch of addresses to Smarty API and returns standardized components.
    
    Parameters:
    - addresses: List of dictionaries containing address data and input_id
    - auth_id: Smarty Auth ID
    - auth_token: Smarty Auth Token
    
    Returns:
    - List of dictionaries with standardized address fields mapped by input_id
    """
    # Build the batch request payload
    batch_payload = []
    for addr in addresses:
        batch_payload.append({
            'input_id': addr['input_id'],
            'street': addr['street'],
            'city': addr['city'],
            'state': addr['state'],
            'zipcode': addr['zipcode'],
            'match': 'invalid'  # Set match strategy per record
        })
    
    params = {
        'auth-id': auth_id,
        'auth-token': auth_token
    }
    
    try:
        # Determine SSL verification setting
        if CUSTOM_CERT_PATH is False:
            verify_setting = False
        elif CUSTOM_CERT_PATH is None:
            verify_setting = True
        else:
            verify_setting = CUSTOM_CERT_PATH
        
        response = requests.post(
            SMARTY_API_URL,
            params=params,
            json=batch_payload,
            headers={'Content-Type': 'application/json'},
            timeout=30,
            verify=verify_setting
        )
        response.raise_for_status()
        
        results_data = response.json()
        
        # Create a results dictionary indexed by input_id
        results_map = {}
        for result in results_data:
            input_id = result.get('input_id')
            if input_id is not None:
                # Flatten all nested fields from Smarty API
                flattened_result = flatten_dict(result)
                results_map[input_id] = flattened_result
        
        # Fill in results for all input addresses (including those with no match)
        all_results = []
        for addr in addresses:
            input_id = addr['input_id']
            if input_id in results_map:
                all_results.append(results_map[input_id])
            else:
                all_results.append({'validation_status': 'No Match Found'})
        
        return all_results
        
    except requests.exceptions.RequestException as e:
        # Return error status for all addresses in the batch
        return [{'validation_status': f'API Error: {str(e)}'} for _ in addresses]

def flatten_dict(d, parent_key='', sep='_'):
    """
    Flatten a nested dictionary structure.
    
    Parameters:
    - d: Dictionary to flatten
    - parent_key: Key prefix for nested items
    - sep: Separator between nested keys
    
    Returns:
    - Flattened dictionary
    """
    items = []
    for k, v in d.items():
        # Skip input_id from API response to avoid column conflicts
        if k == 'input_id':
            continue
            
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten_dict(v, new_key, sep=sep).items())
        elif isinstance(v, list):
            # Convert lists to JSON strings to preserve data
            items.append((new_key, json.dumps(v) if v else None))
        else:
            items.append((new_key, v))
    return dict(items)

def create_empty_result(status):
    """Creates an empty result dictionary with error status."""
    return {'validation_status': status}

def process_in_batches(df):
    """
    Process the dataframe in batches and return results.
    
    Parameters:
    - df: Input dataframe with address data
    
    Returns:
    - List of result dictionaries
    """
    all_results = []
    
    # Get credentials from first row (assuming they're the same for all rows)
    if len(df) == 0:
        return []
    
    first_row = df.iloc[0]
    auth_id = str(first_row.get(AUTH_ID_FIELD, '')) if pd.notna(first_row.get(AUTH_ID_FIELD)) else ''
    auth_token = str(first_row.get(AUTH_TOKEN_FIELD, '')) if pd.notna(first_row.get(AUTH_TOKEN_FIELD)) else ''
    
    # Check for missing credentials
    if not auth_id or not auth_token:
        return [create_empty_result('Missing Credentials') for _ in range(len(df))]
    
    # Process in batches
    for batch_start in range(0, len(df), BATCH_SIZE):
        batch_end = min(batch_start + BATCH_SIZE, len(df))
        batch_df = df.iloc[batch_start:batch_end]
        
        # Prepare batch addresses
        batch_addresses = []
        for index, row in batch_df.iterrows():
            street = str(row.get(STREET_FIELD, '')) if pd.notna(row.get(STREET_FIELD)) else ''
            city = str(row.get(CITY_FIELD, '')) if pd.notna(row.get(CITY_FIELD)) else ''
            state = str(row.get(STATE_FIELD, '')) if pd.notna(row.get(STATE_FIELD)) else ''
            zipcode = str(row.get(ZIP_FIELD, '')) if pd.notna(row.get(ZIP_FIELD)) else ''
            input_id = str(row.get(INPUT_ID_FIELD, '')) if pd.notna(row.get(INPUT_ID_FIELD)) else str(index)
            
            # Check if all address fields are empty
            if not any([street, city, state, zipcode]):
                batch_addresses.append({
                    'input_id': input_id,
                    'street': '',
                    'city': '',
                    'state': '',
                    'zipcode': '',
                    'is_empty': True
                })
            else:
                batch_addresses.append({
                    'input_id': input_id,
                    'street': street,
                    'city': city,
                    'state': state,
                    'zipcode': zipcode,
                    'is_empty': False
                })
        
        # Separate empty addresses from valid ones
        valid_addresses = []
        valid_indices = []
        empty_indices = []
        
        for idx, addr in enumerate(batch_addresses):
            if addr.get('is_empty', False):
                empty_indices.append(idx)
            else:
                valid_addresses.append(addr)
                valid_indices.append(idx)
        
        # Initialize batch results
        batch_results = [None] * len(batch_addresses)
        
        # Process valid addresses
        if valid_addresses:
            valid_results = standardize_addresses_batch(valid_addresses, auth_id, auth_token)
            for i, idx in enumerate(valid_indices):
                batch_results[idx] = valid_results[i]
        
        # Fill in empty address results
        for idx in empty_indices:
            batch_results[idx] = create_empty_result('Empty Input')
        
        all_results.extend(batch_results)
    
    return all_results

# =============================================================================
# Alteryx Input/Output Processing
# =============================================================================

# Read data from Alteryx input anchor
df = Alteryx.read("#1")

# Process all records in batches
results = process_in_batches(df)

# Create results dataframe
results_df = pd.DataFrame(results)

# Combine original data with standardized results
output_df = pd.concat([df.reset_index(drop=True), results_df], axis=1)

# Write output to Alteryx
Alteryx.write(output_df, 1)