Programmatically validate multiple phone numbers using Pinpoint Phone Number Validate service

Below is a sample Python script that reads phone numbers from a CSV file, validates them using the AWS Pinpoint Phone Number Validation service, and prints out the phone number details such as phone number, number type, cleansed phone number (E.164 format), carrier and country.

Before you use this script, make sure that you have a valid AWS account and the Lambda function execution role has the necessary permissions. The policy that needs to be added might look like this:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "mobiletargeting:PhoneNumberValidate"
            ],
            "Resource": "*"
        }
    ]
}

Python Script

import json
import boto3
import csv

pinpoint_client = boto3.client('pinpoint', region_name='us-east-1')  # Change region as per your setup

def lambda_handler(event, context):
    # Read the CSV file from the event (replace with actual input if using S3 or other input source)
    csv_file_content = event['csv_data']  # The input which is expected to contain the CSV data with phone numbers
    
    # Process the CSV
    phone_numbers = read_phone_numbers_from_csv(csv_file_content)
    
    # List to store results
    validation_results = []

    # Loop through each phone number and invoke Pinpoint API for validation
    for phone_number in phone_numbers:
        validation_result = validate_phone_number(phone_number)
        if validation_result:
            validation_results.append(validation_result)
    
    # Output the validation results
    return {
        'statusCode': 200,
        'body': json.dumps(validation_results)
    }

def read_phone_numbers_from_csv(csv_content):
    """
    Reads phone numbers from a CSV string.
    Expects the CSV file to have a column "PhoneNumber".
    """
    phone_numbers = []
    csv_reader = csv.DictReader(csv_content.splitlines())
    
    for row in csv_reader:
        phone_numbers.append(row['PhoneNumber'])
    
    return phone_numbers

def validate_phone_number(phone_number):
    """
    Calls the Pinpoint Phone Number Validate API for the given phone number.
    """
    try:
        response = pinpoint_client.phone_number_validate(
            NumberValidateRequest={
                'PhoneNumber': phone_number
            }
        )
        # Extract needed information from the response
        phone_type = response['NumberValidateResponse'].get('PhoneType', 'Unknown')
        cleansed_e164 = response['NumberValidateResponse'].get('CleansedPhoneNumberE164', 'N/A')
        carrier = response['NumberValidateResponse'].get('Carrier', 'Unknown')
        country = response['NumberValidateResponse'].get('Country', 'Unknown')
        
        # Return validation details
        return {
            'OriginalPhoneNumber': phone_number,
            'PhoneType': phone_type,
            'CleansedPhoneNumberE164': cleansed_e164,
            'Carrier': carrier,
            'Country': country
        }
    
    except Exception as e:
        print(f"Error validating phone number {phone_number}: {str(e)}")
        return None

Event Input Example:

For testing, you can trigger the Lambda function with an event that contains the CSV data as a string, as shown below:

{ "csv_data": "PhoneNumber\n+14694165374\n+61293744000\n+447911123456" }

Notes:

1. CSV Source: If the CSV is stored in S3, you can modify the code to pull the CSV from S3 instead of passing it in the event.
2. Region: Update the region_name in the boto3.client function to your relevant AWS region.
3. Error Handling: Add more sophisticated error handling as needed, especially for large datasets or different API response failures.