Short_io_Obter_Cliques.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "BPyoatiUF9dZ"
      },
      "outputs": [],
      "source": [
        "# @title Importações e Parâmetros Globais { display-mode: \"form\" }\n",
        "# importando bibliotecas\n",
        "\n",
        "\"\"\"\n",
        "  @description 'Python que obtem detalhes dos cliques no Short.io e insere no BigQuery'\n",
        "  @author 'Everton Moraes'\n",
        "  @version '1.0'\n",
        "  @since '08/10/2024'\n",
        "\n",
        "\"\"\"\n",
        "import requests\n",
        "import datetime\n",
        "import pandas as pd\n",
        "import pandas_gbq as pd_gbq\n",
        "from pytz import timezone\n",
        "from google.auth import default\n",
        "from google.cloud import bigquery\n",
        "from google.colab import auth\n",
        "\n",
        "\n",
        "# Parâmetros Globais\n",
        "fuso_horario = timezone('America/Sao_Paulo')\n",
        "\n",
        "# Ambiente BigQuery\n",
        "var_project = <projeto_bigquery>\n",
        "\n",
        "# Autenticando usuario\n",
        "auth.authenticate_user()\n",
        "\n",
        "# Credenciais Pandas BQ\n",
        "credentials, _ = default()\n",
        "pd_gbq.context.credentials = credentials\n",
        "pd_gbq.context.project = var_project\n",
        "\n",
        "api_key= <chave api short.io>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "adxzyjy0HaEU"
      },
      "outputs": [],
      "source": [
        "# @title Funções { display-mode: \"form\" }\n",
        "\n",
        "# Variaveis Globais para as Funções\n",
        "\n",
        "ls_log = []\n",
        "\n",
        "# log\n",
        "def log (mensagem):\n",
        "  \"\"\"\n",
        "  Função para gera o log/acompanhamento das execuções\n",
        "\n",
        "  Args:\n",
        "    mensagem (str): O caminho da pasta atual.\n",
        "  \"\"\"\n",
        "  agora = datetime.datetime.now().astimezone(fuso_horario).strftime(\"%d/%m/%Y %H:%M:%S\")\n",
        "  print(f'{agora} - {mensagem}')\n",
        "  ls_log.append(f'{agora} - {mensagem}')\n",
        "\n",
        "def obter_clicks_data_dominio(id_dominio,data,limite_lote):\n",
        "  \"\"\"\n",
        "  Função obter os clicks por data do Short.io\n",
        "  Documentação do método utilizado: https://developers.short.io/reference/postdomaindomainidlast_clicks\n",
        "\n",
        "  Args:\n",
        "    id_dominio(str): Id do dominio a ser obtido os clicks\n",
        "    data(strg): Data a obter os clicks, formato YYYY-MM-DD\n",
        "\n",
        "  Returns:\n",
        "    dict: Retorno da API\n",
        "  \"\"\"\n",
        "\n",
        "  data_inicio = datetime.datetime.strptime(data, \"%Y-%m-%d\")\n",
        "  data_fim = data_inicio + datetime.timedelta(days=1)\n",
        "\n",
        "  # Header\n",
        "  headers = {\n",
        "      \"accept\": \"*/*\",\n",
        "      \"content-type\": \"application/json\",\n",
        "      \"Authorization\": api_key\n",
        "  }\n",
        "\n",
        "  # Configura parametros\n",
        "  payload = {\n",
        "      \"limit\": limite_lote,\n",
        "      \"period\": \"custom\",\n",
        "      \"tz\": \"America/Sao_Paulo\",\n",
        "      \"startDate\": data_inicio.strftime(\"%Y-%m-%d\"),\n",
        "      \"endDate\": data_fim.strftime(\"%Y-%m-%d\"),\n",
        "  }\n",
        "\n",
        "  url = f\"https://statistics.short.io/statistics/domain/{id_dominio}/last_clicks\"\n",
        "\n",
        "  log(f\"Obtendo os cliques do dia {data_inicio.strftime('%Y-%m-%d')}, do id_dominio: {id_dominio}\")\n",
        "\n",
        "  # Realiza requisicao\n",
        "  request = requests.post(url, json=payload, headers=headers)\n",
        "\n",
        "  request.raise_for_status()\n",
        "  js_retorno = request.json()\n",
        "  df_retorno = pd.DataFrame(js_retorno)\n",
        "\n",
        "  return js_retorno,df_retorno\n",
        "\n",
        "\n",
        "def obter_clicks_data_dominio_browser(id_dominio,browser,data,limite_lote):\n",
        "  \"\"\"\n",
        "  Função obter os clicks por data do Short.io\n",
        "  Documentação do método utilizado: https://developers.short.io/reference/postdomaindomainidlast_clicks\n",
        "\n",
        "  Args:\n",
        "    id_dominio(str): Id do dominio a ser obtido os clicks\n",
        "    browser(str): Browser a ser obtido os clicks\n",
        "    data(strg): Data a obter os clicks, formato YYYY-MM-DD\n",
        "\n",
        "  Returns:\n",
        "    dict: Retorno da API\n",
        "  \"\"\"\n",
        "\n",
        "  data_inicio = datetime.datetime.strptime(data, \"%Y-%m-%d\")\n",
        "  data_fim = data_inicio + datetime.timedelta(days=1)\n",
        "\n",
        "  # Header\n",
        "  headers = {\n",
        "      \"accept\": \"*/*\",\n",
        "      \"content-type\": \"application/json\",\n",
        "      \"Authorization\": api_key\n",
        "  }\n",
        "\n",
        "  # Configura parametros\n",
        "  payload = {\n",
        "      \"limit\": limite_lote,\n",
        "      \"period\": \"custom\",\n",
        "      \"tz\": \"America/Sao_Paulo\",\n",
        "      \"startDate\": data_inicio.strftime(\"%Y-%m-%d\"),\n",
        "      \"endDate\": data_fim.strftime(\"%Y-%m-%d\"),\n",
        "      \"include\": {\n",
        "        \"browsers\": [browser]\n",
        "                },\n",
        "  }\n",
        "\n",
        "  url = f\"https://statistics.short.io/statistics/domain/{id_dominio}/last_clicks\"\n",
        "\n",
        "  log(f\"Obtendo os cliques do dia {data_inicio.strftime('%Y-%m-%d')}, do id_dominio: {id_dominio}, do browser: {browser}\")\n",
        "\n",
        "  # Realiza requisicao\n",
        "  request = requests.post(url, json=payload, headers=headers)\n",
        "\n",
        "  request.raise_for_status()\n",
        "  js_retorno = request.json()\n",
        "  df_retorno = pd.DataFrame(js_retorno)\n",
        "\n",
        "\n",
        "  return js_retorno,df_retorno\n",
        "\n",
        "def obter_ultima_data_bigquery():\n",
        "  \"\"\"\n",
        "  Função para obter a ultima data dos cliques no BigQuery\n",
        "\n",
        "  Returns:\n",
        "    str: Retorno da data\n",
        "  \"\"\"\n",
        "\n",
        "  # Criando um Cliente BigQuery\n",
        "  bq_client = bigquery.Client(project=var_project)\n",
        "\n",
        "  # Query verificar se existe o cod_segmetacao\n",
        "  consulta = bq_client.query(\"SELECT MAX(CAST(A.DATAHORA_ACESSO AS DATE)) FROM `<projeto_bigquery>.<dataset_bigquery>.TB_SHORTIO_CLIQUES`  AS A\")\n",
        "\n",
        "  for row in consulta.result():\n",
        "    ult_data = row[0]\n",
        "\n",
        "  return ult_data\n",
        "\n",
        "def obter_browsers_bigquery():\n",
        "  \"\"\"\n",
        "  Função para obter browsers no BigQuery\n",
        "\n",
        "  Returns:\n",
        "    list: Lista de browsers\n",
        "  \"\"\"\n",
        "  # Criando um Cliente BigQuery\n",
        "  bq_client = bigquery.Client(project=var_project)\n",
        "\n",
        "  # Query verificar se existe o browser\n",
        "  consulta = bq_client.query(\"SELECT DISTINCT BROWSER FROM `<projeto_bigquery>.<dataset_bigquery>.TB_SHORTIO_CLIQUES` ORDER BY 1 ASC\")\n",
        "\n",
        "  ls_final = []\n",
        "\n",
        "  for linha in consulta.result():\n",
        "      ls_final.append(linha[0])\n",
        "\n",
        "  return ls_final\n",
        "\n",
        "\n",
        "def inserir_dados_bigquery(df):\n",
        "  \"\"\"\n",
        "  Função para inserir os dados no BigQuery\n",
        "\n",
        "  \"\"\"\n",
        "  # Copia Dataframe\n",
        "  df_importar = df.copy()\n",
        "\n",
        "  # df_importar a coluna com a data tratada\n",
        "  df_importar['DATAHORA_ACESSO'] = pd.to_datetime(df_importar['dt'], format='%Y-%m-%dT%H:%M:%S.%f-03:00')\n",
        "\n",
        "  # Renome a coluna proto\n",
        "  df_importar = df_importar.rename(columns={'proto': 'protocolo'})\n",
        "  # Reorganiza as colunas\n",
        "  df_importar2 = df_importar[['DATAHORA_ACESSO'] + [col for col in df_importar.columns if col != 'DATAHORA_ACESSO']]\n",
        "  df_importar = df_importar2\n",
        "\n",
        "  log(f\"Aplicado as tratativas nos dados\")\n",
        "\n",
        "  # Insere os dados no BigQuery\n",
        "  pd_gbq.to_gbq(df_importar2, '<dataset_bigquery>.TB_SHORTIO_CLIQUES', project_id='<projeto_bigquery>', if_exists='append')\n",
        "\n",
        "  log(f\"Inseridos {len(df_importar):,} registros no BigQuery\")"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "hJtVmZr9MXAz"
      },
      "outputs": [],
      "source": [
        "# @title Execução{ form-width: \"10px\", display-mode: \"form\" }\n",
        "\n",
        "\n",
        "# Configura intervalos\n",
        "#inicio_intervalo  = datetime.datetime.strptime('2023-10-01', \"%Y-%m-%d\")\n",
        "#inicio_intervalo = datetime.datetime.now(fuso_horario)- datetime.timedelta(days=1)\n",
        "\n",
        "ult_data_bigquery = obter_ultima_data_bigquery()\n",
        "\n",
        "# se o objeto esta vazio seta a data de um ano atrás (tempo limite de dados guardados pela short.io)\n",
        "if ult_data_bigquery == None:\n",
        "  data_inicio = datetime.datetime.now(fuso_horario) - datetime.timedelta(days=370)\n",
        "else:\n",
        "  data_inicio =  datetime.datetime.strptime(str(ult_data_bigquery), \"%Y-%m-%d\")   + datetime.timedelta(days=1)\n",
        "\n",
        "inicio_intervalo  = data_inicio\n",
        "fim_intervalo = datetime.datetime.now(fuso_horario) - datetime.timedelta(days=1)\n",
        "\n",
        "\"\"\"\n",
        "inicio_intervalo  = datetime.datetime.strptime('2024-05-17', \"%Y-%m-%d\")\n",
        "fim_intervalo  = datetime.datetime.strptime('2024-05-17', \"%Y-%m-%d\")\n",
        "\"\"\"\n",
        "\n",
        "# Configura dominios\n",
        "id_dominio = [\n",
        "               ['<id_dominio_1>','<nome_dominio_1>']\n",
        "              ,['<id_dominio_2>','<nome_dominio_2>']\n",
        "              ,['<id_dominio_3>','<nome_dominio_3>']\n",
        "              ]\n",
        "\n",
        "# Data atual\n",
        "data_atual = inicio_intervalo\n",
        "\n",
        "# Loop datas\n",
        "while data_atual.date() <= fim_intervalo.date():\n",
        "\n",
        "  # Loop dominios\n",
        "  for id in id_dominio:\n",
        "\n",
        "    id_atual = id[0]\n",
        "    dominio_atual = id[1]\n",
        "\n",
        "    log(f\"Obtendo os cliques do dominio '{dominio_atual}'\")\n",
        "\n",
        "    # Limite de cliques por lote\n",
        "    limite_lote = 5000\n",
        "\n",
        "    # Obtendo os clicks da data/dominio\n",
        "    js_clicks,df_click = obter_clicks_data_dominio(id_atual,data_atual.strftime(\"%Y-%m-%d\"),limite_lote)\n",
        "\n",
        "    # Verifica se tiveram cliques\n",
        "    if len(df_click) == 0:\n",
        "      log(f\"Não foi localizados cliques no dia {data_atual.strftime('%Y-%m-%d')}\")\n",
        "\n",
        "    # Localizado cliques\n",
        "    if len(df_click) > 0:\n",
        "\n",
        "      log(f\"Obtido {len(df_click):,} cliques do dia {data_atual.strftime('%Y-%m-%d')}\")\n",
        "\n",
        "      # Verifica se atingiu o total do limite do lote\n",
        "      if len(df_click) == limite_lote:\n",
        "\n",
        "          # Otem os browsers exitestes no BigQuery\n",
        "          ls_browsers = obter_browsers_bigquery()\n",
        "\n",
        "          # Loop nos browsers\n",
        "          for browser in ls_browsers:\n",
        "\n",
        "            # Obtem os clicks da data/dominio/browser\n",
        "            js_clicks_browser,df_click_browser =  obter_clicks_data_dominio_browser(id_atual,browser,data_atual.strftime(\"%Y-%m-%d\"),limite_lote)\n",
        "\n",
        "            # Verifica se tiveram cliques\n",
        "            if len(df_click_browser) == 0:\n",
        "              log(f\"Não foi localizados cliques no dia {data_atual.strftime('%Y-%m-%d')}, do dominio {dominio_atual}, do browser {browser}\")\n",
        "\n",
        "            # Localizado cliques\n",
        "            if len(df_click_browser) > 0:\n",
        "\n",
        "              log(f\"Obtido {len(df_click_browser):,} cliques do dia {data_atual.strftime('%Y-%m-%d')}, do dominio {dominio_atual}, do browser {browser}\")\n",
        "\n",
        "              # Insere registros no BigQuery\n",
        "              inserir_dados_bigquery(df_click_browser)\n",
        "\n",
        "      # Se não atingiu o total do limite do lote\n",
        "      if len(df_click) < limite_lote:\n",
        "\n",
        "          # Insere registros no BigQuery\n",
        "          inserir_dados_bigquery(df_click)\n",
        "\n",
        "  # Incrementa a data atual\n",
        "  data_atual = data_atual + datetime.timedelta(days=1)"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}