Guides
«Kiss My Stat!» P.3 VK Ads API + ClickHouse + Async (draft version)
Дима Родин
Сторож в Digital God
  • Скринкаст: про регистрацию приложения VK и авторизацию
  • Регистрация приложения, авторизация
  • Загрузка структуры рекламных кампаний и поиск ссылок объявлений
  • Загрузка статистики, объединение со структурами для получения utm_*
  • Запись статистики VK в ClickHouse
  • Простенький dashboard в Grafana на данных из ClickHouse
  • Создание сервиса для автоматического сбора статистики по расписанию.
  • Скринкаст: запуск в виде сервиса
Канал с анонсами
TG: @kissmystat
Анонсы, дополнительные материалы,
важные оповещение
Чат для вопросов
TG: @kissmystats
Обсуждение, вопросы, помощь, флуд,
закидывание какашками
Бот для навигации
TG: совсем скоро
Ссылки на трансляции,
персональные коммуникации

Ссылки и материалы

Кабинет разработчика в VK

Документация по API. Будем использовать Ads и Wall

Статья про OAuth - авторизацию, которую использует VK и другие.

Нам нужно получить статистику в разрезе UTM меток, чтобы можно было сопоставить с Google Analytics / Yandex Metrika. Сразу скажу, что ссылок в статистике нет и придется совершить путешествие, чтобы найти ссылки и достать из них utm_ метки или шаблон, по которому они строятся.

Шаги:

  • Регистрация приложения VK в кабинете разработчика https://vk.com/dev
  • Используя данные приложения пройти OAuth авторизацию с использованием кода
  • Собрать все структуры рекламы, отыскать там ссылки
  • Собрать статистику, объединить со структурами (достать метки)
  • Сохранить результат в ClickHouse
  • Сделать, чтобы работало автоматически, запустив в виде сервиса
  • Сделать обработку редиректа для приема кода в процессе авторизации
  • Построить легенький дашборд, чтобы посмотреть как из графаны добраться до данных VK в ClickHouse
В руководстве используется ClickHouse, Jupyter, Grafana и другие инструменты. Они все имеются в стандартной установке Rockstat. Если вы не читали предыдущие руководства, где произовдилась установка Rockstat, сделайте это при помощи видео-руководства установка Rockstat в Google Cloud. Важно! Обязательно обновите Rockstat перед началом!

Установим и импортируем набор полезных функций для вывода данных. Любопытных прошу пройти по адресу https://github.com/madiedinro/rodin_helpers_py, не забудьте звездочку постаить, или не пользуйтесь тогда ;)

In [2]:
!pip install -U git+https://github.com/madiedinro/rodin_helpers_py -q
import rodin_helpers as rh

Импортируем необходимые зависимости

In [3]:
import requests
import os
import json
import arrow
from collections import defaultdict
from urllib.parse import urlparse, parse_qs, quote
from dotenv import load_dotenv
from pprint import pprint

Подготовим парочку простеньких функций, которые позволят не получать токен каждый раз:

  • Для сохранения токена
  • Для чтения сохраненного токена
In [4]:
storage_fn = 'vk_data.json'

def save_dict(d):
    raw = json.dumps(d)
    with open(storage_fn, 'w') as f:
        f.write(raw)
        
# Загрузка данных из файла, если таковой имеется. 
# Если файл отсутствует это нормально и ошибки не будет
def read_dict():
    try:
        with open(storage_fn, 'r') as f:
            raw = f.read()
            return json.loads(raw)
    except:
        pass
    return {}

Данные выданные при регистрации приложения

Итак, вы получили от VK параметры приложения, подставьте их в код или запишите в файл creds_fn

In [6]:
# версия API, требуется ее знать и передавать в запросах
vk_v = '5.92'
# Параметры приложения
VK_APP_ID = '666777'
VK_APP_SECRET = 'oe0OsaelohSooph4noh1eu'

VK_TOKEN = read_dict().get('access_token')

# Подсказка для самих себя. Дальше поймете
if VK_TOKEN:
    print('Токен уже есть, получать не требуется')

# Форматируем даты в формат, требуемый VK: 2019-01-01
d1 = arrow.get('2019-04.15').format('YYYY-MM-DD')
d2 = arrow.get('2019-04-21').format('YYYY-MM-DD')# Формируем объект запроса к API
Токен уже есть, получать не требуется

Авторизация

Требуется пройти по ссылке, она отфутболит на страницу где можно взять код (вообще задумано что автоматом должно получаться). Его обработка не сделана, поэтому просто возьмите его из строки адреса браузера

In [8]:
# пропускаем ячейку, если токен уже есть
from urllib.parse import urlencode
if not VK_TOKEN:
    
    code_params = {
        'redirect_uri': 'https://oauth.vk.com/blank.html',
        'client_id': VK_APP_ID,
        'scope': 'ads,offline',
        'response_type': 'code'
    }
    vk_auth_url = f'https://oauth.vk.com/authorize?' + urlencode(code_params)
    print('redirect to:', vk_auth_url)
    print('-----------------------')
    print(rh.show_link(vk_auth_url))

Указываем код и получаем токен

In [9]:
code = 'd37e8edce44ecf4a02'

# пропускаем ячейку, если токен уже есть
if not VK_TOKEN:

    token_params = {
        'client_id': VK_APP_ID,
        'client_secret': VK_APP_SECRET,
        'code': code,
        'redirect_uri': code_params["redirect_uri"],
        
    }
    
    token_url= f'https://oauth.vk.com/access_token?' + urlencode(token_params)
    responce = requests.get(token_url)
    
    if responce.status_code == 200:
        auth_result = responce.json()
        # Сохраняем токен в файл, чтобы в следующий раз не требовалось
        # проходить процедуру получения токена через подтверждение кодом
        save_dict(auth_result)
        VK_TOKEN = auth_result['access_token']
        user_id = auth_result['user_id']
        # Обрежем токен, чтобы удостовериться, что он есть, но не палить целиком
        print('Ваши ключи для доступа сохранены в переменную.', user_id, VK_TOKEN[:20])

    # Если код ответа отличается от 200, печатаем сообщение с ошибкой
    else:
        print('error!', responce.status_code, responce.text)

