Import_MailChimp_Engagments_to_MixPanel.py

import base64
import urllib.request
import json
from datetime import datetime, timedelta
import re

import mailchimp
import mixpanel
import requests

MP_API_SECRET = b"replacemen"
MP_TOKEN = 'replaceme'
MP_API_KEY = 'replaceme'
MC_TIME_ZONE = 6

MC_API_KEY = 'replaceme'

class CustomMixPanel (mixpanel.Mixpanel):
    EXPORT_ENDPOINT = 'https://data.mixpanel.com/api'
    VERSION = '2.0'

    def __init__(self, token, api_key):
        self._api_key = api_key
        super().__init__(token)

    def track_old(self, distinct_id, event_name, epoch = None, properties=None, meta=None):
        """Record an event.

        :param str distinct_id: identifies the user triggering the event
        :param str event_name: a name describing the event
        :param int epoch: unix timpstamp of prior event
        :param dict properties: additional data to record; keys should be
            strings, and values should be strings, numbers, or booleans
        :param dict meta: overrides Mixpanel special properties

        ``properties`` should describe the circumstances of the event, or
        aspects of the source or user associated with it. ``meta`` is used
        (rarely) to override special values sent in the event object.
        """

        if epoch == None:
            epoch = int(self._now())

        all_properties = {
            'token': self._token,
            'distinct_id': distinct_id,
            'time': epoch,
            'mp_lib': 'python',
            '$lib_version':  mixpanel.__version__,
        }
        if properties:
            all_properties.update(properties)
        event = {
            'event': event_name,
            'properties': all_properties,
        }
        if meta:
            event.update(meta)

        log_date = datetime.fromtimestamp(epoch)
        now_date = datetime.fromtimestamp(int(self._now()))

        #Use regular MP import
        if (now_date - log_date).days >= 5:
            self._consumer.send('imports', mixpanel.json_dumps(event, cls=self._serializer), self._api_key)
        else:
            self._consumer.send('events', mixpanel.json_dumps(event, cls=self._serializer))

    def export(self, encoded_secret, params, http_method='GET', format='json'):

        request_url = '/'.join([self.EXPORT_ENDPOINT, str(self.VERSION)] + ['export'])
        if http_method == 'GET':
            data = None
            request_url = request_url + '/?' + self.unicode_urlencode(params)
        else:
            data = self.unicode_urlencode(params)

        auth = base64.b64encode(encoded_secret).decode("ascii")
        headers = {'Authorization': 'Basic {encoded_secret}'.format(encoded_secret=auth)}

        request = urllib.request.Request(request_url, data, headers)
        response = urllib.request.urlopen(request, timeout=600)
        str_response = response.read().decode('utf8')
        lines = str_response.splitlines(True)
        records = []
        for line in lines:
            obj = json.loads(line)
            records.append(obj)
        return records

    def unicode_urlencode(self, pms):
        """
            Convert lists to JSON encoded strings, and correctly handle any
            unicode URL parameters.
        """
        if isinstance(pms, dict):
            pms = list(pms.items())
        for i,param in enumerate(pms):
            if isinstance(param[1], list):
                pms.remove(param)
                pms.append ((param[0], json.dumps(param[1]),))

        return urllib.parse.urlencode(
            [(k, v) for k, v in pms]
        )

class CustomMailChimp(mailchimp.Mailchimp):

    def get_all_activity_for_camgaign(self, campaign_id):
        url = self.root.replace('2.0/', 'export/1.0/campaignSubscriberActivity/')
        params = json.dumps({'apikey': MC_API_KEY,'id':campaign_id})
        r = self.session.post(url, data=params, headers={'content-type': 'application/json'})

        if r.status_code != requests.codes.ok:
            raise self.cast_error(r.text)

        result = []
        if r.text != ' ':
            for line in r.text.splitlines():
                result.append(json.loads(line))

        return result

def run_import (mp_from_date, mp_to_date):

    mc = CustomMailChimp(MC_API_KEY)

    mp = CustomMixPanel(MP_TOKEN, MP_API_KEY)

    lists = mc.campaigns.list(filters={'status':'sent'}, limit=100)

    params = {
            'event': ['MailChimp Open', 'MailChimp Click'],
            'to_date': mp_to_date,
            'from_date': mp_from_date
        }

    print ("Loading existing events")

    existing_events = mp.export(MP_API_SECRET, params)

    existing_event_hash = {}

    print ("{0} events found".format(len(existing_events)))

    for event in existing_events:
        key = "email: {0} | epoch {1}".format(event["properties"]['distinct_id'], event["properties"]['epoch'])
        existing_event_hash[key] = event

    for alist in lists['data']:

        campaign_id = alist['id']

        subject = alist['subject']

        title = alist['title']

        archive_url = alist['archive_url']

        campaign_activity = mc.get_all_activity_for_camgaign(campaign_id)

        print ("Processing: {0} events from {1}".format(len(campaign_activity),title))

        for line in campaign_activity:
            email = [key for key in line.keys()][0]
            for engagement in line[email]:
                dt = datetime.strptime(engagement['timestamp'], '%Y-%m-%d %H:%M:%S')
                #MailChimp should send an epoch, but instead sends a timestamp that looks 'off' by the
                #Time zone specified in https://us12.admin.mailchimp.com/account/details/.
                dt = dt - timedelta(hours=MC_TIME_ZONE)
                epoch = int(dt.timestamp())

                key = "email: {0} | epoch {1}".format(email, epoch)

                if key not in existing_event_hash:

                    obj = {
                        "ip": engagement["ip"],
                        "subject": subject,
                        "title":title,
                        "url": archive_url,
                        "action": engagement["action"].title(),
                        "epoch": epoch,
                        "campaign_id": campaign_id
                    }

                    tags = re.findall('\[(.*?)\]', title)

                    if len(tags) > 0:
                        obj["tags"] = tags

                    mp.track_old(email, "MailChimp {0}".format(engagement["action"].title()), epoch, obj)

run_import ("2015-01-01", "2016-08-02")

Sample 'open' showing up in MixPanel. We use [tag] in the title to filter by 'tags' when using MixPanel