Fix Klaviyo Email Marketing subscription status for users, after accidentally enabling Email Marketing on list import

fix_klaviyo_profile_marketing_status.py

# I accidentally resubscribed all users during import to a list in Klaviyo.
# This script will find the subscription status prior to that change and create a suppresion list.
#
# Description: This script will fetch all profiles from a list, then fetch all events for each profile.
# It will then filter out the latest event for each profile and determine if the user is subscribed or unsubscribed.
# It will then write the results to a file and extract emails of users who have unsubscribed.

import json

import requests

API_KEY = "pk_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"  # Your Klaviyo API key
LIST_ID = "LdsIs"  # The ID of the list you imported into


headers = {
    "accept": "application/json",
    "revision": "2024-07-15",
    "Authorization": f"Klaviyo-API-Key {API_KEY}",
}


def get_all_metrics():
    # https://developers.klaviyo.com/en/reference/get_metrics
    metrics = []
    url = 'https://a.klaviyo.com/api/metrics?filter=equals(integration.name,"Klaviyo")'
    while url:
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
            data = response.json()
            metrics.extend(data["data"])
            url = data.get("links", {}).get("next")
        except Exception as e:
            print(e)
            break

    with open("klaviyo_metrics.json", "w") as f:
        f.write(json.dumps(metrics, indent=4))

    # find the metric_id for "Subscribed to Email Marketing", "Unsubscribed from Email Marketing", "Manually Suppressed from Email Marketing"
    unsubscribe_metric_id = None
    subscribe_metric_id = None
    manually_suppressed_metric_id = None
    for metric in metrics:
        metric_name = metric["attributes"]["name"]
        if metric_name == "Subscribed to Email Marketing":
            subscribe_metric_id = metric["id"]

        elif metric_name == "Unsubscribed from Email Marketing":
            unsubscribe_metric_id = metric["id"]

        elif metric_name == "Manually Suppressed from Email Marketing":
            manually_suppressed_metric_id = metric["id"]

    return subscribe_metric_id, unsubscribe_metric_id, manually_suppressed_metric_id


def get_profiles_for_list(list_id):
    # https://developers.klaviyo.com/en/reference/get_list_profiles

    profiles = []
    url = f"https://a.klaviyo.com/api/lists/{list_id}/profiles/?page[size]=100"
    while url:
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
            data = response.json()
            profiles.extend(data["data"])
            url = data.get("links", {}).get("next")
        except Exception as e:
            print(e)
            break

    with open("klaviyo_profiles.json", "w") as f:
        f.write(json.dumps(profiles, indent=4))


def get_events_for_profiles(metric_id):

    # iterate through profiles loaded from klaviyo_profiles.json. for each user, get their events and store them in a list
    with open("klaviyo_profiles.json", "r") as f:
        profiles = json.load(f)

    events = []
    for index, profile in enumerate(profiles):
        profile_id = profile["id"]

        url = f'https://a.klaviyo.com/api/events/?filter=equals(profile_id,"{profile_id}"),equals(metric_id,"{metric_id}")&page[size]=100'
        while url:
            try:
                print(
                    f"fetching events for profile {profile_id} at index {index} of {len(profiles)}"
                )

                response = requests.get(url, headers=headers)
                response.raise_for_status()
                data = response.json()
                events.extend(data["data"])
                url = data.get("links", {}).get("next")
            except Exception as e:
                print(e)
                break

    # write events to a file
    with open(f"klaviyo_events_{metric_id}.json", "w") as f:
        f.write(json.dumps(events, indent=4))


def analyse_events(
    subscribe_metric_id, unsubscribe_metric_id, manually_suppressed_metric_id, list_id
):

    # load profiles from klaviyo_profiles.json
    with open("klaviyo_profiles.json", "r") as f:
        profiles = json.load(f)

    # load events from klaviyo_events_ddd.json and klaviyo_events_aaa.json, then combine all events into a single list
    with open(f"klaviyo_events_{unsubscribe_metric_id}.json", "r") as f:
        events_unsubscribed = json.load(f)

    with open(f"klaviyo_events_{subscribe_metric_id}.json", "r") as f:
        events_subscribed = json.load(f)

    with open(f"klaviyo_events_{manually_suppressed_metric_id}.json", "r") as f:
        events_manually_suppressed = json.load(f)

    all_events = events_unsubscribed + events_subscribed + events_manually_suppressed

    # remove all events with attributes.event_properties.method == "LIST_IMPORT"
    all_events = [
        event
        for event in all_events
        if (
            event.get("attributes", {}).get("event_properties", {}).get("method")
            != "LIST_IMPORT"
            and event.get("attributes", {}).get("event_properties", {}).get("list_id")
            != list_id
        )
    ]

    # sort events by profile id and timestamp
    all_events.sort(
        key=lambda x: (
            x["relationships"]["profile"]["data"]["id"],
            x["attributes"]["timestamp"],
        )
    )

    # for each profile id, use attributes.timestamp to filter out the latest event, remove the older ones
    latest_profile_events = {}
    for event in all_events:
        profile_id = event["relationships"]["profile"]["data"]["id"]
        timestamp = event["attributes"]["timestamp"]
        if profile_id not in latest_profile_events:
            latest_profile_events[profile_id] = event
        elif latest_profile_events[profile_id]["attributes"]["timestamp"] < timestamp:
            latest_profile_events[profile_id] = event

    # for each entry in latest_profile_events, check "relationships"."metric"."id" against subscribe_metric_id and unsubscribe_metric_id to determine if the user is subscribed or unsubscribed. store result in new variable
    for profile_id, event in latest_profile_events.items():
        metric_id = event["relationships"]["metric"]["data"]["id"]
        if metric_id == subscribe_metric_id:
            latest_profile_events[profile_id]["subscribed"] = True
        elif metric_id == unsubscribe_metric_id:
            latest_profile_events[profile_id]["subscribed"] = False

    # macth profile_id with email from klaivyo_profiles.json
    for profile in profiles:
        profile_id = profile["id"]
        email = profile["attributes"]["email"]
        if profile_id in latest_profile_events:
            latest_profile_events[profile_id]["email"] = email

    # write sorted events to a file
    with open("klaviyo_events_filtered.json", "w") as f:
        f.write(json.dumps(latest_profile_events, indent=4))

    # extract emails of users who have unsubscribed. one email per line. export as suppressed_emails.csv
    with open("suppressed_emails.csv", "w") as f:
        for profile_id, event in latest_profile_events.items():
            if not event["subscribed"]:
                f.write(f"{event['email']}\n")


get_profiles_for_list(LIST_ID)

subscribe_metric_id, unsubscribe_metric_id, manually_suppressed_metric_id = (
    get_all_metrics()
)

get_events_for_profiles(unsubscribe_metric_id)
get_events_for_profiles(subscribe_metric_id)
get_events_for_profiles(manually_suppressed_metric_id)

analyse_events(
    subscribe_metric_id, unsubscribe_metric_id, manually_suppressed_metric_id, LIST_ID
)