Llegó la hora de acabar nuestra aventura a través de las profundidades de Hubspot. Si habéis llegado hasta aquí pero no sabéis muy bien cómo, quizá es que os falte visitar la parte 1 de esta guía. Como dijimos en el final de esa parte, la expedición continuaría con los siguientes pasos:
- Extraer las estadísticas de los emails.
- Extraer los eventos de email
- Merge y creación de fichero csv
- Conclusiones
En esta parte, como ya hemos visto las partes más fundamentales del código y de lo que estamos extrayendo, no me pararé tanto a explicar paso a paso qué es lo que vamos consiguiendo con cada uno de los scripts. Eso sí, si hay algo que todavía no hemos visto o que creo importante visualizar debidamente me extenderé sobre esa parte.
Extraer estadísticas de Emails
¿Os acordáis que habíamos extraído los ids de los emails en el paso anterior verdad? Para el planteamiento que hemos realizado en esta infraestructura, esto era fundamental. A través de esos ids que hemos obtenido vamos a realizar otra llamada a Hubspot y con ella conseguimos los datos «clásicos» de estadísticas sobre emails (aperturas, clicks, etc.). Vamos a ver el código y comentamos la finalidad y las partes más importantes del mismo:
import argparse
import json
import urllib.parse
import os
from typing import Any, Dict, List, Optional
from vantablack import utilities
def load_email_id(file_path: str) -> str:
#load email ids from list
with open(file_path) as read_from_file:
data = json.load(read_from_file)
return data
def marketing_emails_to_directory(directory: str, data: Any, filename: Any) -> None:
#writing email data to files in directory
path = os.path.join(directory, filename)
with open(path, "w") as f:
return json.dump(data, f)
def make_statistics_dict(
#function to create the dictionary with only the information that we need
email_name: str, email_id: str, campaign_name: str, campaign_id: str, primaryEmailCampaignId:str, subject: str, publishDate: Any,
sent: Any, open: Any, delivered: Any, bounce: Any, unsubscribed: Any, click: Any, notsent: Any, hardbounced: Any,softbounced: Any) -> Dict:
return {
'email_name': email_name,
'email_id': email_id,
'campaign_name': campaign_name,
'campaign_id': campaign_id,
'primaryEmailCampaignId': primaryEmailCampaignId,
'subject': subject,
'publishDate': publishDate,
'sent': sent,
'open': open,
'delivered': delivered,
'bounce': bounce,
'unsubscribed': unsubscribed,
'click': click,
'notsent': notsent,
'hardbounced': hardbounced,
'softbounced': softbounced
}
def getting_email_statistics(api_key: str, email_id: str) -> Optional[List]:
#calling hubspot api to get the information that we want
get_marketing_emails_with_statistics = "https://api.hubapi.com/marketing-emails/v1/emails/with-statistics/"
parameter_dict = {'hapikey': api_key}
headers: Dict = {}
parameters = urllib.parse.urlencode(parameter_dict)
get_url = get_marketing_emails_with_statistics + email_id + "?" + parameters
r = utilities.get_with_retries(get_url, headers, 2, 3)
response_dict = r.json()
email_data: List = []
if 'campaignName' in response_dict and 'stats' in response_dict:
if 'open' in response_dict['stats']['counters']:
for campaign in response_dict['allEmailCampaignIds']:
email_statistics = make_statistics_dict(
email_name=response_dict['name'],
email_id=response_dict['id'],
campaign_name=response_dict['campaignName'],
campaign_id=response_dict['campaign'],
subject=response_dict['subject'],
primaryEmailCampaignId=campaign,
publishDate=response_dict['publishDate'],
sent=response_dict['stats']['counters']['sent'],
open=response_dict['stats']['counters']['open'],
delivered= response_dict['stats']['counters']['delivered'],
bounce=response_dict['stats']['counters']['bounce'],
unsubscribed=response_dict['stats']['counters']['unsubscribed'],
click=response_dict['stats']['counters']['click'],
notsent=response_dict['stats']['counters']['notsent'],
hardbounced=response_dict['stats']['counters']['hardbounced'],
softbounced=response_dict['stats']['counters']['softbounced'])
email_data.append(email_statistics)
return email_data
def dump(directory: str, file_ids_path: str, api_key_path: str) -> None:
#dumping all email data indivudally to file
api_key = utilities.load_api_key_from_file(api_key_path)
email_ids = load_email_id(file_ids_path)
for email_id in email_ids:
marketing_emails = getting_email_statistics(api_key, email_id)
if marketing_emails != None:
marketing_emails_to_directory(directory, marketing_emails, "{}.json".format(email_id))
if __name__ == '__main__':
parser = argparse.ArgumentParser(
description="List files with email data in directory and dump them to an output file."
)
parser.add_argument(
"--directory",
required=True,
type=str,
help="path to directory to write the files")
parser.add_argument(
"--file_ids_path",
required=True,
type=str,
help="path to directory where to extract email Ids")
parser.add_argument(
"--api_key_path",
required=True,
type=str,
help="path to credentials")
args = parser.parse_args()
dump(args.directory, args.file_ids_path, args.api_key_path)
Para realizar la llamada de este código, lanzaremos lo siguiente en nuestra terminal (recordad que si no estáis en Windows debéis cambiar el env\Scripts por env/bin y ^ por /):
env\Scripts\python -m vantablack.email_statistics_to_directory^
--directory env\email_statistics^
--file_ids_path env\emails_directory\emails_ids.json^
--api_key_path credentials/hubspot_api_keyComo véis, aquí nos encontramos con un código algo más largo del que habíamos visto en la parte anterior, pero no os asustéis porque esto viene dado principalmente por el hecho de la función que hemos subrayado en el código, en la línea 20. ¿Para qué sirve esta función? Básicamente va a crear un diccionario con únicamente los datos que queremos extraer de la llamada que vamos a realizar más tarde. Y, ¿por qué esto es importante? Esta técnica hace que podamos almacenar los resultados del diccionario en una lista más tarde sin que se sobreescriban los datos (recordad la lógica y estructura de un diccionario en python), que es lo que está ocurriendo de la línea 55 a la 73.
En definitiva, este código lo que está haciendo es recoger los ids que hemos extraído en el paso anterior e iterar sobre ellos para realizar la debida llamada a Hubspot con la que obtener los datos que necesitamos. Lo importante de este código es, principalmente, que además de los datos estadísticos debemos obtener el campo PrimaryEmailCampaignId a través de AllEmailCampaignIds (línea 55), ya que este será con el que podamos llamar a y realizar el merge con los eventos de click que extraigamos a continuación.
Extraer eventos de Email
Creo que se está entendiendo que todo lo que estamos haciendo tiene una implicación en el siguiente paso que damos. Al igual que en Karate Kid, estamos entrenando el músculo para que adquiera los reflejos necesarios y luego sólo tengamos que recordar el mantra «dar cera, pulir cera». Por tanto, y como ya habréis imaginado, para obtener los eventos de click que necesitamos para acabar nuestro análisis, utilizaremos los ids de campaña que hemos extraído en el paso anterior. Vamos al código y tranquilos, que esto ya está casi terminado:
import argparse
import json
import urllib.parse
import os
from typing import Any, Iterable, Dict, List
from vantablack import utilities
def iterate_over_campaign_id(directory_path: str) -> Iterable[Dict]:
#iterate through files in directory to extract primaryEmailCampaignId
directory_files = os.listdir(directory_path)
for file in directory_files:
if file.endswith(".json"):
json_path = os.path.join(directory_path, file)
with open(json_path) as read_from_directory:
for data in json.load(read_from_directory):
yield (data['primaryEmailCampaignId'])
def make_email_events(url: str, date: Any, primaryEmailCampaignId: str, eventId: str, email_user: str) -> Dict:
return {
#dictionary with information needed
'url': url,
'date': date,
'primaryEmailCampaignId': primaryEmailCampaignId,
'eventId': eventId,
'email_user':email_user
}
def email_events_to_directory(directory: str, data: Any, filename: Any) -> None:
#creating files in a directory with the email event data
path = os.path.join(directory, filename)
with open(path, "w") as f:
return json.dump(data, f)
def get_campaign_events(api_key: str, campaign_id: Dict, event_type: str) -> List:
#calling hubspot api to get email events data
get_all_events_url = "https://api.hubapi.com/email/public/v1/events?"
limit = 1000
parameter_dict = {'hapikey': api_key, 'campaignId': campaign_id, 'eventType': event_type, 'limit':limit}
headers: Dict = {}
campaigns_events: List = []
while True:
parameters = urllib.parse.urlencode(parameter_dict)
get_url = get_all_events_url + parameters
r = utilities.get_with_retries(get_url, headers, 2, 3)
response_dict = r.json()
if response_dict is not None:
campaigns_events.extend(response_dict["events"])
parameter_dict['offset'] = response_dict['offset']
if response_dict["hasMore"] == False:
return campaigns_events
def getting_email_events_data(api_key: str, campaign_id: Dict, event_type: str) -> List:
campaign_events = get_campaign_events(api_key, campaign_id, event_type)
events_data = []
for item in campaign_events:
#creating the dictionary
events = make_email_events(
url = item['url'],
date = item['created'],
primaryEmailCampaignId = item['emailCampaignId'],
eventId = item['id'],
email_user=item['recipient'])
events_data.append(events)
print("list from events has {} results".format(len(events_data)))
return events_data
def dump(directory: str, directory_ids_path: str, api_key_path: str, eventType: str) -> None:
#dumping all click events to files in directory
api_key = utilities.load_api_key_from_file(api_key_path)
campaign_ids = iterate_over_campaign_id(directory_ids_path)
for campaign_id in campaign_ids:
if campaign_id != 0:
email_events = getting_email_events_data(api_key, campaign_id, eventType)
if len(email_events) != 0:
email_events_to_directory(directory, email_events, "{}.json".format(campaign_id))
if __name__ == '__main__':
parser = argparse.ArgumentParser(
description="Create a list with all click events by campaign and dump them to an output file in a directory."
)
parser.add_argument(
"--directory",
required=True,
type=str,
help="path to directory to write the files")
parser.add_argument(
"--directory_ids_path",
required=True,
type=str,
help="path to directory where to extract primaryEmailCampaignId")
parser.add_argument(
"--api_key_path",
required=True,
type=str,
help="path to credentials")
parser.add_argument(
"--eventType",
required=True,
type=str,
help="type of event that we want to extract")
args = parser.parse_args()
dump(args.directory, args.directory_ids_path, args.api_key_path, args.eventType)
Para realizar la llamada de este código, lanzaremos lo siguiente en nuestra terminal (recordad que si no estáis en Windows debéis cambiar el env\Scripts por env/bin y ^ por /):
env\Scripts\python -m vantablack.email_click_events_to_directory^
--directory env\email_click_events^
--directory_ids_path env\email_statistics^
--api_key_path credentials/hubspot_api_key^
--eventType CLICKMe apuesto lo que queráis a que a estas alturas la estructura de este script ya os es familiar, ¿verdad? Cómo seguramente habréis adivinado como seres excepcionales que sois, este código se encarga de atrapar esos ids de campaign email de los que hemos hablado antes, iterar sobre ellos y realizar la llamada para que nos devuelva la información que necesitamos sobre los eventos de click de los usuarios de Vantablack. No olvidemos lo que queríamos conseguir en última instancia: cómo interactúan los usuarios con los emails de nuestro cliente. Eso lo conseguimos gracias a este paso. Ahora sólo nos queda darle algo de forma.
Merge y creación de fichero csv
Damas y caballeros, llegamos al final verdadero de nuestro problema (aunque podríamos seguir un poquito más, pero lo dejaremos para otra entrada) ya que este es el paso en el que obligamos a las dos partes que hemos extraído de datos a llevarse extremadamente bien. Debemos conseguir un fichero csv que siga más o menos esta estructura, donde url será el click con el que han interactuado los usuarios en el email:
Si todo ha salido como hemos ido sugiriendo en esta guía, las tres carpetas que hemos creado al principio dentro de env se habrán estado llenando de una cantidad considerable de ficheros. Esto nos es útil porque evitamos los errores de duplicados o falta de datos de manera exponencial. Ahora lo que debemos hacer es recoger todos esos ficheros y crear un código que sea capaz de unirlos a partir de un campo común que los identifique (lo habéis adivinado, PrimaryEmailCampaignId). Y este código reza así:
import json
import argparse
import os
import pandas as pd
from typing import Any
def getting_data_to_join(directory_path: str) -> Any:
#loading files to use in our function
directory_files = os.listdir(directory_path)
for file in directory_files:
if file.endswith(".json"):
json_path = os.path.join(directory_path, file)
with open(json_path) as read_from_directory:
data = json.load(read_from_directory)
yield (data)
def data_to_csv(directory_emails: str, directory_events: str, filename:str):
emails = []
events = []
for email_data in getting_data_to_join(directory_emails):
for email in email_data:
emails.append(email)
for events_data in getting_data_to_join(directory_events):
for event in events_data:
events.append(event)
#creating the datasets to join
emails_data = pd.DataFrame(emails)
events_data = pd.DataFrame(events)
#merging on primaryEmailCampaignId
joining_data = emails_data.merge(events_data, on='primaryEmailCampaignId', how='inner')
#csv to directory
joining_data.to_csv(filename, sep=';', index=False)
if __name__ == '__main__':
parser = argparse.ArgumentParser(
description="Load event and email data and create an output join file."
)
parser.add_argument(
"--directory_emails",
required=True,
type=str,
help="path to directory to load email data")
parser.add_argument(
"--directory_events",
required=True,
type=str,
help="path to directory to load events data")
parser.add_argument(
"--filename",
required=True,
type=str,
help="name for csv file")
args = parser.parse_args()
data_to_csv(args.directory_emails, args.directory_events, args.filename)
Para realizar la llamada de este código, lanzaremos lo siguiente en nuestra terminal (recordad que si no estáis en Windows debéis cambiar el env\Scripts por env/bin y ^ por /):
env\Scripts\python -m vantablack.merging_events_with_emails^
--directory_emails env\email_statistics^
--directory_events env\email_click_events^
--filename joined_data.csvCreo que no hay nada en este código que os vaya a sorprender en este momento de la película, salvo la incorporación de una nueva librería y un pequeño código que la recoge. Estoy hablando, como no, de pandas. Pandas es una librería que incorpora una cantidad increíble de materiales y herramientas para trabajar con datos de forma sencilla en python y que se ha ido haciendo progresivamente más popular. Gracias a ella puedes manipular datos en cuestión de segundos sin tener que pasar por procesos que en otro momento hubieran costado mucho más. En este caso, la vamos a utilizar para tres cosas:
– Para crear dos DataFrames (un conjunto de datos ordenados en formato tabla, que es el «estándar» de Pandas) con las listas que hemos generado a partir de eventos y emails.
– Realizar la unión de esos dos DataFrames a través del campo único del que ya hemos hablado anteriormente.
– Generar el fichero csv con la unión ya realizada, con la estructura que hemos visto al principio de esta sección.
Todos estos pasos los tenéis highlighteados en el código y, cómo podéis observar, son extremadamente sencillos de interpretar. En las líneas 27 y 28 simplemente son la creación de los dataframes. En la línea 30 realizamos el merge con el módulo correspondiente de pandas y lo almacenamos en una nueva variable. De esta sintaxis es importante señalar que el campo único sobre el que debe realizar el merge se indica a través del parámetro «on=» y el formato de unión a través del parámetro «how=». Este tipo de uniones tiene su origen en los joins de SQL, que básicamente siguen esta estructura:
Y lo más alucinante de Pandas creo que viene aquí, cuando con sólo una línea de código le podemos decir que genere un csv con la nueva variable que habíamos creado anteriormente. Esto lo realizamos a través del módulo «to_csv» (tampoco hacía falta ser originales en el naming) que vemos en la línea 32. That’s it. Mira tu carpeta. Efectivamente: ahí está tu csv.
Conclusiones
Por fin puedes decir que eres un completo explorador de Hubspot. En este viaje has aprendido mucho, pero yo diría que lo más importante es lo siguiente:
- Acotar un problema determinado y ofrecer una solución
- Generar un código capaz de extraer datos de una API Rest
- Trabajar con archivos json
- Trabajar con ficheros y directorios
- Trabajar con módulos de librerias externas
- Generar un paquete y crear tus propios módulos
- Realizar un merge entre dos tipos diferentes de datos
- Extraer un archivo csv con el que trabajar el análisis
No está de más comentar que esta aproximación es una de las muchas que se pueden realizar (algunas infinitamente más rápidas y eficientes). He utilizado este enfoque principalmente por motivos didácticos, ya que creo que muchos de los pasos realizados aquí son esenciales para entender bien el proceso que se está llevando a cabo y ayudan a solventar cómo lidiar con determinadas situaciones. En cualquier caso, creo que no está nada mal para empezar con la sección, pero sin duda queda mucho más por recorrer. De momento, debo darte la enhorabuena por llegar hasta aquí, eres un auténtico #apicaller. Te espero en los comentarios.
Visión de Funnel 