Парсер Google Analytics: Python-скрипт для KPI-отчётов

📅 Опубликовано: 29 апреля 2026 🔄 Обновлено: 26 июля 2026

Аналитика

Парсер Google Analytics: Python-скрипт для KPI-отчётов

Коротко: Python-скрипт для автоматического парсинга Google Analytics может сократить время на создание KPI-отчётов с 3-4 часов до 15 минут. Используя Google Analytics Reporting API v4 и библиотеки pandas/matplotlib, можно собирать ключевые метрики (конверсии, трафик, воронки) в автоматические дашборды с точностью 99,8%.

Содержание

Зачем автоматизировать выгрузку данных из Google Analytics?
Как настроить Google Analytics API для Python-скрипта?
Какие библиотеки нужны для парсинга аналитики?
Как написать базовый парсер для KPI-метрик?
Как создать автоматический дашборд с BI-визуализацией?
Интеграция с Яндекс.Метрикой и A/B тестами
Как масштабировать решение для больших проектов?

Зачем автоматизировать выгрузку данных из Google Analytics?

За последние 5 лет мы в DS495 перевели на автоматизацию аналитических отчётов более 80 проектов. И каждый раз клиенты задают один вопрос: "А зачем вообще париться с парсерами, когда есть готовые дашборды в самой Google Analytics?"

Отвечу честно — потому что ручная работа с отчётами убивает продуктивность. Представьте: каждую неделю аналитик тратит 3-4 часа на то, чтобы:

Зайти в 5-7 разных счётчиков
Выставить нужные даты и фильтры
Скопировать данные в Excel
Свести всё в единый отчёт
Построить графики и посчитать динамику

За месяц это 12-16 часов рутины. За год — 150+ часов, которые можно потратить на анализ трендов, поиск точек роста конверсии или настройку A/B тестов.

Python-парсер решает эту боль кардинально. После настройки скрипт выполняет ту же работу за 15 минут без участия человека. Плюс даёт возможности, которых нет в стандартном интерфейсе GA:

Автоматический парсер может объединять данные из Google Analytics, Яндекс.Метрики и CRM в единый дашборд. Это позволяет видеть полную воронку от клика до оплаты в реальном времени.

Например, один из наших e-commerce клиентов экономит 40 часов в месяц на формировании отчётов по 12 рекламным каналам. А данные обновляются каждые 6 часов автоматически.

Задача	Ручная работа	Python-парсер	Экономия времени
Еженедельный KPI-отчёт	3,5 часа	15 минут	92%
Анализ воронки по каналам	2 часа	5 минут	96%
Мониторинг конверсий A/B тестов	1,5 часа	Автоматически	100%
Сводка по BI-показателям	4 часа	20 минут	90%

Как настроить Google Analytics API для Python-скрипта?

Первым делом нужно получить доступ к Google Analytics через API. Звучит страшно, но на самом деле Google сделал процесс довольно простым — если знать, куда нажимать.

Пошаговая инструкция для настройки API:

Создаём проект в Google Cloud Console — переходим на console.cloud.google.com, жмём "Новый проект" и даём ему понятное имя типа "Analytics Parser"
Включаем Google Analytics Reporting API — в разделе "APIs & Services" ищем "Google Analytics Reporting API" и активируем
Создаём Service Account — это технический аккаунт для скрипта. В разделе "Credentials" выбираем "Service Account" и скачиваем JSON-файл с ключами
Даём доступ к счётчику Analytics — копируем email сервисного аккаунта и добавляем его как пользователя в нужный счётчик GA с правами "Читатель"

Самая частая ошибка на этом этапе — забыть дать права сервисному аккаунту в самом Google Analytics. Скрипт будет ругаться на "403 Forbidden", и полчаса потратите на поиск проблемы.

Ещё один момент — обязательно включите двухфакторную аутентификацию на аккаунте Google, где создаёте API-проект. Google стал строже к безопасности и может заблокировать доступ без 2FA.

После настройки получаете JSON-файл примерно такого вида:

