Настройка автоматического мониторинга позиций сайта в поиске

Настройка автоматического мониторинга позиций сайта в поиске

Позиции в поиске не статичны: апдейты алгоритмов Google/Яндекс, активность конкурентов, изменения на самом сайте — всё это влияет на ранжирование. Знать об изменениях через неделю-две — значит реагировать поздно. Автоматический мониторинг позиций даёт сигнал в течение 24 часов.

Источники данных

Существует три принципиально разных подхода к получению позиций:

Google Search Console API — бесплатно, данные за 28 дней, реальные позиции по всем запросам, но с задержкой 2–3 дня. Не позволяет проверить конкретную позицию «прямо сейчас». Хорошо для отслеживания общих трендов и исторического анализа.

Платные API (SE Ranking, Serpstat, DataForSEO, Serpwow) — проверяют позицию по конкретному ключевому слову для заданного региона прямо сейчас. Стоимость: от $0.001 до $0.05 за запрос в зависимости от сервиса и объёма.

Прямой парсинг SERP — технически возможен, но нарушает ToS Google/Яндекс, требует прокси-ротацию, ненадёжен. Не используется в продакшн-мониторинге.

Интеграция DataForSEO SERP API

DataForSEO — один из наиболее доступных по цене API для проверки позиций с поддержкой Google, Яндекс, Bing:

import requests
import json
from base64 import b64encode

class DataForSEOClient:
    BASE_URL = 'https://api.dataforseo.com/v3'

    def __init__(self, login: str, password: str):
        creds = b64encode(f'{login}:{password}'.encode()).decode()
        self.headers = {
            'Authorization': f'Basic {creds}',
            'Content-Type': 'application/json',
        }

    def check_positions(
        self,
        keyword: str,
        target_domain: str,
        location_code: int = 2840,  # USA; 2643 = UK, 2112 = Russia
        language_code: str = 'en',
        depth: int = 100,
    ) -> dict:
        payload = [{
            'keyword': keyword,
            'target': target_domain,
            'location_code': location_code,
            'language_code': language_code,
            'depth': depth,
        }]

        resp = requests.post(
            f'{self.BASE_URL}/serp/google/organic/live/advanced',
            headers=self.headers,
            data=json.dumps(payload),
            timeout=60,
        )
        resp.raise_for_status()
        return resp.json()

    def parse_position(self, response: dict, target_domain: str) -> dict | None:
        tasks = response.get('tasks', [])
        if not tasks:
            return None

        items = tasks[0].get('result', [{}])[0].get('items', [])
        for item in items:
            if item.get('type') == 'organic':
                domain = item.get('domain', '')
                if target_domain in domain:
                    return {
                        'position': item.get('rank_absolute'),
                        'url': item.get('url'),
                        'title': item.get('title'),
                        'featured_snippet': item.get('rank_absolute') == 0,
                    }
        return None  # не найден в топ-100

Яндекс через XML API

Для сайтов с российской аудиторией важен Яндекс. Яндекс предоставляет XML API для поиска — отдельная квота, требует регистрации:

import xml.etree.ElementTree as ET

def check_yandex_position(
    query: str,
    target_domain: str,
    user: str,
    key: str,
    region: int = 213,  # Москва
    depth: int = 100,
) -> int | None:
    url = 'https://yandex.ru/search/xml'
    params = {
        'user': user,
        'key': key,
        'query': query,
        'lr': region,
        'l10n': 'ru',
        'sortby': 'rlv',
        'filter': 'none',
        'groupby': f'attr=d.mode=deep.groups-on-page={depth}.docs-in-group=1',
    }

    resp = requests.get(url, params=params, timeout=30)
    root = ET.fromstring(resp.content)

    for i, doc in enumerate(root.findall('.//doc'), start=1):
        domain_el = doc.find('domain')
        if domain_el is not None and target_domain in domain_el.text:
            return i

    return None

Структура базы данных для хранения позиций

