mixpanel_export.py

#! /usr/bin/env python
#
# Mixpanel, Inc. -- http://mixpanel.com/
#
# Python API client library to consume mixpanel.com analytics data.
#
# Copyright 2010-2013 Mixpanel, Inc
# Copyright 2015, Andrew Grigorev <[email protected]>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import hashlib
import urllib
import urllib2
import time
import json


class MixpanelExport(object):

    def __init__(self, api_key, api_secret,
                 endpoint='https://data.mixpanel.com/api',
                 version='2.0'):
        self.api_key = api_key
        self.api_secret = api_secret
        self.endpoint = endpoint
        self.version = version

    def _request(self, from_date, to_date, event=None, where=None, bucket=None):
        params = {
            'from_date': from_date,
            'to_date': to_date,
        }
        if event is not None:
            params['event'] = event
        if where is not None:
            params['where'] = where
        if bucket is not None:
            params['bucket'] = bucket

        params['api_key'] = self.api_key
        params['expire'] = int(time.time()) + 600   # Grant this request 10 minutes.
        params['sig'] = self.hash_args(params)

        request_url = '%s/%s/export/?%s' % (
            self.endpoint, self.version,
            self.unicode_urlencode(params)
        )

        return urllib2.urlopen(request_url, timeout=120)

    def export_to_file(self, fp, from_date, to_date, event=None, where=None, bucket=None):
        request = self._request(from_date, to_date, event, where, bucket)
        while True:
            chunk = request.read(4096)
            if not chunk:
                break
            fp.write(chunk)

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

        return urllib.urlencode(
            [(k, isinstance(v, unicode) and v.encode('utf-8') or v) for k, v in params]
        )

    def hash_args(self, args, secret=None):
        """
            Hashes arguments by joining key=value pairs, appending a secret, and
            then taking the MD5 hex digest.
        """
        for a in args:
            if isinstance(args[a], list): args[a] = json.dumps(args[a])

        args_joined = ''
        for a in sorted(args.keys()):
            if isinstance(a, unicode):
                args_joined += a.encode('utf-8')
            else:
                args_joined += str(a)

            args_joined += '='

            if isinstance(args[a], unicode):
                args_joined += args[a].encode('utf-8')
            else:
                args_joined += str(args[a])

        hash = hashlib.md5(args_joined)

        if secret:
            hash.update(secret)
        elif self.api_secret:
            hash.update(self.api_secret)
        return hash.hexdigest()


if __name__ == '__main__':

    import os
    import sys

    if len(sys.argv) < 2:
        print("""
    Usage: %s FROM_DATE [TO_DATE]

    Use MIXPANEL_API_KEY and MIXPANEL_API_SECRET environment variables to pass
    Mixpanel credentials.
    """ % sys.argv[0])
        sys.exit(1)

    mixpanel = MixpanelExport(
        os.environ['MIXPANEL_API_KEY'],
        os.environ['MIXPANEL_API_SECRET'],
    )
    from_date = sys.argv[1]
    if len(sys.argv) == 3:
        to_date = sys.argv[2]
    else:
        to_date = from_date

    mixpanel.export_to_file(sys.stdout, from_date, to_date)