Интеграция Google PageSpeed Insights API для мониторинга скорости сайта

Интеграция Google PageSpeed Insights API для мониторинга скорости сайта

PageSpeed Insights (PSI) возвращает два типа данных: лабораторные (Lighthouse, симулированное окружение) и полевые (Chrome UX Report, реальные пользователи). Оба важны, но по-разному. Лабораторные показывают состояние прямо сейчас — после деплоя, после добавления нового скрипта. Полевые — реальный опыт пользователей за последние 28 дней. Мониторинг нужен для обоих.

Получение ключа API и базовый запрос

PSI API бесплатный. Ключ создаётся в Google Cloud Console → APIs & Services → Credentials. Без ключа лимит — 400 запросов в день, с ключом — 25 000.

import requests
from typing import Literal

PSI_API_URL = 'https://www.googleapis.com/pagespeedonline/v5/runPagespeed'

def fetch_psi(
    url: str,
    api_key: str,
    strategy: Literal['mobile', 'desktop'] = 'mobile',
) -> dict:
    params = {
        'url': url,
        'key': api_key,
        'strategy': strategy,
        'category': ['performance', 'seo', 'accessibility', 'best-practices'],
    }
    resp = requests.get(PSI_API_URL, params=params, timeout=60)
    resp.raise_for_status()
    return resp.json()

Извлечение Core Web Vitals

Структура ответа PSI многоуровневая. Полевые данные (CrUX) находятся в loadingExperience, лабораторные — в lighthouseResult.audits.

def extract_field_data(psi_response: dict) -> dict:
    exp = psi_response.get('loadingExperience', {})
    metrics = exp.get('metrics', {})

    def metric(key):
        m = metrics.get(key, {})
        return {
            'percentile': m.get('percentile'),
            'category': m.get('category'),  # FAST / AVERAGE / SLOW
        }

    return {
        'overall_category': exp.get('overall_category'),
        'lcp': metric('LARGEST_CONTENTFUL_PAINT_MS'),
        'fid': metric('FIRST_INPUT_DELAY_MS'),
        'cls': metric('CUMULATIVE_LAYOUT_SHIFT_SCORE'),
        'fcp': metric('FIRST_CONTENTFUL_PAINT_MS'),
        'inp': metric('INTERACTION_TO_NEXT_PAINT'),
        'ttfb': metric('EXPERIMENTAL_TIME_TO_FIRST_BYTE'),
    }

def extract_lab_data(psi_response: dict) -> dict:
    audits = psi_response.get('lighthouseResult', {}).get('audits', {})
    categories = psi_response.get('lighthouseResult', {}).get('categories', {})

    def audit_val(key, field='numericValue'):
        return audits.get(key, {}).get(field)

    return {
        'performance_score': categories.get('performance', {}).get('score'),
        'lcp_ms': audit_val('largest-contentful-paint'),
        'fcp_ms': audit_val('first-contentful-paint'),
        'tbt_ms': audit_val('total-blocking-time'),
        'cls': audit_val('cumulative-layout-shift'),
        'speed_index': audit_val('speed-index'),
        'tti_ms': audit_val('interactive'),
        'server_response_time_ms': audit_val('server-response-time'),
    }

Сохранение результатов

CREATE TABLE psi_results (
    id SERIAL PRIMARY KEY,
    url TEXT NOT NULL,
    strategy VARCHAR(10) NOT NULL,
    measured_at TIMESTAMP DEFAULT NOW(),

    -- Полевые данные (CrUX)
    field_overall_category TEXT,
    field_lcp_ms INTEGER,
    field_lcp_category TEXT,
    field_cls NUMERIC(6,4),
    field_cls_category TEXT,
    field_inp_ms INTEGER,
    field_inp_category TEXT,
    field_fcp_ms INTEGER,
    field_ttfb_ms INTEGER,

    -- Лабораторные данные (Lighthouse)
    lab_performance_score NUMERIC(4,2),
    lab_lcp_ms INTEGER,
    lab_fcp_ms INTEGER,
    lab_tbt_ms INTEGER,
    lab_cls NUMERIC(6,4),
    lab_tti_ms INTEGER,
    lab_speed_index INTEGER,
    lab_server_response_ms INTEGER
);

