Importe informes de llamadas programadas de CoMagic a BigQuery en un horario usando Google Cloud Functions

Para que

Con la compleja estructura de las campañas publicitarias y una gran cantidad de llamadas, se hacen necesarias herramientas adicionales para almacenar, procesar y analizar información sobre las llamadas entrantes. A menudo necesita un acceso rápido a los datos durante un largo período de tiempo. A veces necesita un procesamiento de datos complejo, correlacionando llamadas a un canal o campaña específicos.

Una de las opciones para acelerar el trabajo, que también proporciona beneficios adicionales, es importar llamadas de CoMagic a Google BigQuery. Mucho se ha escrito sobre los beneficios de BigQuery, así que pasemos a la creación.

Para crear una importación automática, necesitará:

1. Cuenta de Google (si aún no lo está) con el proyecto creado
2. Conocimiento de Python
3. Presentación de la documentación de Google Cloud

Aquí se describe cómo crear un proyecto. Después de crear el proyecto, debe crear un conjunto de datos en BigQuery. Documentación de BQ e instrucciones para crear un conjunto de datos .

Recuperando datos de CoMagic

Volviendo a la documentación de CoMagic. Para obtener una lista de llamadas o llamadas, necesitamos la sección de informes.

Creamos una clase simple para trabajar con la API de CoMagic. Todos los requisitos necesarios se indicarán al final en el enlace a GitHub.

import json import requests import random import pandas as pd class ComagicClient: """     CoMagic""" def __init__(self, login, password): """     CoMagic""" self.login = login self.password = password self.base_url = 'https://dataapi.comagic.ru/v2.0' self.payload_ = {"jsonrpc":"2.0", "id":1, "method":None, "params": None} self.token = self.get_token(self.login, self.password) def base_request(self, method, params): """     CoMagic.      API   .    JSON-like . : https://www.comagic.ru/support/api/data-api/""" id_ = random.randrange(10**7) #  payload = self.payload_.copy() payload["method"] = method payload["params"] = params payload["id"] = id_ self.r = requests.post(self.base_url, data=json.dumps(payload)) self.last_response = json.loads(self.r.text) return self.last_response def get_token(self, login, password): """   .       .  .""" method = "login.user" params = {"login":self.login, "password":self.password} response = self.base_request(method, params) token = response['result']['data']['access_token'] return token def get_report_per_page(self, method, params): """  .      10000 .    .     110000 .     JSON-like .""" response = self.base_request(method, params) print(f"""  c {params["date_from"]}  {params["date_till"]}.  = {params["offset"]}""") result = response['result']['data'] if len(result) < 10000: return result else: params['offset'] += 10000 add_result = self.get_report_per_page(method, params) return result + add_result def get_basic_report(self, method, fields, date_from, date_till, filter=None, offset=0): """   .       method  fields.       .       ,   ,       . method -- <string>   date_from -- <string>  .  "YYYY-MM-DD hh:mm:ss" date_till -- <string>  .  "YYYY-MM-DD hh:mm:ss" fields -- <list>,    filter [] - <dict>  offset [] -- <int>  return -- <list>  """ params = {"access_token":self.token, "limit":10000, "date_from":date_from, "date_till":date_till, "fields": fields, "offset": offset} if filter: params['filter'] = filter report = self.get_report_per_page(method, params) return report 

Ahora debe determinar qué tipo de datos se necesitan. Los datos deben procesarse y hacerse visibles para poder cargarlos en BigQuery.