Попробуем совершить запрос к API

Запросим информацию о рекламном аккаунте

In [10]:
# Словарь базовых параметров: авторизационный токен и нужная версия API. 
# Требуется передавать со всеми запросами.
common_params = {
    'access_token': VK_TOKEN,
    'v': vk_v
}
vk_accounts_url = 'https://api.vk.com/method/ads.getAccounts'
resp = requests.get(vk_accounts_url, params=common_params)
pprint(resp.json())
{
  'response': [
    {
      'access_role': 'admin',
      'account_id': 1603421955,
      'account_name': 'Личный кабинет',
      'account_status': 1,
      'account_type': 'general'
    }
  ]
}

Легко, не правда ли?)

API VK всегда выдает инфоммацию в виде {'response': ...

Готовим функцию для запросов к API VK

Ф-ция и переменные, которые потребуются для дальнейшего общения с VK

In [11]:
# Шаблон для адресов методов VK (можно узнать в документации, 
# ссылки на которую в самом верху)
vk_call_tmpl = 'https://api.vk.com/method/{method}'
# Ставим пометку, нужно ли запрашивать удаленные элементы
vk_include_deleted = 0

# Функция выполнения запроса к VK
def vk_req(method, params=None, data = None, http_method = 'get'):
    # мы специально не используем дефолтовые параметры = {}, 
    # там есть подводные камни, которые мы не будем разбирать. 
    # Просто поверьте.
    if params is None:
        params = {}
    # Что касается http запросов, что такое data, что такое params смело 
    # ступайте в документацию по протоколу HTTP и библиотеки requests :) или на курсы digital god :)
    if data is None:
        data = {}
    # Если запрашиваем постом, то перемещаем параметры в тело запроса. 
    if http_method == 'post':
        # Дополняем data содержимым params
        data.update(params)
        params = {}
    # Но в параметрах все равно передадутся базовые параметры (версия и токен)
    final_params = {**common_params, **params}
    # выполняем запрос, который вернет объект ответа Response
    resp = requests.request(http_method, vk_call_tmpl.format(method=method), data=data,  params=final_params)
    # Воспользуемся ислючениями для остановки выполнения программы.
    # Это позволит избежать можества дополнительных проверок дальше
    if resp.status_code != 200:
        print(resp.status_code, resp.text())
        raise Exception('Wrong response code')
    # преобразовываем текстовый json в формат данных python
    result = resp.json()
    # API всегда выдает ответ в {'response':, можно это сразу и обработать
    if 'response' not in result:
        print(resp.status_code, result)
        raise Exception('No response in result')
    return result['response']

Запрашиваем аккаунты текущего пользователя, чтобы удостовериться, что все так

In [19]:
vk_accs = vk_req('ads.getAccounts')
account_id = vk_accs[0]['account_id']

pprint(vk_accs)
[
  {
    'access_role': 'admin',
    'account_id': 1603421955,
    'account_name': 'Личный кабинет',
    'account_status': 1,
    'account_type': 'general'
  }
]

Получаем список кампаний

Интересующие я заранее пометил префиксом ||dg|| в интерфейсе VK, так в обработку попадут только нужные мне кампании

In [36]:
# Словарь, куда будем дописывать структуру кампаний
vk_camps = {}
# Отдельно создадим список с айдишками кампаний, он нам дальше пригодится
vk_camps_ids = []
# Выполняем запрос
call_params={'account_id': account_id, 'include_deleted': vk_include_deleted}
vk_camps_resp = vk_req('ads.getCampaigns', call_params)
for camp in vk_camps_resp:
    if True or camp.get('name', '').startswith('||dg||'):
        
        camp_id = str(camp['id'])
        vk_camps_ids.append(camp_id)
        vk_camps[camp_id] = camp

print('vk_camps_ids:', vk_camps_ids)
# вернуть только первую кампанию
pprint(vk_camps[vk_camps_ids[0]])
vk_camps_ids: ['1005274056', '1005373774', '1005373994', '1010819423', '1010920266', '1011946968']
{
  'all_limit': '0',
  'create_time': '1458118362',
  'day_limit': '100',
  'id': 1005274056,
  'name': 'Brand',
  'start_time': '0',
  'status': 1,
  'stop_time': '0',
  'type': 'normal',
  'update_time': '1460039405'
}
In [37]:
pprint(vk_camps)
{
  '1005274056': {
    'all_limit': '0',
    'create_time': '1458118362',
    'day_limit': '100',
    'id': 1005274056,
    'name': 'Brand',
    'start_time': '0',
    'status': 1,
    'stop_time': '0',
    'type': 'normal',
    'update_time': '1460039405'
  },
  '1005373774': {
    'all_limit': '0',
    'create_time': '1460037020',
    'day_limit': '100',
    'id': 1005373774,
    'name': 'Callback',
    'start_time': '0',
    'status': 1,
    'stop_time': '0',
    'type': 'normal',
    'update_time': '1460039401'
  },
  '1005373994': {
    'all_limit': '0',
    'create_time': '1460039285',
    'day_limit': '200',
    'id': 1005373994,
    'name': 'Calltracking',
    'start_time': '0',
    'status': 1,
    'stop_time': '0',
    'type': 'normal',
    'update_time': '1460544665'
  },
  '1010819423': {
    'all_limit': '0',
    'create_time': '1542506368',
    'day_limit': '1000',
    'id': 1010819423,
    'name': '||dg||-pingui',
    'start_time': '0',
    'status': 0,
    'stop_time': '0',
    'type': 'promoted_posts',
    'update_time': '1551213143'
  },
  '1010920266': {
    'all_limit': '0',
    'create_time': '1543512178',
    'day_limit': '1000',
    'id': 1010920266,
    'name': '||dg||-lookalike',
    'start_time': '0',
    'status': 0,
    'stop_time': '0',
    'type': 'promoted_posts',
    'update_time': '1551213141'
  },
  '1011946968': {
    'all_limit': '0',
    'create_time': '1554432291',
    'day_limit': '0',
    'id': 1011946968,
    'name': '||dg|| spring',
    'start_time': '0',
    'status': 1,
    'stop_time': '0',
    'type': 'promoted_posts',
    'update_time': '1554442030'
  }
}

Достаем Объявления

Они нужны чтобы достать оттуда ссылки

In [31]:
# В этом словаре будет вся-вся инфа по объявлениям
ads = {}
# Отдельно сохраним айдишки объявлений в списке, чтобы запросить по ним статистику
vk_ads_ids = []
# Строим запрос к методу

call_params = {
    'account_id': account_id, 
    'include_deleted': vk_include_deleted, 
    'campaign_ids': json.dumps(vk_camps_ids)
}
print('vk_ads_req_params=', call_params)
# Выполняем запрос
vk_ads_resp = vk_req('ads.getAds', params=call_params)
for i, ad in enumerate(vk_ads_resp):
    vk_ads_ids.append(ad['id'])
    ad_id = ad['id']
    ads[ad_id] = ad # {
        # Сохраняем структуру объявления
#         'ad': ad,
#         Дописываем прямо к объявлению структуру кампании
#         'camp': camps.get(ad['campaign_id'])
#    }

print(f'processed {i} ads')
print('---')
# list(ads.keys())[0] - получить первый ключ справочника с объявлениями. Такая вот сложность, чтобы вывести всего один элемент
# pprint(ads[list(ads.keys())[0]])


ads_list = list(ads.values())
pprint(ads_list[1])
vk_ads_req_params= {'account_id': 1603421955, 'include_deleted': 0, 'campaign_ids': '[1010819423, 1010920266, 1011946968]'}
processed 21 ads
---
{
  'ad_format': 9,
  'ad_platform': 'desktop',
  'ad_platform_no_ad_network': 1,
  'age_restriction': '0',
  'all_limit': '0',
  'approved': '2',
  'campaign_id': 1011946968,
  'category1_id': '433',
  'category2_id': '0',
  'cost_type': 1,
  'cpm': '18000',
  'create_time': '1554442332',
  'day_limit': '0',
  'events_retargeting_groups': {'29547114': [2]},
  'id': '53179873',
  'impressions_limit': 3,
  'name': 'compilder-dgret-pc',
  'start_time': '0',
  'status': 1,
  'stop_time': '0',
  'update_time': '1554443156'
}

Только вот ссылок тут нет. Ищем дальше и находим метод вads.getAdsLayout

Чтобы каждый раз не париться с выводом первого элемента (я это делаю для наглядности, а если выводить все, то придется очень много скроллить) напишем специальную ф-цию.

In [27]:
def first(mydict):
    return mydict[list(mydict.keys())[0]]

Достаем разметку объявлений

Как оказалось их тут тоже нет, но есть ссылки на специально созданные под эти объявления рекламные посты. Дополнительно придется построить справочник соответсвия постов и объявлений wall_ids. Layout объявлениея просто допишем к словарю ads.

In [32]:
# Маска, по которой можно отличить, что объявление является постом
POST_PREFIX = 'http://vk.com/wall'
wall_ids = {}

call_params={
    'account_id': account_id,
    'include_deleted': vk_include_deleted,
    'campaign_ids': json.dumps(vk_camps_ids)
}

vk_ads_lays = {}
vk_ads_lay_resp = vk_req('ads.getAdsLayout', params=call_params)
    
# Обходим результаты запроса
for i, ad in enumerate(vk_ads_lay_resp):
    
    ad_id = ad.get('id')
    camp_id = ad.get('campaign_id')
    
    # Проверяем что у нас есть такая кампания
    if not camp_id or camp_id not in vk_camps:
        continue

    # Проверяем что такое объявдение получено
    if ad_id not in ads:
        continue

    # Достаем ссылку из словаря по ключу link_url
    link_url = ad.get('link_url')
    
    if link_url and link_url.startswith(POST_PREFIX):
        # Находим ID записи на стене
        wall_id = link_url[len(POST_PREFIX):]
        # Берем список из словаря, если элемент отсутствует, то будет выдано дефолтовое значение - пустой список
        wall_id_ids = wall_ids.get(wall_id, [])
        wall_id_ids.append(ad_id)
        # Записываем в словарь
        wall_ids[wall_id] = wall_id_ids
        # Сохраняем в общем справочнике объявлений
        ad['wall_id'] = wall_id
        
    ads[ad_id]['layout'] = ad

print(f'processed {i} ads')
print('---')
pprint(first(ads))
print('---')
pprint(wall_ids)
processed 21 ads
---
{
  'ad_format': 9,
  'ad_platform': 'mobile',
  'ad_platform_no_ad_network': 1,
  'age_restriction': '0',
  'all_limit': '0',
  'approved': '2',
  'campaign_id': 1011946968,
  'category1_id': '433',
  'category2_id': '0',
  'cost_type': 1,
  'cpm': '18000',
  'create_time': '1554442360',
  'day_limit': '0',
  'events_retargeting_groups': {'29547114': [2]},
  'id': '53179883',
  'impressions_limit': 3,
  'layout': {
    'ad_format': 9,
    'age_restriction': '0',
    'campaign_id': 1011946968,
    'cost_type': 1,
    'id': '53179883',
    'image_src': '',
    'link_domain': '',
    'link_url': 'http://vk.com/wall-173839661_39',
    'preview_link':
      'https://vk.com/ads?act=preview_ad&mid=181916064&id=53179883&t='
      '1555948170&hash=ae736f3194ea114043',
    'title': '',
    'wall_id': '-173839661_39'
  },
  'name': 'compiler-dgret-mob',
  'start_time': '0',
  'status': 1,
  'stop_time': '0',
  'update_time': '1554443156'
}
---
{
  '-173839661_16': ['48601957'],
  '-173839661_17': ['48602092'],
  '-173839661_18': ['48602127'],
  '-173839661_19': ['48617289'],
  '-173839661_20': ['48617331'],
  '-173839661_22': ['48617395'],
  '-173839661_23': ['48617432'],
  '-173839661_25': ['49726014'],
  '-173839661_26': ['49726084'],
  '-173839661_32': ['53178004'],
  '-173839661_33': ['53178032'],
  '-173839661_34': ['53179727'],
  '-173839661_35': ['53179753'],
  '-173839661_36': ['53179820'],
  '-173839661_37': ['53179840'],
  '-173839661_38': ['53179873'],
  '-173839661_39': ['53179883'],
  '-173839661_5': ['48455205'],
  '-173839661_6': ['48455209'],
  '-173839661_7': ['48586777'],
  '-173839661_8': ['48586790'],
  '-173839661_9': ['48598806']
}

Достаем посты

Делаем это только по подготовленной таблице соответствия постов (словарь, где ключ это id поста).

In [17]:
# Как и с выводом первого элемента, берем ключи словаря и преобразовываем их в список [*wall_ids.keys()]
vk_posts = {}
vk_posts_resp = vk_req('wall.getById', {'posts': ",".join([*wall_ids.keys()])})
for post in vk_posts_resp:
    post_id = '{from_id}_{id}'.format(**post)
    # По ранее составленному справочнику связей постов и объявлений находим соответствие
    for ad_id in wall_ids[post_id]:
        ads[ad_id]['post'] = post
        # Ищем ссылку в аттачах
        for attach in post.get('attachments'):
            if attach['type'] == 'link':
                link = attach['link']
                # Дописываем к объявлению
                ads[ad_id]['url'] = link.get('url')

print(f'processed {i} posts\n---')
pprint(post)
processed 21 posts
---
{
  'attachments': [
    {
      'link': {
        'button': {
          'action': {
            'type': 'open_url',
            'url': 'https://digitalgod.be/rockstar'
          },
          'title': 'Подробнее'
        },
        'caption': 'digitalgod.be',
        'description': 'Программа изучения программирования, созданная специально для Digital',
        'is_favorite': False,
        'photo': {
          'album_id': -28,
          'date': 1543193727,
          'id': 456258906,
          'owner_id': 2000024793,
          'sizes': [
            {
              'height': 90,
              'type': 'a',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c81e/mZi08Chu5B8.jpg',
              'width': 200
            },
            {
              'height': 179,
              'type': 'b',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c81c/gy_ZryJ1ck8.jpg',
              'width': 400
            },
            {
              'height': 200,
              'type': 'c',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c820/pDqEEr-snUQ.jpg',
              'width': 200
            },
            {
              'height': 100,
              'type': 'd',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c821/sQuBQ0TL1s4.jpg',
              'width': 100
            },
            {
              'height': 50,
              'type': 'e',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c822/R8Kpq928UWI.jpg',
              'width': 50
            },
            {
              'height': 480,
              'type': 'k',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c81d/Zl9H0XelbBg.jpg',
              'width': 1072
            },
            {
              'height': 240,
              'type': 'l',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c81f/NvCFRaf5hT8.jpg',
              'width': 537
            },
            {
              'height': 1075,
              'type': 'o',
              'url': 'https://pp.userapi.com/c845019/v845019587/13c81b/FM1OH_rHAXI.jpg',
              'width': 2401
            }
          ],
          'text': ''
        },
        'title': 'Курс изучения программирования специально для digital специалистов',
        'url': 'https://digitalgod.be/rockstar'
      },
      'type': 'link'
    }
  ],
  'can_edit': 1,
  'comments': {'count': 0},
  'created_by': 181916064,
  'date': 1543193670,
  'from_id': -173839661,
  'id': 5,
  'is_favorite': False,
  'is_promoted_post_stealth': True,
  'likes': {'count': 0},
  'marked_as_ads': 0,
  'owner_id': -173839661,
  'post_type': 'post_ads',
  'reposts': {'count': 0},
  'text':
    'Мощно подготовиться к Digital Rockstar. Всего 4 дня по выходным. '
    'Освоим основы Python, работы с ОС Linux и конечно API.'
}

Как видите в посте содержится оооочень много всего. Но нас интересует лишь ссылка, ее то мы и достали

Взглянем что насобирали

В итоге по каждому объявлению получили вот такую структуру

In [18]:
pprint(first(ads))
---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
<ipython-input-18-4af350c2c75c> in <module>
----> 1 pprint(first(ads))

NameError: name 'first' is not defined

Поехали дальше. Доставать разметку

Заготовки для обработки разметки

Пример разметки:

utm_source=vk&utm_medium=cpm&utm_campaign={campaign_id}&utm_content={ad_id}
In [19]:
# Соответствия между параметрами статистики и шаблона
PARAMS_MAP = {
    'campaign_name': '{camp[name]}',
    'campaign_id': '{camp[id]}',
    'ad_id': '{ad[id]}'    
}
# Функция, которая получает на вход ссылку с шаблоном и объявление а на выходе дает полностью сформированную ссылку
def set_params(tmpl, ad):
    for prop, value in PARAMS_MAP.items():
        key_map = PARAMS_MAP.get(prop, prop.lower()) 
        value_esc = value.format(**ad)
        tmpl = tmpl.replace('{'+ prop + '}', value_esc)
    return tmpl

# Из ссылки достает utm_* метки. 
def extract_utms(url):
    parsed_url = urlparse(url)
    query = parse_qs(parsed_url.query)
    return {k: v[0] for k,v in query.items() if k.startswith('utm_')}

Давайте попробуем посмотреть, что делают эти функции.

In [20]:
tmpl = 'https://digitalgod.be/?utm_source=vk&utm_medium=cpm&utm_campaign={campaign_id}&utm_content={ad_id}'
set_params(tmpl, {'ad': {'id': 49726084}, 'camp': {'id':1010819423, 'name': 'ololo'}})
Out[20]:
'https://digitalgod.be/?utm_source=vk&utm_medium=cpm&utm_campaign=1010819423&utm_content=49726084'
In [21]:
pprint(extract_utms('https://digitalgod.be/?utm_source=vk&utm_medium=cpm&utm_campaign=1010819423&utm_content=49726084'))
{
  'utm_campaign': '1010819423',
  'utm_content': '49726084',
  'utm_medium': 'cpm',
  'utm_source': 'vk'
}

Как видите, мы успешно экстраполировали шаблон, и затем разобрали его при помощи функций разбора строк

In [22]:
stat_params = {
    'period': 'day',
    'account_id': account_id,
    'ids_type': 'ad',
    'ids': ",".join(vk_ads_ids), 
    'date_from': d1,
    'date_to': d2,
}
print('--- params ')
pprint(stat_params)
print('---')
vk_stats = vk_req('ads.getStatistics', stat_params)
# Список под статистику, ведь статистика это набор однотипных записей неизвестного кол-ва
vk_ads_stats = []
for ad_stats in vk_stats:
    # Находим объявление в заготовленном справочнике
    ad_id = str(ad_stats['id'])
    # Если у нас такое объявление есть
    ad = ads.get(ad_id)
    if not ad:
        continue
    # Проверяем, что у объявления имеется url 
    url = ad.get('url', None)
    if not url:
        continue
    # Подставляет шаблонные параметры, если имеются
    final_url = set_params(url, ad)
    # Достаем utm_*
    utms = extract_utms(final_url)
    # Проверяем что есть хоть какая-то статистика
    if not ad_stats.get('stats'):
        continue
    # pop - забрать из словаря (прям забрать) значение по ключу.
    ad_stat = ad_stats.get('stats')
    # Обходим пришедшие по объявлению данные
    for day_stat in ad_stat:
        # Готовим дополнение для словаря записи строки статистики
        upd = {
            'date': day_stat.pop('day'),
            'ad_id': ad_id,
            'campaign_id': ads[ad_id]['camp']['id'],
            'account_id': account_id
        }
        # Объединяем все данные
        rec = {**utms, **day_stat,**upd}
        vk_ads_stats.append(rec)
--- params 
{
  'account_id': 1603421955,
  'date_from': '2019-01-01',
  'date_to': '2019-01-25',
  'ids':
    '53179883,53179873,53179840,53179820,53179753,53179727,53178032,'
    '53178004,49726084,49726014,48617432,48617395,48617331,48617289,'
    '48602127,48602092,48601957,48598806,48586790,48586777,48455209,'
    '48455205',
  'ids_type': 'ad',
  'period': 'day'
}
---
In [23]:
pprint(vk_ads_stats[:2])
[
  {
    'account_id': 1603421955,
    'ad_id': '49726084',
    'campaign_id': 1010819423,
    'date': '2019-01-12',
    'impressions': 29,
    'reach': 16,
    'spent': '2.60',
    'utm_campaign': 'dr4',
    'utm_content': 'kiborg-vid',
    'utm_source': 'vk',
    'utm_term': 'retarg-mob'
  },
  {
    'account_id': 1603421955,
    'ad_id': '49726084',
    'campaign_id': 1010819423,
    'date': '2019-01-13',
    'impressions': 31,
    'reach': 26,
    'utm_campaign': 'dr4',
    'utm_content': 'kiborg-vid',
    'utm_source': 'vk',
    'utm_term': 'retarg-mob'
  }
]

Осмотрим структуры, которые были у нас по пути

In [24]:
print('--- статистика одного объявления ')
pprint(ad_stat)
print('--- дополнение к одной из записей в стате объявления')
pprint(upd)
print('--- финальная версия одной записи в стате ')
pprint(rec)
--- статистика одного объявления 
[
  {'impressions': 10, 'reach': 9},
  {
    'clicks': 2,
    'impressions': 23,
    'reach': 19
  },
  {
    'clicks': 3,
    'impressions': 42,
    'reach': 36,
    'spent': '16.80'
  },
  {
    'impressions': 33,
    'reach': 30,
    'spent': '21.60'
  },
  {'impressions': 31, 'reach': 29},
  {
    'clicks': 1,
    'impressions': 24,
    'reach': 17,
    'spent': '24.80'
  },
  {
    'clicks': 1,
    'impressions': 18,
    'reach': 15,
    'spent': '7.60'
  },
  {'impressions': 13, 'reach': 10},
  {
    'impressions': 10,
    'reach': 10,
    'spent': '2.80'
  },
  {'impressions': 7, 'reach': 7},
  {
    'impressions': 7,
    'reach': 7,
    'spent': '4.00'
  },
  {'impressions': 7, 'reach': 7},
  {'impressions': 4, 'reach': 4},
  {
    'clicks': 1,
    'impressions': 8,
    'reach': 8
  },
  {'impressions': 5, 'reach': 5},
  {
    'impressions': 12,
    'reach': 10,
    'spent': '6.00'
  },
  {'impressions': 6, 'reach': 6},
  {'impressions': 8, 'reach': 8},
  {
    'impressions': 6,
    'reach': 6,
    'spent': '5.40'
  },
  {'impressions': 7, 'reach': 7},
  {'impressions': 9, 'reach': 7},
  {'impressions': 6, 'reach': 5}
]
--- дополнение к одной из записей в стате объявления
{
  'account_id': 1603421955,
  'ad_id': '48598806',
  'campaign_id': 1010819423,
  'date': '2019-01-25'
}
--- финальная версия одной записи в стате 
{
  'account_id': 1603421955,
  'ad_id': '48598806',
  'campaign_id': 1010819423,
  'date': '2019-01-25',
  'impressions': 6,
  'reach': 5,
  'utm_campaign': 'dr4',
  'utm_content': 'wonderwoman-vid',
  'utm_source': 'vk',
  'utm_term': 'retarg-mob'
}

Вот что получили в итоге. Статистика + метки, все что нужно!

In [25]:
rh.print_rows(vk_ads_stats, limit=10)
account_id ad_id campaign_id clicks date impressions reach spent utm_campaign utm_content utm_source utm_term
1603421955 49726084 1010819423 - 2019-01-12 29 16 2.60 dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-13 31 26 - dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-14 43 29 17.20 dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-15 18 13 - dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-16 23 16 - dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-17 21 18 11.20 dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 1 2019-01-18 12 11 - dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-19 15 10 - dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-20 12 8 - dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-21 15 13 - dr4 kiborg-vid vk retarg-mob
1603421955 49726084 1010819423 - 2019-01-22 7 7 13.00 dr4 kiborg-vid vk retarg-mob

Запись в ClickHouse

Импортируем библиотеку simplech и создаем инстанс клиента. Сразу воспользуемся асинхронной версией, чтобы потом меньше переделывать.

In [26]:
from simplech import ClickHouse
ch = ClickHouse()
In [3]:
str(False).lower()
Out[3]:
'false'

Воспользуемся TableDiscovery из модуля simplech для автоматической генерации схемы БД

Тут скрыт, важный и интересный кусочек руководства.Чтобы открыть потребуется выполнить задание! Для этого авторизуйтесь у бота в tg @digitalgodbot и запросите доступ командой /access guides_vk_ads_api. Блок откроется, когда вы выполните задание.

Выполняем SQL запрос

In [29]:
td.merge_tree(execute=True)
In [30]:
query = f'SELECT count() FROM {td.table}'
print('query:', query)
result = ch.select(query)
print('result:', result)
query: SELECT count() FROM vk_stat
result: 1390

При использовании td.difference(d1, d2) и td.push нет необходимости самостоятельно вызывать ch.flush в конце, он выполнится автоматически

In [31]:
# в целях наглядности запишем сюда то, что отправилось на запись в БД
written = []

# инициализируем оброаботчик разницы
with td.difference(d1.format('YYYY-MM-DD'), d2.format('YYYY-MM-DD'), vk_ads_stats) as d:
    # он является асинхронным генератором и его следует обойти циклом
    for row in d:
        # добавляем в буфер на запись
        td.push(row)
        # пишем себе, чтобы потом поглядеть
        written.append(row)
In [32]:
rh.print_rows(written, limit=10)

Посмотрим что в таблице

In [33]:
query = 'SELECT * FROM vk_stat  ORDER BY date'
rows = [r for r in ch.objects_stream(query)]
rh.print_rows(rows, limit=10)
account_id ad_id campaign_id clicks date impressions join_rate reach spent utm_campaign utm_content utm_source utm_term
1603421955 49726014 1010819423 0 2019-01-04 -22 0 -19 -8.8 dr4 kiborg-vid vk retarg-mob
1603421955 48617395 1010920266 1 2019-01-04 133 0 108 24 dr4 wonderwoman-vid vk lookalike-retarg-pc
1603421955 48602092 1010819423 0 2019-01-04 13 0 12 1.6 dr4 kiborg-vid vk retarg-pc
1603421955 49726014 1010819423 0 2019-01-04 22 0 19 8.8 dr4 kiborg-vid vk retarg-mob
1603421955 48617395 1010920266 -1 2019-01-04 -133 0 -108 -24 dr4 wonderwoman-vid vk lookalike-retarg-pc
1603421955 48617331 1010920266 -1 2019-01-04 -113 0 -99 -16.8 dr4 krot vk lookalike-retarg-pc
1603421955 48598806 1010819423 0 2019-01-04 -10 0 -9 - dr4 wonderwoman-vid vk retarg-mob
1603421955 48617289 1010920266 -4 2019-01-04 -340 0 -291 -62.4 dr4 krot vk lookalike-retarg-mob
1603421955 48617432 1010920266 -5 2019-01-04 -274 0 -244 -49.2 dr4 wonderwoman-vid vk lookalike-retarg-mob
1603421955 48602092 1010819423 0 2019-01-04 -13 0 -12 -1.6 dr4 kiborg-vid vk retarg-pc
1603421955 48602127 1010819423 0 2019-01-04 -25 0 -22 - dr4 kiborg-vid vk retarg-mob

Результат

У нас все получилось. Данные были получены и записаны в ClickHouse. Обрабатывается "дописывание" свежих данных. Многим и этого хватит а кому не хватит, пошли дальше!

Автоматическая работа по расписанию в виде сервиса

Теперь надо превратить это из формата блокнота, где надо все выполнять руками в удобный набор функций, которые можно будет перенести в сервис, где они будут выполняться по расписанию. Более того, будет автоматически обрабатываться получение токена.

Заранее прошу прощения, но подробно расписывать не буду, только какие-то специфичные места. Для всего остального есть документация (ее мало, но есть) и доки питона.

Соберем вместе все, что нам понадобится. Просто соберем, тут не все сможет запуститься

import asyncio
from band import expose, cleanup, worker, settings as cfg, logger, response
import aiohttp
from collections import defaultdict
from urllib.parse import urlparse, parse_qs, quote
import json
from simplech import AsyncClickHouse
import arrow
from aiocron import crontab




vk_v = '5.92'
# Превращаем в обычный шаблон, чтобы применить параметры в момент использования
token_url_tmpl = 'https://oauth.vk.com/access_token?client_id={id}&client_secret={secret}&code={code}&redirect_uri={redir}'
# ссылка, куда отправлять для авторизации
vk_auth = f'https://oauth.vk.com/authorize?client_id={cfg.app_id}&redirect_uri={cfg.app_redir}&scope=ads,offline&response_type=code&v={vk_v}'
vk_call_tmpl = 'https://api.vk.com/method/{method}'
# Шаблон для адресов методов VK (можно узнать в документации, ссылки на которую в самом верху)
vk_call_tmpl = 'https://api.vk.com/method/{method}'
# Ставим пометку, нужно ли запрашивать удаленные элементы
vk_include_deleted = 0
# Папка с конфигами
conf_dir = ''
# Файлы с переменными окружения. Использются для конфигурации приложений
# creds_fn = f'{conf_dir}.env.vk' - в сервисах рокстат есть специальные конфиги
token_fn = f'{conf_dir}.env.vk_token'
POST_PREFIX = 'http://vk.com/wall'
# Соответствия между параметрами статистики и шаблона
params_map = {
    'campaign_name': '{camp[name]}',
    'campaign_id': '{camp[id]}',
    'ad_id': '{ad[id]}'
}

# С редиректом тут будет отдельная история, можно сразу его обработать, не копируя ключ из строки адреса
# https://{{DOMAIN}}/vk_collect/get_code
# Тут:
# {{DOMAIN}} - домен с рокстат - подставляется автоматом
# vk_collect - название сервиса в рокстат
# get_code - название метода в сервисе
# >>> Пометка: не забыть поместить в конфиг APP_ID, APP_SECRET
common_params = {
    'access_token': None,
    'v': vk_v
}

# создаем клиента ClickHouse
ch = AsyncClickHouse()

Добавить в config.yml

app_redir: https://{{DOMAIN}}/vk_collect/get_code
app_id: {{APP_ID}}
app_secret: {{APP_SECRET}}

Создать файл .env с содержимым

APP_ID=ваши данные от приложения vk
APP_SECRET=ваши данные от приложения vk

Перепишем функцию выполнения запроса к API в асинхронный формат

In [34]:
async def vk_req_async(method, params=None, data=None, http_method='get'):
    if params is None:
        params = {}
    if http_method == 'post':
        data = data or {}
        data.update(params)
        params = {}
    final_params = {**common_params, **params}
    url = vk_call_tmpl.format(method=method)
    logger.debug('requesting', u=url, p=final_params)
    async with aiohttp.ClientSession() as session:
        async with session.request(http_method, url, json=data, params=final_params) as res:
            if res.status != 200:
                print(res.status, await res.text())
                raise Exception('Wrong response code')
            result = await res.json()
            if 'response' not in result:
                print(res.status, result)
                raise Exception('No response in result')
            return result['response']

Пример использования

await vk_req_async('ads.getAccounts')
[
  {
    'access_role': 'admin',
    'account_id': 1603421955,
    'account_name': 'Личный кабинет',
    'account_status': 1,
    'account_type': 'general'
  }
]

Набор необходимых функций для хранения токена и разбора url

def save_token(access_token):
    with open(token_fn, 'w') as f:
        f.write(access_token)


def load_token():
    try:
        with open(token_fn) as f:
            # read читает все содержимое файла. strip убирает переносы строк по краям
            return f.read().strip()
    except Exception:
        logger.exception('ex')


# Функция, которая получает на вход ссылку с шаблоном и объявление и на выходе дает полностью сформированную ссылку
def set_params(url, ad):
    for prop, value in params_map.items():
        key_map = params_map.get(prop, prop.lower())
        value_esc = value.format(**ad)
        url = url.replace('{'+ key_map + '}', value_esc)
    return url


# Из ссылки достает utm_* метки.
def extract_utms(url):
    parsed_url = urlparse(url)
    query = parse_qs(parsed_url.query)
    return {k: v[0] for k,v in query.items() if k.startswith('utm_')}
In [35]:
async def load_data(d1, d2):
    # Аккаунт
    vk_accs = await vk_req_async('ads.getAccounts')
    account_id = vk_accs[0]['account_id']
    
    # Кампании
    camps = {}
    vk_camps_ids = []
    vk_camps = await vk_req_async('ads.getCampaigns', params={'account_id': account_id, 'include_deleted': vk_include_deleted})
    for camp in vk_camps:
        if camp.get('name', '').startswith('||dg||'):
            vk_camps_ids.append(camp['id'])
            camps[camp['id']] = camp

    # Немного подождем
    await asyncio.sleep(0.5)
    
    # Объявления
    ads = {}
    vk_ads_ids = []
    vk_ads_req_params = {
        'account_id': account_id,
        'include_deleted': vk_include_deleted,
        'campaign_ids': json.dumps(vk_camps_ids)
    }
    
    
    
    vk_ads_resp = await vk_req_async('ads.getAds', params=vk_ads_req_params)
    for i, ad in enumerate(vk_ads_resp):
        vk_ads_ids.append(ad['id'])
        ad_id = ad['id']
        ads[ad_id] = {
            'ad': ad,
            'camp': camps.get(ad['campaign_id'])
        }

    wall_ids = defaultdict(list)

    al_params = {
        'account_id': account_id,
        'include_deleted': vk_include_deleted,
        'campaign_ids': json.dumps(vk_camps_ids)
    }
    await asyncio.sleep(1)
    vk_ads_lay = await vk_req_async('ads.getAdsLayout', params=al_params)

    for i, ad in enumerate(vk_ads_lay):
        ad_id = ad.get('id')
        camp_id = ad.get('campaign_id')
        if not camp_id or camp_id not in vk_camps_ids:
            continue
        if ad_id not in ads:
            continue
        link_url = ad.get('link_url')
        if link_url and link_url.startswith(POST_PREFIX):
            wall_id = link_url[len(POST_PREFIX):]
            wall_ids[wall_id].append(ad_id)
            ad['wall_id'] = wall_id
        ads[ad_id]['layout'] = ad

    await asyncio.sleep(1)
    vk_posts = await vk_req_async('wall.getById', {'posts': ",".join([*wall_ids.keys()])})
    for post in vk_posts:
        post_id = '{from_id}_{id}'.format(**post)
        for ad_id in wall_ids[post_id]:
            ads[ad_id]['post'] = post
            for attach in post.get('attachments'):
                if attach['type'] == 'link':
                    link = attach['link']
                    ads[ad_id]['url'] = link.get('url')

    # Формируем объект запроса к API
    stat_params = {
        'period': 'day',
        'account_id': account_id,
        'ids_type': 'ad',
        'ids': ",".join(vk_ads_ids),
        'date_from': d1.format('YYYY-MM-DD'),
        'date_to': d2.format('YYYY-MM-DD'),
    }

    await asyncio.sleep(1)
    vk_stats = await vk_req_async('ads.getStatistics', stat_params)
    # Список под статистику, ведь статистика это набор однотипных записей неизвестного кол-ва
    vk_ads_stats = []
    for ad_stats in vk_stats:
        # Находим объявление в заготовленном справочнике
        ad_id = str(ad_stats['id'])
        # Если у нас такое объявление есть
        ad = ads.get(ad_id)
        if not ad:
            continue
        # Проверяем, что у объявления имеется url
        url = ad.get('url', None)
        if not url:
            continue
        # Подставляет шаблонные параметры, если имеются
        final_url = set_params(url, ad)
        # Достаем utm_*
        utms = extract_utms(final_url)
        # Проверяем что есть хоть какая-то статистика
        if not ad_stats.get('stats'):
            continue
        # pop - забрать из словаря (прям забрать) значение по ключу.
        ad_stat = ad_stats.pop('stats')
        # Обходим пришедшие по объявлению данные
        for day_stat in ad_stat:
            # Готовим дополнение для словаря записи строки статистики
            upd = {
                'date': day_stat.pop('day'),
                'ad_id': ad_id,
                'campaign_id': ads[ad_id]['camp']['id'],
                'account_id': account_id
            }
            # Объединяем все данные
            rec = {**utms, **day_stat,**upd}
            vk_ads_stats.append(rec)

    print(vk_ads_stats)
    return vk_ads_stats

Сделаем обработчик авторизации (возврата по ссылке, с кодом для получения токена)

# @expose.handler() - помечает фунуцию обработчиком внешних запросов.
# Она будет доступна по адресу, который мы указали в vk_redir = f'https://<ваш домен>/vk_collect/get_code'
# при условии, что название сервиса будет vk_collect
@expose.handler()
async def get_code(data, **params):
    # так работают обработчики в рокстат. Все входящие данные приходят в data
    # а вот **params это приемник всех остальных параметров, заранее неизвестно сколько их будет
    # поэтому у обработчиков запросов надо всегда добавлять словарь, куда они попадут
    code = data.get('code')

    token_url = token_url_tmpl.format(
        id=cfg.app_id,
        secret=cfg.app_secret,
        code=code,
        redir=cfg.app_redir)

    async with aiohttp.ClientSession() as session:
        async with session.get(token_url) as res:
            if res.status == 200:
                result = await res.json()
                print(result)
                common_params['access_token'] = result['access_token']
                save_token(result['access_token'])
            else:
                print(result.status, await result.text())

Осталось дело за сохранением. Получаем описание таблицы и копируем его в код

In [36]:
print(td.pycode())
td_vk_stat = ch.discover('vk_stat', columns={
        'utm_source': 'String', 
        'utm_campaign': 'String', 
        'utm_content': 'String', 
        'utm_term': 'String', 
        'spent': 'Float64', 
        'impressions': 'Int64', 
        'reach': 'Int64', 
        'date': 'Date', 
        'ad_id': 'Int64', 
        'campaign_id': 'Int64', 
        'account_id': 'Int64', 
        'clicks': 'Int64'})\
    .metrics('spent', 'clicks', 'impressions', 'reach')\
    .dimensions('campaign_id', 'utm_term', 'utm_campaign', 'utm_source', 'account_id', 'utm_content', 'ad_id', 'date')\
    .date('date')\
    .idx('account_id', 'date')

In [37]:
td_vk_stat = ch.discover('vk_stat', columns={
        'utm_source': 'String',
        'utm_campaign': 'String',
        'utm_content': 'String',
        'utm_term': 'String',
        'impressions': 'Int64',
        'reach': 'Int64',
        'date': 'Date',
        'ad_id': 'Int64',
        'campaign_id': 'Int64',
        'account_id': 'Int64',
        'spent': 'Float64',
        'clicks': 'Int64'})\
    .metrics('clicks', 'reach', 'impressions', 'spent')\
    .dimensions('ad_id', 'utm_content', 'account_id', 'utm_term', 'utm_campaign', 'campaign_id', 'date', 'utm_source')\
    .date('date')\
    .idx('account_id', 'date')


async def save_data(d1, d2, data):
    await td_vk_stat.merge_tree(execute=True)
    # инициализируем оброаботчик разницы
    async with td_vk_stat.difference(d1, d2, data) as d:
        # он является асинхронным генератором и его следует обойти циклом
        async for row in d:
            # добавляем в буффер на запись
            td_vk_stat.push(row)
        # Смотрем статистику, что у нас получилось
        logger.info('diff stat', stat=d.stat)

Воркер

# @crontab('5 5 * * *') - стандартрный формат кронтаб,
# где описывается расписание запуска. Тут указано запускать 5:05 ежедневно
@crontab('5 5 * * *')
# @worker() - делает функцию "воркером", если в ней имеется бесконечный цикл. Запускает при старте
@worker()
async def work():
    logger.info('auth link:', link=vk_auth)
    token = load_token()
    if not token:
        logger.warn('authorize please')
        return
    print(token)
    common_params['access_token'] = token

    d1 = arrow.now().shift(days=-2).format('YYYY-MM-DD')
    d2 = arrow.now().shift(days=-1).format('YYYY-MM-DD')

    data = await load_data(d1, d2)
    await save_data(d1, d2, data)

Достаем свои параметры

Будьте аккуратны, aiohttp очень быстрый, можно легко попасть на

{
  'error': {
    'error_code': 9,
    'error_msg': 'Flood control',
    'request_params': [
      {'key': 'oauth', 'value': '1'},
      {'key': 'method', 'value': 'ads.getAds'},
      {'key': 'v', 'value': '5.92'},
      {'key': 'account_id', 'value': '1603421955'},
      {'key': 'include_deleted', 'value': '0'},
      {'key': 'campaign_ids', 'value': '[1010819423, 1010920266]'}
    ]
  }
}

В этом случае используйте

await asyncio.sleep(1)

длительность ожидания придется подобрать


Скринкаст полного цикла создания сервиса


Простенький Dashboard в Grafana

Запросы:

SELECT
    $timeSeries as t,
    sum(impressions)
FROM $table
WHERE $timeFilter
GROUP BY t
ORDER BY t

---

SELECT
    $timeSeries as t,
    sum(spent)
FROM $table
WHERE $timeFilter
GROUP BY t
ORDER BY t

---

$columns(
    utm_content,
    sum(impressions) c)
FROM $table

---


$columns(
    utm_term,
    sum(impressions) c)
FROM $table

На этом все. Ждите новых серий :)


Канал с анонсами
TG: @kissmystat
Анонсы, дополнительные материалы,
важные оповещение
Чат для вопросов
TG: @kissmystats
Обсуждение, вопросы, помощь, флуд,
закидывание какашками
Бот для навигации
TG: совсем скоро
Ссылки на трансляции,
персональные коммуникации