CREATE TABLE tracked_keywords (
    id SERIAL PRIMARY KEY,
    keyword TEXT NOT NULL,
    target_domain TEXT NOT NULL,
    search_engine VARCHAR(20) DEFAULT 'google',  -- google, yandex, bing
    location_code INTEGER,
    language_code VARCHAR(10),
    active BOOLEAN DEFAULT true,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE position_history (
    id SERIAL PRIMARY KEY,
    keyword_id INTEGER REFERENCES tracked_keywords(id),
    position INTEGER,        -- NULL если не найден в топ-100
    url TEXT,
    checked_at DATE NOT NULL,
    UNIQUE(keyword_id, checked_at)
);

CREATE INDEX idx_positions_keyword_date
    ON position_history(keyword_id, checked_at DESC);

Ежедневный запуск мониторинга

import psycopg2
from datetime import date

def run_daily_check(db_conn, dfs_client: DataForSEOClient, target_domain: str):
    today = date.today().isoformat()

    with db_conn.cursor() as cur:
        cur.execute('''
            SELECT id, keyword, search_engine, location_code, language_code
            FROM tracked_keywords
            WHERE active = true
        ''')
        keywords = cur.fetchall()

    for kw_id, keyword, engine, loc_code, lang_code in keywords:
        try:
            response = dfs_client.check_positions(
                keyword=keyword,
                target_domain=target_domain,
                location_code=loc_code or 2840,
                language_code=lang_code or 'en',
            )
            result = dfs_client.parse_position(response, target_domain)
            position = result['position'] if result else None
            url = result['url'] if result else None

            with db_conn.cursor() as cur:
                cur.execute('''
                    INSERT INTO position_history (keyword_id, position, url, checked_at)
                    VALUES (%s, %s, %s, %s)
                    ON CONFLICT (keyword_id, checked_at) DO UPDATE
                    SET position = EXCLUDED.position, url = EXCLUDED.url
                ''', (kw_id, position, url, today))
            db_conn.commit()

        except Exception as e:
            print(f'Error checking {keyword}: {e}')

Алерты на изменения позиций

def detect_significant_changes(db_conn, threshold: int = 5) -> list[dict]:
    with db_conn.cursor() as cur:
        cur.execute('''
            WITH ranked AS (
                SELECT
                    k.keyword,
                    p.position,
                    p.checked_at,
                    LAG(p.position) OVER (
                        PARTITION BY p.keyword_id ORDER BY p.checked_at
                    ) AS prev_position
                FROM position_history p
                JOIN tracked_keywords k ON k.id = p.keyword_id
                WHERE p.checked_at >= CURRENT_DATE - INTERVAL '2 days'
            )
            SELECT keyword, prev_position, position,
                   (COALESCE(prev_position, 101) - COALESCE(position, 101)) AS change
            FROM ranked
            WHERE prev_position IS NOT NULL
              AND ABS(COALESCE(prev_position, 101) - COALESCE(position, 101)) >= %s
            ORDER BY ABS(change) DESC
        ''', (threshold,))
        return [
            {
                'keyword': row[0],
                'prev': row[1],
                'current': row[2],
                'change': row[3],
                'direction': 'up' if row[3] > 0 else 'down',
            }
            for row in cur.fetchall()
        ]

Нотификации в Telegram

import httpx

def send_telegram_alert(bot_token: str, chat_id: str, changes: list[dict]):
    if not changes:
        return

    lines = ['*Изменения позиций за сутки:*\n']
    for ch in changes[:20]:
        arrow = '↑' if ch['direction'] == 'up' else '↓'
        prev = ch['prev'] or '100+'
        curr = ch['current'] or '100+'
        lines.append(f"{arrow} `{ch['keyword']}`: {prev} → {curr}")

    text = '\n'.join(lines)
    httpx.post(
        f'https://api.telegram.org/bot{bot_token}/sendMessage',
        json={'chat_id': chat_id, 'text': text, 'parse_mode': 'Markdown'},
    )

Расчёт объёма запросов и стоимости

Для 100 ключевых слов с ежедневной проверкой — 100 API-запросов в день, 3000 в месяц. По тарифам DataForSEO ($0.003–0.005 за Google SERP) это $9–15/месяц. При 500 словах — $45–75/месяц. Яндекс XML API бесплатный при небольшом объёме (квота зависит от параметров аккаунта).

Сроки

Настройка мониторинга с хранением в PostgreSQL и Telegram-алертами для одного домена — 2–3 рабочих дня. Добавление визуализации (Grafana/Metabase), поддержки нескольких сайтов, автоматического импорта ключевых слов из GSC — 4–6 дней.