Cree una clase auxiliar y defina los datos recibidos de CoMagic.

 class ComagicHandler(ComagicClient): """    ,   CoMagic""" time_partition_field = 'PARTITION_DATE' def __init__(self, login, password, first_call_date): self.day_before_first_call = pd.to_datetime(first_call_date) - pd.Timedelta(days=1) super().__init__(login, password) def get_calls_report(self, date_from, date_till): """        .           .    Pandas DataFrame.      .      Connector    .    .    .  Pnadas.DataFrame""" method = "get.calls_report" fields = ['id', 'visitor_id', 'person_id', 'start_time', 'finish_reason', 'is_lost', 'tags', 'campaign_name','communication_number', 'contact_phone_number', 'talk_duration', 'clean_talk_duration', 'virtual_phone_number', 'ua_client_id', 'ym_client_id', 'entrance_page', 'gclid', 'yclid', 'visitor_type', 'visits_count', 'visitor_first_campaign_name', 'visitor_device', 'site_domain_name','utm_source', 'utm_medium', 'utm_campaign', 'utm_content', 'eq_utm_source', 'eq_utm_medium', 'eq_utm_campaign', 'attributes'] #   CoMagic calls_data = self.get_basic_report(method, fields, date_from, date_till) # DataFrame df = pd.DataFrame(calls_data) #    .    . df[self.time_partition_field] = pd.to_datetime(df.start_time).apply(lambda x: x.date()) #  tags,   BigQuery       ,  # CoMagic.    . df['tags'] = df.tags.apply(lambda x: x if x == None else [i['tag_name'] for i in x]) return df 

Envío de datos a BigQuery

Después de recibir y convertir los datos de CoMagic, debe enviarlos a BigQuery.

 from google.cloud import bigquery from google.cloud.exceptions import NotFound import pandas as pd class BQTableHanler: """     BigQuery""" time_partition_field = 'PARTITION_DATE' def __init__(self, full_table_id, service_account_file_key_path = None): """       `myproject.mydataset.mytable`.  ,   Application Default Credentials,           .""" self.full_table_id = full_table_id project_id, dataset_id, table_id = full_table_id.split(".") self.project_id = project_id self.dataset_id = dataset_id self.table_id = table_id if service_account_file_key_path: #      from google.oauth2 import service_account self.credentials = service_account.Credentials.from_service_account_file( service_account_file_key_path, scopes=["https://www.googleapis.com/auth/cloud-platform"],) self.bq_client = bigquery.Client(credentials = self.credentials, project = self.project_id) else: self.bq_client = bigquery.Client() self.dataset = self.bq_client.get_dataset(self.dataset_id) self.location = self.dataset.location self.table_ref = self.dataset.table(self.table_id) def get_last_update(self): """        Pandas datetime.      False.""" try: self.bq_client.get_table(self.full_table_id) except NotFound as error: return False query = f"""SELECT MAX({self.time_partition_field}) as last_call FROM `{self.full_table_id}`""" result = self.bq_client.query(query,location=self.location).to_dataframe() date = pd.to_datetime(result.iloc[0,0]).date() return date def insert_dataframe(self, dataframe): """      BigQuery.     Pandas DataFrame.    ,       .""" job_config = bigquery.LoadJobConfig() #     job_config._properties['load']['timePartitioning'] = {'type': 'DAY', 'field': self.time_partition_field} result = self.bq_client.load_table_from_dataframe(dataframe, self.table_ref, job_config=job_config).result() return result 

Determinar la lógica para actualizar los datos.

Como existe un límite en el número de filas de datos recibidas de CoMagic, es necesario limitar el número de datos solicitados. Limitaremos el período de solicitud. Para hacer esto, necesita una función auxiliar que dividirá un gran período de tiempo en segmentos de una longitud específica.

 def interval_split(array, interval): """      .   ,      2,    -   ,     -    . : get_intervals([1,2,3,4,5,6,7], 3) => [[1,3], [4,6], [7]] get_intervals([1,2,3], 4) => [[1,3]]""" intervals = [] iw, i = 0, 0 l = len(array) for v in array: if i==0 or (i)%interval==0: intervals.append([v]) if (i+1)%interval == 0 or (i+1) == l: intervals[iw].append(v) iw+=1 i+=1 return intervals 

Esto es necesario cuando se cargan datos por primera vez, cuando necesita descargar datos durante un largo período de tiempo. El período se divide en varios períodos pequeños. Por cierto, es mejor hacerlo sin usar la función de nube, ya que tienen un límite de tiempo. Bueno, o, como opción, puede ejecutar la función muchas, muchas veces.

