Mauticのコンタクト ページヒットイベントをBigQueryへ読み込む

$ curl -G -s -X GET https://<Mautic サイトURL>/api/contacts/activity \
-d 'filters[includeEvents][]=page.hit' \
-d 'filters[dateFrom]=2020-10-01' \
-d 'filters[dateTo]=2020-10-25' \
-u "<ユーザー名>:<パスワード>"

$ python -m venv venv
$ source venv/bin/activate

$ pip install google-cloud-storage

import urllib.request
import json
import base64
import os
from google.cloud import storage


def upload_blob(bucket_name, source_file_name, destination_blob_name):
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(source_file_name)


def main():
    user = os.environ['USER']
    password = os.environ['PASSWORD']
    url = os.environ['URL']
    date_from = os.environ['DATE_FROM']
    date_to = os.environ['DATE_TO']
    bucket = os.environ['BUCKET']
    dest_folder = os.environ['DEST_FOLDER']
    api_endpoint = '%s/api/contacts/activity' % (url)
    params = {
        'filters[includeEvents][]': 'page.hit',
        'filters[dateFrom]': date_from,
        'filters[dateTo]': date_to
    }
    basic_user_and_pasword = base64.b64encode(
        '{}:{}'.format(user, password).encode('utf-8'))
    request_url = '%s?%s' % (
        api_endpoint, urllib.parse.urlencode(params))
    req = urllib.request.Request(request_url,
                                 headers={"Authorization": "Basic " +
                                          basic_user_and_pasword.decode(
                                              'utf-8')
                                          })
    res = urllib.request.urlopen(req)
    body = res.read()
    data = json.loads(body)
    max_page = data['maxPages']

    for page in range(1, int(max_page)+1):
        params.update(page=page)
        request_url = '%s?%s' % (
            api_endpoint, urllib.parse.urlencode(params))
        req = urllib.request.Request(request_url,
                                     headers={"Authorization": "Basic " +
                                              basic_user_and_pasword.decode(
                                                  'utf-8')
                                              })
        try:
            res = urllib.request.urlopen(req)
            body = res.read()
            data = json.loads(body)
            file_name = '%s_%s_events%d.json' % (date_from, date_to, page)
            for e in data['events']:
                with open(file_name, 'a') as file:
                    message = '%s\n' % (json.dumps(e, ensure_ascii=False))
                    file.write(message)
            dest_file = '%s/%s' % (dest_folder, file_name)
            upload_blob(bucket, file_name, dest_file)
            os.remove(file_name)

        except urllib.error.URLError as e:
            print(e.reason)


if __name__ == "__main__":
    main()

export USER='test@example.com'
export PASSWORD='test'
export URL='https://example.com'
export DATE_FROM='2020-10-01'
export DATE_TO='2020-10-25'
export BUCKET='example-bucket'
export DEST_FOLDER='mautic/contacts_activity'

$ python main.py

$ pip install bigquery_schema_generator
$ source venv/bin/activate

$ gsutil cat gs://<GCS バケット名>/mautic/contacts_activity/2020-10-01_2020-10-05_events1.json | generate-schema > schema.json

$ bq --location=asia-northeast1 mk --dataset mautic

$ bq load \
--source_format=NEWLINE_DELIMITED_JSON \
--ignore_unknown_values \
mautic.contacts_activity \
gs://<GCS バケット名>/mautic/contacts_activity/* \
schema.json

SELECT
  timestamp,
  details.hit.device,
  contactId,
  details.hit.query.page_url,
FROM
  `<project_id>.mautic.contacts_activity`
WHERE
  details.hit.query.page_url LIKE '%gclid%'
ORDER BY
  timestamp DESC;