{
  "type": "service_account",
  "project_id": "analytics-parser-12345",
  "private_key_id": "...",
  "client_email": "analytics-parser@analytics-parser-12345.iam.gserviceaccount.com"
}

Этот файл — как ключи от квартиры. Никому не показывайте и не коммитьте в Git. Мы обычно кладём его в отдельную папку credentials/ и добавляем в .gitignore.

Какие библиотеки нужны для парсинга аналитики?

Для полноценного парсера Google Analytics нужен набор Python-библиотек. Каждая решает свою задачу, и вместе они дают мощный инструмент для работы с аналитикой.

Основные зависимости для проекта:

google-api-python-client — официальная библиотека для работы с Google API, включая Analytics
oauth2client — для аутентификации через сервисные аккаунты
pandas — обработка и анализ данных, без неё никуда в аналитике
matplotlib/seaborn — построение графиков и визуализаций для дашбордов
requests — для интеграции с Яндекс.Метрикой и другими API
schedule — автоматический запуск скриптов по расписанию

Устанавливаем всё одной командой:

pip install google-api-python-client oauth2client pandas matplotlib seaborn requests schedule openpyxl

Опционально можно добавить:

plotly — интерактивные графики для веб-дашбордов
sqlalchemy — если нужно сохранять данные в базу
telegram-bot — для отправки отчётов в мессенджер

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

Структура файлов в итоге выглядит примерно так:

analytics_parser/
├── credentials/
│   └── service_account.json
├── src/
│   ├── ga_parser.py
│   ├── dashboard.py
│   └── utils.py
├── reports/
├── requirements.txt
└── main.py

Как написать базовый парсер для KPI-метрик?

Теперь переходим к самому интересному — коду парсера. Покажу рабочий пример, который мы используем для большинства клиентов. Он собирает основные KPI за выбранный период и формирует удобную таблицу.

Базовый класс для работы с Google Analytics:

from googleapiclient.discovery import build
from oauth2client.service_account import ServiceAccountCredentials
import pandas as pd
from datetime import datetime, timedelta

class GAParser:
    def __init__(self, credentials_path, view_id):
        self.credentials_path = credentials_path
        self.view_id = view_id
        self.service = self._authenticate()
    
    def _authenticate(self):
        """Аутентификация через сервисный аккаунт"""
        scope = ['https://www.googleapis.com/auth/analytics.readonly']
        credentials = ServiceAccountCredentials.from_json_keyfile_name(
            self.credentials_path, scope
        )
        return build('analyticsreporting', 'v4', credentials=credentials)
    
    def get_kpi_report(self, start_date, end_date):
        """Получение основных KPI за период"""
        body = {
            'reportRequests': [{
                'viewId': self.view_id,
                'dateRanges': [{'startDate': start_date, 'endDate': end_date}],
                'metrics': [
                    {'expression': 'ga:sessions'},
                    {'expression': 'ga:users'},
                    {'expression': 'ga:pageviews'},
                    {'expression': 'ga:bounceRate'},
                    {'expression': 'ga:avgSessionDuration'},
                    {'expression': 'ga:goal1Completions'},  # Основная цель
                    {'expression': 'ga:goal1ConversionRate'}
                ],
                'dimensions': [{'name': 'ga:date'}]
            }]
        }
        
        response = self.service.reports().batchGet(body=body).execute()
        return self._parse_response(response)
    
    def _parse_response(self, response):
        """Преобразование ответа API в pandas DataFrame"""
        report = response['reports'][0]
        rows = report['data'].get('rows', [])
        
        data = []
        for row in rows:
            date = row['dimensions'][0]
            metrics = [float(val) for val in row['metrics'][0]['values']]
            data.append([date] + metrics)
        
        columns = ['date', 'sessions', 'users', 'pageviews', 
                   'bounce_rate', 'avg_duration', 'conversions', 'conversion_rate']
        
        df = pd.DataFrame(data, columns=columns)
        df['date'] = pd.to_datetime(df['date'])
        return df

Этот базовый парсер умеет:

Подключаться к Google Analytics через API
Запрашивать ключевые метрики за любой период
Возвращать данные в удобном формате pandas DataFrame

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

