Интеграция Google Search Console API для мониторинга SEO сайта

Интеграция Google Search Console API для мониторинга SEO сайта

Google Search Console даёт данные, которых нет больше нигде: реальные поисковые запросы, по которым показывается сайт, CTR по позициям, покрытие индексации, ошибки Core Web Vitals из полевых данных. Ручная проверка раз в неделю — это потеря сигналов. API позволяет вытягивать эти данные автоматически, строить собственные дашборды и настраивать алерты на деградацию.

Настройка доступа к API

Авторизация через Google OAuth 2.0. Для серверного мониторинга используется Service Account — без интерактивного входа:

1. Создать проект в Google Cloud Console
2. Включить Google Search Console API
3. Создать Service Account, скачать JSON-ключ
4. В GSC добавить email сервисного аккаунта как пользователя ресурса (Settings → Users and permissions)

from google.oauth2 import service_account
from googleapiclient.discovery import build

SCOPES = ['https://www.googleapis.com/auth/webmasters.readonly']
SERVICE_ACCOUNT_FILE = 'gsc-service-account.json'

credentials = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_FILE,
    scopes=SCOPES
)

service = build('searchconsole', 'v1', credentials=credentials)

Получение данных поиска

Основной метод — searchanalytics.query. Параметры: диапазон дат, измерения (query, page, country, device, date), фильтры, лимит строк (максимум 25 000 на запрос).

def fetch_search_data(
    service,
    site_url: str,
    start_date: str,
    end_date: str,
    dimensions: list[str] = ['query', 'page'],
    row_limit: int = 5000
) -> list[dict]:
    request = {
        'startDate': start_date,
        'endDate': end_date,
        'dimensions': dimensions,
        'rowLimit': row_limit,
        'startRow': 0,
    }

    response = service.searchanalytics().query(
        siteUrl=site_url,
        body=request
    ).execute()

    rows = response.get('rows', [])
    results = []

    for row in rows:
        entry = {dim: row['keys'][i] for i, dim in enumerate(dimensions)}
        entry.update({
            'clicks': row.get('clicks', 0),
            'impressions': row.get('impressions', 0),
            'ctr': row.get('ctr', 0),
            'position': row.get('position', 0),
        })
        results.append(entry)

    return results

Для выгрузки более 25 000 строк — пагинация через startRow:

def fetch_all_rows(service, site_url, body):
    all_rows = []
    start_row = 0
    while True:
        body['startRow'] = start_row
        response = service.searchanalytics().query(
            siteUrl=site_url, body=body
        ).execute()
        rows = response.get('rows', [])
        if not rows:
            break
        all_rows.extend(rows)
        start_row += len(rows)
        if len(rows) < body.get('rowLimit', 1000):
            break
    return all_rows

Мониторинг позиций по ключевым страницам

Типичная задача — отслеживать позиции для определённых страниц по брендовым и небрендовым запросам. Фильтрация по странице:

def get_page_positions(service, site_url, page_url, days=28):
    from datetime import date, timedelta
    end_date = date.today().isoformat()
    start_date = (date.today() - timedelta(days=days)).isoformat()

    body = {
        'startDate': start_date,
        'endDate': end_date,
        'dimensions': ['query'],
        'dimensionFilterGroups': [{
            'filters': [{
                'dimension': 'page',
                'operator': 'equals',
                'expression': page_url,
            }]
        }],
        'rowLimit': 1000,
    }

    rows = service.searchanalytics().query(
        siteUrl=site_url, body=body
    ).execute().get('rows', [])

    return sorted(
        [{'query': r['keys'][0], 'position': r['position'], 'clicks': r['clicks']}
         for r in rows],
        key=lambda x: x['position']
    )

Проверка индексации

API покрытия URL (sitemaps и urlInspection) позволяет проверять статус индексации отдельных страниц:

def inspect_url(service, site_url, inspect_url):
    response = service.urlInspection().index().inspect(
        body={
            'inspectionUrl': inspect_url,
            'siteUrl': site_url,
        }
    ).execute()

    result = response.get('inspectionResult', {})
    index_status = result.get('indexStatusResult', {})

    return {
        'coverageState': index_status.get('coverageState'),
        'robotsTxtState': index_status.get('robotsTxtState'),
        'indexingState': index_status.get('indexingState'),
        'lastCrawlTime': index_status.get('lastCrawlTime'),
        'pageFetchState': index_status.get('pageFetchState'),
        'googleCanonical': index_status.get('googleCanonical'),
        'userCanonical': index_status.get('userCanonical'),
    }

coverageState может быть: Submitted and indexed, Crawled - currently not indexed, Discovered - currently not indexed, Excluded by noindex tag и другие.

Алерты на деградацию

Полезная автоматизация — сравнивать текущую неделю с предыдущей и слать уведомление, если падение кликов превышает порог:

import smtplib
from email.mime.text import MIMEText

def check_traffic_drop(current: list, previous: list, threshold: float = 0.2) -> list:
    prev_map = {r['keys'][0]: r for r in previous}
    alerts = []
    for row in current:
        page = row['keys'][0]
        curr_clicks = row.get('clicks', 0)
        prev_clicks = prev_map.get(page, {}).get('clicks', 0)
        if prev_clicks > 50 and curr_clicks < prev_clicks * (1 - threshold):
            drop_pct = (1 - curr_clicks / prev_clicks) * 100
            alerts.append({
                'page': page,
                'prev_clicks': prev_clicks,
                'curr_clicks': curr_clicks,
                'drop_pct': round(drop_pct, 1),
            })
    return sorted(alerts, key=lambda x: -x['drop_pct'])

Хранение и визуализация

Собранные данные пишутся в PostgreSQL или BigQuery для долгосрочного хранения и трендов:

CREATE TABLE gsc_search_analytics (
    id SERIAL PRIMARY KEY,
    site_url TEXT NOT NULL,
    query TEXT,
    page TEXT,
    country TEXT,
    device TEXT,
    date DATE NOT NULL,
    clicks INTEGER DEFAULT 0,
    impressions INTEGER DEFAULT 0,
    ctr NUMERIC(6,4),
    position NUMERIC(8,2),
    collected_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_gsc_date_page ON gsc_search_analytics(date, page);
CREATE INDEX idx_gsc_query ON gsc_search_analytics(query);

Визуализация — Grafana с PostgreSQL-источником, Metabase, или собственный дашборд. GSC хранит данные только 16 месяцев — накопление в собственной БД позволяет строить долгосрочные тренды.

Квоты API

GSC API имеет ограничения: 1200 запросов в минуту на проект, 200 запросов на пользователя в 100 секунд. При ежедневном сборе данных за 28-дневное окно по нескольким измерениям это не проблема. При bulk-выгрузке исторических данных нужен экспоненциальный backoff:

import time
from googleapiclient.errors import HttpError

def execute_with_retry(request, max_retries=5):
    for attempt in range(max_retries):
        try:
            return request.execute()
        except HttpError as e:
            if e.resp.status in [429, 500, 503]:
                wait = 2 ** attempt
                time.sleep(wait)
            else:
                raise
    raise Exception('Max retries exceeded')

Сроки

Базовая интеграция (ежедневный сбор кликов/позиций в БД) — 2 рабочих дня. С алертами на деградацию, проверкой индексации, Grafana-дашбордом — 4–5 дней. Настройка под несколько сайтов/ресурсов GSC с общим хранилищем — 5–7 дней.