Creamos una clase de conector para vincular la tabla BigQuery donde queremos almacenar los datos y los datos de CoMagic.

 from helpfunctions import interval_split import pandas as pd class Connector: """      """ time_partition_field = 'PARTITION_DATE' #  -.       def __init__ (self, source, dest): """          """ self.source = source self.dest = dest self.source.time_partition_field = self.time_partition_field self.dest.time_partition_field = self.time_partition_field def insert_data_in_dest(self, start_date, end_date): """      .          ,     .""" dates = pd.date_range(start_date, end_date) week_intervals = interval_split(dates, 7) #     7  for week_interval in week_intervals: date_from = week_interval[0].strftime("%Y-%m-%d") + " 00:00:00" date_till = week_interval[1].strftime("%Y-%m-%d") + " 23:59:59" calls_df = self.source.get_calls_report(date_from, date_till) self.dest.insert_dataframe(calls_df) print (f"  {date_from}  {date_till}   ") return True def update_dest_data(self): #     BigQuery last_date = self.dest.get_last_update() if not last_date: #    last_date = self.source.day_before_first_call yesterday = pd.Timestamp.today(tz='Europe/Moscow').date() - pd.Timedelta(days=1) if last_date == yesterday: print("  ") else: last_date = last_date + pd.Timedelta(days=1) self.insert_data_in_dest(last_date, yesterday) return True 

A continuación, prescribimos la función principal para actualizar los datos, que se lanzarán en un horario.

 from connector import Connector from bqhandler import BQTableHanler from comagichandler import ComagicHandler from credfile import * def main(event, context): """    event, context  : https://cloud.google.com/functions/docs/writing/background#functions-writing-background-hello-pubsub-python""" #import base64 #pubsub_message = base64.b64decode(event['data']).decode('utf-8') # c      comagic_handler = ComagicHandler(COMAGIC_LOGIN, COMAGIC_PASSWORD, FIRST_CALL_DATE) bq_handelr = BQTableHanler(full_table_id, google_credintials_key_path) #  connector = Connector(comagic_handler, bq_handelr) #     connector.update_dest_data() 

Configurar Google Cloud Platform

Recopilamos todos los archivos en un archivo ZIP. En el archivo credfile.py, ingresamos el nombre de usuario y la contraseña de CoMagic para recibir el token, así como el nombre completo de la tabla en BigQuery y la ruta al archivo de la cuenta de servicio si el script se inicia desde la máquina local.

Crear una función en la nube

• Ir a la consola
• Si aún no se ha creado ninguna función, haga clic en "Crear función"
• En el campo de activación, seleccione PUB / SUB
• Crea un nuevo tema para PUB / SUB. Por ejemplo, 'update_calls'
• Fuente: carga ZIP (archivo ZIP local)
• Entorno: Python 3.7
• Descargar el archivo zip
• Elegir un segmento temporal de almacenamiento en la nube
• En el campo `llamado función` escribimos 'main'
• Memoria asignada: opcional

Configuración del planificador y PUB / SUB

En el último paso, creamos el activador `update_calls`. Este tema automático ha aparecido en la lista de temas .

Ahora, con Cloud Scheduler necesita configurar el disparador. cuándo se disparará y se iniciará el GCF.

• Ir a la consola
• En el campo de frecuencia en el formato CRON, configure cuándo debe dispararse el disparador y se inicia la función.
• Destino: Pub / Sub
• Asunto: registre el tema que se especificó al crear la función: "update_calls"
• Carga útil * (Carga útil): esta es la información que se transferirá a Pub / Sub y a la función principal

Ahora el script se lanzará diariamente a las 01:00 y los datos de la llamada se actualizarán al final del día anterior.

Enlace a GitHub para ejecutar desde la computadora local
Enlace de GitHub al archivo ZIP