# Инициализация парсера
parser = GAParser('credentials/service_account.json', 'VIEW_ID')

# Получение данных за последнюю неделю
end_date = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')

kpi_data = parser.get_kpi_report(start_date, end_date)
print(kpi_data.head())

Результат — таблица с ключевыми показателями по дням:

Дата	Сессии	Пользователи	Конверсия, %	Отказы, %
2024-01-15	1,247	1,156	2.8%	45.2%
2024-01-16	1,389	1,298	3.1%	42.7%
2024-01-17	1,156	1,089	2.9%	44.8%

Нужна помощь с этой задачей? Команда DS495 решит её под ключ. Обсудить проект →

Как создать автоматический дашборд с BI-визуализацией?

Собрать данные — это половина дела. Главная ценность парсера в том, чтобы эти данные красиво визуализировать и автоматически обновлять. Покажу, как мы делаем дашборды для клиентов.

Для создания графиков используем связку matplotlib + seaborn. Это даёт достаточно гибкости для большинства задач аналитики:

import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime
import os

class AnalyticsDashboard:
    def __init__(self, data):
        self.data = data
        plt.style.use('seaborn-v0_8')
        sns.set_palette("husl")
    
    def create_kpi_dashboard(self, save_path='reports/'):
        """Создание комплексного дашборда с основными KPI"""
        fig, axes = plt.subplots(2, 3, figsize=(18, 12))
        fig.suptitle('KPI Dashboard - ' + datetime.now().strftime('%d.%m.%Y'), 
                     fontsize=16, fontweight='bold')
        
        # График трафика
        axes[0, 0].plot(self.data['date'], self.data['sessions'], 
                       marker='o', linewidth=2, markersize=4)
        axes[0, 0].set_title('Сессии по дням')
        axes[0, 0].grid(True, alpha=0.3)
        
        # График конверсий
        axes[0, 1].plot(self.data['date'], self.data['conversion_rate'], 
                       marker='s', linewidth=2, color='green')
        axes[0, 1].set_title('Конверсия, %')
        axes[0, 1].grid(True, alpha=0.3)
        
        # График отказов
        axes[0, 2].plot(self.data['date'], self.data['bounce_rate'], 
                       marker='^', linewidth=2, color='red')
        axes[0, 2].set_title('Показатель отказов, %')
        axes[0, 2].grid(True, alpha=0.3)
        
        # Воронка по источникам трафика
        traffic_sources = self._get_traffic_sources()
        axes[1, 0].pie(traffic_sources.values(), labels=traffic_sources.keys(), 
                      autopct='%1.1f%%', startangle=90)
        axes[1, 0].set_title('Источники трафика')
        
        # Тренд пользователей
        axes[1, 1].fill_between(self.data['date'], self.data['users'], 
                               alpha=0.5, color='blue')
        axes[1, 1].plot(self.data['date'], self.data['users'], 
                       linewidth=2, color='darkblue')
        axes[1, 1].set_title('Уникальные пользователи')
        axes[1, 1].grid(True, alpha=0.3)
        
        # Среднее время сессии
        axes[1, 2].bar(self.data['date'], self.data['avg_duration']/60, 
                      alpha=0.7, color='orange')
        axes[1, 2].set_title('Время сессии, мин')
        axes[1, 2].grid(True, alpha=0.3)
        
        plt.tight_layout()
        
        # Сохранение
        filename = f"kpi_dashboard_{datetime.now().strftime('%Y%m%d_%H%M')}.png"
        filepath = os.path.join(save_path, filename)
        plt.savefig(filepath, dpi=300, bbox_inches='tight')
        plt.close()
        
        return filepath

Дополнительно создаём Excel-отчёт с детальными данными:

def create_excel_report(self, filename='kpi_report.xlsx'):
    """Создание детального отчёта в Excel"""
    with pd.ExcelWriter(filename, engine='openpyxl') as writer:
        # Основные KPI
        self.data.to_excel(writer, sheet_name='Daily_KPI', index=False)
        
        # Сводная статистика
        summary = self._calculate_summary()
        summary.to_excel(writer, sheet_name='Summary', index=False)
        
        # Сравнение с предыдущим периодом
        comparison = self._get_period_comparison()
        comparison.to_excel(writer, sheet_name='Period_Comparison', index=False)
    
    return filename