CREATE INDEX idx_psi_url_time ON psi_results(url, measured_at DESC);

Мониторинг списка страниц

Для сайта с несколькими приоритетными страницами (главная, топ-10 посадочных, корзина, карточка товара) запуск по расписанию:

import time
from datetime import datetime

PAGES_TO_MONITOR = [
    'https://example.com/',
    'https://example.com/catalog/',
    'https://example.com/product/bestseller/',
    'https://example.com/checkout/',
]

def run_monitoring(api_key: str, db_conn):
    results = []
    for url in PAGES_TO_MONITOR:
        for strategy in ('mobile', 'desktop'):
            try:
                data = fetch_psi(url, api_key, strategy)
                field = extract_field_data(data)
                lab = extract_lab_data(data)
                results.append({
                    'url': url,
                    'strategy': strategy,
                    'measured_at': datetime.utcnow(),
                    'field': field,
                    'lab': lab,
                })
                time.sleep(2)  # избегаем rate limit
            except Exception as e:
                print(f'Error for {url} ({strategy}): {e}')
    return results

Алерты при деградации

Отслеживаем падение Lighthouse score и переход CWV из «хорошо» в «требует улучшений»:

THRESHOLDS = {
    'lab_performance_score': 0.7,   # ниже 70 — алерт
    'lab_lcp_ms': 4000,             # > 4s
    'lab_tbt_ms': 600,              # > 600ms
    'field_lcp_category': 'SLOW',   # полевые данные стали медленными
}

def check_alerts(current: dict, thresholds: dict) -> list[str]:
    alerts = []
    lab = current.get('lab', {})
    field = current.get('field', {})

    if lab.get('performance_score', 1) < thresholds['lab_performance_score']:
        score = round(lab['performance_score'] * 100)
        alerts.append(f"Performance score упал до {score} (порог: {int(thresholds['lab_performance_score']*100)})")

    if lab.get('lab_lcp_ms', 0) > thresholds['lab_lcp_ms']:
        alerts.append(f"LCP = {lab['lab_lcp_ms']}ms (порог: {thresholds['lab_lcp_ms']}ms)")

    if field.get('lcp', {}).get('category') == 'SLOW':
        alerts.append(f"Полевой LCP стал SLOW (реальные пользователи)")

    return alerts

Сравнение с предыдущим замером

def compare_results(current: dict, previous: dict) -> dict:
    changes = {}
    lab_keys = ['lab_performance_score', 'lab_lcp_ms', 'lab_tbt_ms', 'lab_cls']
    for key in lab_keys:
        curr_val = current.get(key)
        prev_val = previous.get(key)
        if curr_val is not None and prev_val is not None:
            delta = curr_val - prev_val
            changes[key] = {
                'previous': prev_val,
                'current': curr_val,
                'delta': round(delta, 4),
                'direction': 'worse' if (
                    (key == 'lab_performance_score' and delta < 0) or
                    (key != 'lab_performance_score' and delta > 0)
                ) else 'better',
            }
    return changes

Важные ограничения

PSI API запускает Lighthouse в облаке Google, а не у вас. Результаты варьируются между запусками на 5–15% — это нормально. Для надёжных показателей лучше запускать 3 измерения и брать медиану. Также PSI не поддерживает авторизованные страницы — для мониторинга личного кабинета, корзины после логина нужен локальный Lighthouse через Node.js.

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

Запуск через cron

# Crontab: каждый день в 6:00
0 6 * * * /usr/bin/python3 /opt/monitoring/psi_monitor.py >> /var/log/psi_monitor.log 2>&1

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

Сроки

Скрипт сбора + хранение в PostgreSQL + алерты на email/Telegram — 1–2 рабочих дня. С Grafana-дашбордом, сравнением версий до/после деплоя, интеграцией в CI/CD pipeline — 3–4 дня.