def _calculate_summary(self):
    """Расчёт сводной статистики за период"""
    summary_data = {
        'Метрика': ['Всего сессий', 'Уникальные пользователи', 'Конверсия средняя', 
                   'Отказы средние', 'Время сессии среднее'],
        'Значение': [
            int(self.data['sessions'].sum()),
            int(self.data['users'].sum()),
            f"{self.data['conversion_rate'].mean():.2f}%",
            f"{self.data['bounce_rate'].mean():.2f}%",
            f"{self.data['avg_duration'].mean()/60:.1f} мин"
        ],
        'Динамика': [
            self._calculate_trend('sessions'),
            self._calculate_trend('users'),
            self._calculate_trend('conversion_rate'),
            self._calculate_trend('bounce_rate'),
            self._calculate_trend('avg_duration')
        ]
    }
    return pd.DataFrame(summary_data)

Автоматизируем создание отчётов с помощью scheduler:

import schedule
import time
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.image import MIMEImage

def automated_reporting():
    """Автоматическое создание и отправка отчётов"""
    # Парсим данные
    parser = GAParser('credentials/service_account.json', 'VIEW_ID')
    data = parser.get_kpi_report(
        (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d'),
        datetime.now().strftime('%Y-%m-%d')
    )
    
    # Создаём дашборд
    dashboard = AnalyticsDashboard(data)
    image_path = dashboard.create_kpi_dashboard()
    excel_path = dashboard.create_excel_report()
    
    # Отправляем по email
    send_report_email(image_path, excel_path)

# Настройка расписания
schedule.every().monday.at("09:00").do(automated_reporting)
schedule.every().wednesday.at("09:00").do(automated_reporting)
schedule.every().friday.at("09:00").do(automated_reporting)

# Запуск планировщика
while True:
    schedule.run_pending()
    time.sleep(60)

Результат — автоматические отчёты 3 раза в неделю с актуальными графиками и Excel-файлом для детального анализа. Клиенты получают их на почту без какого-либо участия аналитиков.

Интеграция с Яндекс.Метрикой и A/B тестами

Большинство российских проектов использует связку Google Analytics + Яндекс.Метрика. Логично объединить данные из обеих систем в единый дашборд. Плюс добавить мониторинг A/B тестов для полной картины.

Для работы с Яндекс.Метрикой используем Management API:

import requests
import json

class YandexMetrikaParser:
    def __init__(self, token, counter_id):
        self.token = token
        self.counter_id = counter_id
        self.base_url = "https://api-metrika.yandex.net/stat/v1/data"
    
    def get_basic_metrics(self, start_date, end_date):
        """Получение базовых метрик из Яндекс.Метрики"""
        params = {
            'id': self.counter_id,
            'metrics': 'ym:s:visits,ym:s:users,ym:s:pageviews,ym:s:bounceRate',
            'date1': start_date,
            'date2': end_date,
            'group': 'day',
            'oauth_token': self.token
        }
        
        response = requests.get(self.base_url, params=params)
        if response.status_code == 200:
            return self._parse_metrika_response(response.json())
        else:
            raise Exception(f"Ошибка API Метрики: {response.status_code}")
    
    def _parse_metrika_response(self, data):
        """Парсинг ответа Метрики в DataFrame"""
        metrics_data = []
        for item in data['data']:
            date_str = item['dimensions'][0]['name']
            metrics = item['metrics']
            metrics_data.append([
                date_str, metrics[0], metrics[1], metrics[2], metrics[3]
            ])
        
        df = pd.DataFrame(metrics_data, 
                         columns=['date', 'visits', 'users', 'pageviews', 'bounce_rate'])
        df['date'] = pd.to_datetime(df['date'])
        return df

Объединяем данные из GA и Метрики:

class UnifiedAnalytics:
    def __init__(self, ga_parser, ym_parser):
        self.ga_parser = ga_parser
        self.ym_parser = ym_parser
    
    def get_combined_report(self, start_date, end_date):
        """Объединённый отчёт из Google Analytics и Яндекс.Метрики"""
        # Данные из GA
        ga_data = self.ga_parser.get_kpi_report(start_date, end_date)
        
        # Данные из Метрики
        ym_data = self.ym_parser.get_basic_metrics(start_date, end_date)
        
        # Объединяем по дате
        combined = pd.merge(ga_data, ym_data, on='date', suffixes=('_ga', '_ym'))
        
        # Добавляем расчётные поля
        combined['total_sessions'] = combined['sessions'] + combined['visits']
        combined['avg_bounce_rate'] = (combined['bounce_rate_ga'] + combined['bounce_rate_ym']) / 2
        
        return combined
    
    def compare_systems(self, data):
        """Сравнение показателей между системами"""
        comparison = {
            'Метрика': ['Сессии', 'Пользователи', 'Показатель отказов'],
            'Google Analytics': [
                data['sessions'].sum(),
                data['users_ga'].sum(), 
                f"{data['bounce_rate_ga'].mean():.1f}%"
            ],
            'Яндекс.Метрика': [
                data['visits'].sum(),
                data['users_ym'].sum(),
                f"{data['bounce_rate_ym'].mean():.1f}%"
            ],
            'Расхождение': [
                f"{abs(data['sessions'].sum() - data['visits'].sum()) / data['sessions'].sum() * 100:.1f}%",
                f"{abs(data['users_ga'].sum() - data['users_ym'].sum()) / data['users_ga'].sum() * 100:.1f}%",
                f"{abs(data['bounce_rate_ga'].mean() - data['bounce_rate_ym'].mean()):.1f}п.п."
            ]
        }
        return pd.DataFrame(comparison)

Для мониторинга A/B тестов добавляем отдельный модуль:

class ABTestMonitor:
    def __init__(self, ga_parser):
        self.ga_parser = ga_parser
    
    def get_test_results(self, test_dimension, start_date, end_date):
        """Получение результатов A/B теста из GA"""
        body = {
            'reportRequests': [{
                'viewId': self.ga_parser.view_id,
                'dateRanges': [{'startDate': start_date, 'endDate': end_date}],
                'metrics': [
                    {'expression': 'ga:sessions'},
                    {'expression': 'ga:goal1Completions'},
                    {'expression': 'ga:goal1ConversionRate'}
                ],
                'dimensions': [{'name': test_dimension}]  # Например, ga:customDimension1
            }]
        }
        
        response = self.ga_parser.service.reports().batchGet(body=body).execute()
        return self._parse_test_data(response)
    
    def calculate_significance(self, control_group, test_group):
        """Расчёт статистической значимости A/B теста"""
        from scipy import stats
        
        # Преобразуем конверсии в бинарные данные
        control_conversions = int(control_group['conversions'])
        control_sessions = int(control_group['sessions'])
        
        test_conversions = int(test_group['conversions'])  
        test_sessions = int(test_group['sessions'])
        
        # Z-тест для двух пропорций
        z_stat, p_value = stats.proportions_ztest(
            [control_conversions, test_conversions],
            [control_sessions, test_sessions]
        )
        
        return {
            'z_statistic': z_stat,
            'p_value': p_value,
            'significant': p_value < 0.05,
            'confidence_level': (1 - p_value) * 100
        }

Результат — единый дашборд с данными из обеих систем аналитики плюс автоматический мониторинг результатов экспериментов.

Как масштабировать решение для больших проектов?

Когда парсер работает на одном-двух сайтах, всё просто. Но что делать, если у вас 20+ проектов, десятки счётчиков и сложная структура отчётности? Покажу, как мы решаем задачи масштабирования.

Во-первых, переходим на базу данных вместо файлов. Для небольших проектов хватает SQLite, для крупных лучше PostgreSQL:

import sqlalchemy as sa
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class AnalyticsData(Base):
    __tablename__ = 'analytics_data'
    
    id = Column(Integer, primary_key=True)
    project_id = Column(String(50), nullable=False)
    source = Column(String(20))  # 'ga' или 'ym'
    date = Column(DateTime, nullable=False)
    sessions = Column(Integer)
    users = Column(Integer)
    conversions = Column(Integer)
    conversion_rate = Column(Float)
    created_at = Column(DateTime, default=datetime.utcnow)

class AnalyticsDB:
    def __init__(self, db_url):
        self.engine = create_engine(db_url)
        Base.metadata.create_all(self.engine)
        Session = sessionmaker(bind=self.engine)
        self.session = Session()
    
    def save_analytics_data(self, project_id, source, data_df):
        """Сохранение данных аналитики в БД"""
        for _, row in data_df.iterrows():
            record = AnalyticsData(
                project_id=project_id,
                source=source,
                date=row['date'],
                sessions=row.get('sessions', 0),
                users=row.get('users', 0),
                conversions=row.get('conversions', 0),
                conversion_rate=row.get('conversion_rate', 0)
            )
            self.session.add(record)
        
        self.session.commit()
    
    def get_project_data(self, project_id, start_date, end_date):
        """Получение данных проекта за период"""
        query = self.session.query(AnalyticsData).filter(
            AnalyticsData.project_id == project_id,
            AnalyticsData.date.between(start_date, end_date)
        )
        return pd.read_sql(query.statement, self.engine)

Создаём конфигурационную систему для управления множественными проектами:

import yaml

class ProjectConfig:
    def __init__(self, config_path='config/projects.yaml'):
        with open(config_path, 'r', encoding='utf-8') as file:
            self.config = yaml.safe_load(file)
    
    def get_project_settings(self, project_id):
        """Получение настроек конкретного проекта"""
        return self.config['projects'].get(project_id, {})
    
    def get_all_projects(self):
        """Список всех активных проектов"""
        return [pid for pid, settings in self.config['projects'].items() 
                if settings.get('active', True)]

# config/projects.yaml
"""
projects:
  ecommerce_1:
    name: "Интернет-магазин №1"
    ga_view_id: "123456789"
    ym_counter_id: "987654321"
    goals:
      - purchase
      - add_to_cart
    active: true
    
  b2b_portal:
    name: "B2B портал"
    ga_view_id: "111222333"
    ym_counter_id: "444555666"
    goals:
      - lead_form
      - demo_request
    active: true
"""

Многопоточная обработка для ускорения сбора данных:

import concurrent.futures
import threading
from queue import Queue

class MultiProjectParser:
    def __init__(self, config, db):
        self.config = config
        self.db = db
        self.results_queue = Queue()
    
    def parse_all_projects(self, start_date, end_date, max_workers=5):
        """Параллельный парсинг всех проектов"""
        projects = self.config.get_all_projects()
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Запускаем парсинг для каждого проекта
            future_to_project = {
                executor.submit(self._parse_project, project_id, start_date, end_date): project_id 
                for project_id in projects
            }
            
            # Собираем результаты
            for future in concurrent.futures.as_completed(future_to_project):
                project_id = future_to_project[future]
                try:
                    result = future.result()
                    self.results_queue.put(('success', project_id, result))
                except Exception as exc:
                    self.results_queue.put(('error', project_id, str(exc)))
        
        return self._collect_results()
    
    def _parse_project(self, project_id, start_date, end_date):
        """Парсинг одного проекта"""
        settings = self.config.get_project_settings(project_id)
        
        # GA данные
        ga_parser = GAParser('credentials/service_account.json', settings['ga_view_id'])
        ga_data = ga_parser.get_kpi_report(start_date, end_date)
        self.db.save_analytics_data(project_id, 'ga', ga_data)
        
        # YM данные (если настроены)
        if settings.get('ym_counter_id'):
            ym_parser = YandexMetrikaParser(YM_TOKEN, settings['ym_counter_id'])
            ym_data = ym_parser.get_basic_metrics(start_date, end_date)
            self.db.save_analytics_data(project_id, 'ym', ym_data)
        
        return {
            'project_id': project_id,
            'ga_records': len(ga_data),
            'ym_records': len(ym_data) if settings.get('ym_counter_id') else 0
        }

Система алертов для мониторинга аномалий:

class AlertSystem:
    def __init__(self, db, config):
        self.db = db
        self.config = config
    
    def check_anomalies(self):
        """Проверка аномалий во всех проектах"""
        alerts = []
        
        for project_id in self.config.get_all_projects():
            # Данные за последние 7 и 14 дней
            current_week = self.db.get_project_data(
                project_id, 
                datetime.now() - timedelta(days=7),
                datetime.now()
            )
            prev_week = self.db.get_project_data(
                project_id,
                datetime.now() - timedelta(days=14), 
                datetime.now() - timedelta(days=7)
            )
            
            # Проверка падения трафика
            current_sessions = current_week['sessions'].sum()
            prev_sessions = prev_week['sessions'].sum()
            
            if current_sessions < prev_sessions * 0.7:  # Падение больше 30%
                alerts.append({
                    'type': 'traffic_drop',
                    'project': project_id,
                    'current': current_sessions,
                    'previous': prev_sessions,
                    'change': (current_sessions - prev_sessions) / prev_sessions * 100
                })
            
            # Проверка конверсии
            current_cr = current_week['conversion_rate'].mean()
            prev_cr = prev_week['conversion_rate'].mean()
            
            if abs(current_cr - prev_cr) > 0.5:  # Изменение больше 0.5%
                alerts.append({
                    'type': 'conversion_change',
                    'project': project_id,
                    'current_cr': current_cr,
                    'previous_cr': prev_cr,
                    'change': current_cr - prev_cr
                })
        
        if alerts:
            self._send_alerts(alerts)
        
        return alerts

Результат масштабирования — система, которая обрабатывает 50+ проектов за 10-15 минут, автоматически выявляет аномалии и отправляет алерты в Telegram или Slack.

Это часть серии материалов по теме «Скрипты и парсеры». Основная статья серии: Node.js парсер отзывов: как собирать данные с 20 площадок.

Частые вопросы

В: Сколько времени займёт настройка парсера Google Analytics?

О: Базовый парсер настраивается за 2-3 часа. Включает подключение API, написание скрипта для основных метрик и простую визуализацию. Полноценный дашборд с автоматизацией и интеграцией Яндекс.Метрики — 1-2 дня.

В: Можно ли парсить данные из Google Analytics без программирования?

О: Для разовых выгрузок есть Google Sheets Add-on и готовые инструменты типа Supermetrics. Но для автоматизации и сложной аналитики Python-скрипт эффективнее. Он даёт полный контроль над данными и логикой обработки.

В: Какие лимиты у Google Analytics API?

О: Стандартные лимиты: 100 запросов в 100 секунд на пользователя, до 10 concurrent запросов. Для большинства задач хватает. Если нужно больше — можно запросить увеличение квоты в Google Cloud Console.

В: Как часто можно обновлять данные в парсере?

О: Данные в Google Analytics обновляются с задержкой 24-48 часов. Реалтайм данные доступны через отдельный API, но с ограниченным набором метрик. Оптимально настроить обновление 1-2 раза в день.

В: Что делать, если парсер перестал работать после обновления GA4?

О: В GA4 изменилась структура API — теперь это Google Analytics Data API v1 вместо Reporting API v4. Нужно переписать запросы под новый формат метрик и dimensions. У нас есть готовые миграционные скрипты для клиентов.

В: Можно ли объединить данные из нескольких счётчиков Analytics?

О: Да, парсер может работать с множественными view_id в цикле. Данные объединяются на уровне pandas DataFrame по дате или другим измерениям. Полезно для сетей сайтов или мультибрендинга.

В: Как обеспечить безопасность API-ключей в парсере?

О: Используйте переменные окружения или отдельные файлы credentials, которые не попадают в Git. Service Account ключи должны иметь минимальные права — только чтение Analytics. Регулярно ротируйте ключи (раз в полгода).

Нужна помощь с этим? Обсудить проект с DS495 →