Как создать ротатор прокси для трекера позиций с резидентными прокси IPRoyal и Python

Обзор

Трекеры позиций делают одно и то же снова и снова: отправляют запрос к поисковой системе из определённой локации и записывают, где появляется домен. Без прокси эта схема быстро ломается. Поисковые системы локализуют результаты по IP запрашивающего и ограничивают или блокируют IP-адреса, которые обращаются к ним массово.

В этом руководстве показано, как запустить трекер позиций на Python через резидентные прокси IPRoyal по SOCKS5: собрать URL прокси, сохранять липкие сессии, когда нужна пагинация, ротировать, когда она не нужна, и обрабатывать сбои, которые вы действительно встретите в продакшене.

IPRoyal предлагает резидентные, ISP- и датацентровые прокси по HTTP, SOCKS5 и туннелированию в стиле Shadowsocks, заявляет пул из более чем 32 млн IP и тарифицирует резидентный трафик по модели pay-as-you-go, а для некоторых продуктов доступны фиксированные месячные планы. Хост, порт, имя пользователя и пароль шлюза находятся в вашей панели IPRoyal.

Какой тип прокси подходит для трекинга позиций?

Ротация против липких сессий

• Ротируйте каждый запрос, когда вам нужна только одна страница результатов на ключевое слово.
• Удерживайте липкую сессию (от нескольких минут примерно до 30 минут), когда вы переходите по страницам, раскрываете блоки связанных вопросов или вам нужен один и тот же исходящий IP для последовательности запросов.
• Нацеливайтесь на страну, соответствующую локали, по которой вы отчитываетесь. Проверка позиций для de-DE через исходящий IP в США — это не проверка немецких позиций.

Таргетинг по сессии и стране управляется через имя пользователя прокси или поддомен шлюза. Скопируйте точный шаблон из вашей панели IPRoyal, а не угадывайте; в коде ниже используются широко распространённые плейсхолдеры -session- и -country-.

Предварительные требования

• Python 3.9 или новее
• Дополнение SOCKS для Requests: pip install "requests[socks]"
• Хост, порт, имя пользователя и пароль шлюза IPRoyal из вашей панели
• Файл ключевых слов, по одной строке на ключевое слово: запрос, код страны и локаль

Шаги

1. Экспортируйте учётные данные, чтобы они никогда не хранились в исходном файле.

export IPROYAL_HOST="gateway-host-from-dashboard"
export IPROYAL_PORT="gateway-port-from-dashboard"
export IPROYAL_USER="your-username"
export IPROYAL_PASS="your-password"

В Windows PowerShell вместо этого используйте $env:IPROYAL_HOST = "...".

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

import os
import random
import string
import time

import requests

PROXY_HOST = os.environ['IPROYAL_HOST']
PROXY_PORT = os.environ['IPROYAL_PORT']
PROXY_USER = os.environ['IPROYAL_USER']
PROXY_PASS = os.environ['IPROYAL_PASS']

LOCALE_HEADERS = {
    'en-US': {'Accept-Language': 'en-US,en;q=0.9'},
    'de-DE': {'Accept-Language': 'de-DE,de;q=0.9'},
    'ja-JP': {'Accept-Language': 'ja-JP,ja;q=0.9'},
}


def new_session_id() -> str:
    """Короткий уникальный токен для закрепления липкой сессии."""
    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))


def proxy_url(country: str, session_id: str | None = None) -> str:
    user = PROXY_USER
    if session_id:
        user = f'{user}-session-{session_id}'
    if country:
        user = f'{user}-country-{country.lower()}'
    return f'socks5h://{user}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}'

3. Оберните запрос в повторные попытки с экспоненциальной задержкой. Одна неудачная проверка не должна прерывать прогон на 2000 ключевых слов.

def fetch_serp(query: str, country: str, locale: str,
               session_id: str | None = None, attempts: int = 3) -> str | None:
    proxy = proxy_url(country, session_id)
    proxies = {'http': proxy, 'https': proxy}
    headers = {
        'User-Agent': 'Mozilla/5.0 (compatible; RankTracker/1.0)',
        **LOCALE_HEADERS.get(locale, {}),
    }

    for attempt in range(1, attempts + 1):
        try:
            response = requests.get(
                'https://your-permitted-search-endpoint/search',
                params={'q': query, 'hl': locale},
                proxies=proxies,
                headers=headers,
                timeout=30,
            )
            response.raise_for_status()
            return response.text
        except requests.RequestException as exc:
            print(f'попытка {attempt} не удалась для {query!r}: {exc}')
            time.sleep(2 ** attempt)

    return None

Направьте URL на поисковый интерфейс, который вам разрешено запрашивать, например официальный API или ваш собственный разрешённый источник данных. Скрапинг поисковой системы может нарушать её условия обслуживания, поэтому убедитесь в правомерности, прежде чем масштабировать процесс.

4. Назначайте одну липкую сессию на ключевое слово, чтобы пагинация и последующие запросы использовали один и тот же исходящий IP.

def check_keyword(row: dict) -> dict:
    session_id = new_session_id()
    html = fetch_serp(row['query'], row['country'], row['locale'], session_id=session_id)
    return {
        'query': row['query'],
        'country': row['country'],
        'position': rank_position(html, row['domain']) if html else None,
        'status': 'ok' if html else 'failed',
    }

5. Извлекайте позиции с помощью парсера, который вы уже используете.

def rank_position(html: str | None, domain: str) -> int | None:
    if not html:
        return None
    # Разберите с помощью lxml, selectolax или BeautifulSoup, сопоставьте URL результатов
    # с `domain` и верните индекс первого совпадения, начиная с 1.
    ...

6. Ограничивайте параллелизм. Резидентные пулы большие, но не безграничные, а одновременная долбёжка эндпоинта с множества IP — самый быстрый путь к блокировке.

from concurrent.futures import ThreadPoolExecutor


def run_batch(keywords: list[dict], workers: int = 8) -> list[dict]:
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(check_keyword, keywords))

• Начните с 5–8 воркеров и увеличивайте число, только пока доля успешных запросов остаётся выше примерно 95%.
• Добавьте небольшую случайную задержку между повторными попытками, чтобы они не приходили синхронно.

7. Сохраняйте сырой HTML на короткий период, чтобы можно было перепарсить историю без повторного скрапинга.

import json
from pathlib import Path


def save_result(result: dict) -> None:
    out = Path('rank-results')
    out.mkdir(exist_ok=True)
    name = result['query'][:40].replace('/', '_')
    with (out / f'{name}.json').open('w', encoding='utf-8') as handle:
        json.dump(result, handle, ensure_ascii=False)

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

# Пример записи в cron: запускать трекер в 06:00 и 18:00 по местному времени
0 6,18 * * * cd /srv/rank-tracker && /usr/bin/python3 track.py >> logs/rank.log 2>&1

Устранение неполадок

• 407 Proxy Authentication Required: неверно имя пользователя или пароль, либо специальный символ в пароле не закодирован. Процентно закодируйте @, :, / и #, прежде чем вставлять пароль в URL.
• Missing dependencies for SOCKS support: Requests требует дополнение SOCKS. Выполните pip install "requests[socks]" в том же виртуальном окружении, что и ваш скрипт.
• Результаты показывают не ту страну: таргетинг по стране является частью имени пользователя прокси. Выведите proxy_url(...) для одного ключевого слова и убедитесь, что ожидаемый код страны присутствует, прежде чем винить поисковую систему.
• Каждый запрос использует один и тот же IP: вы повторно используете ID сессии. Опустите компонент сессии, чтобы провайдер мог ротировать.
• Внезапно блокируется целая страна: снизьте параллелизм для этой страны, ротируйте агрессивнее и проверьте, не начал ли эндпоинт возвращать страницы согласия или капчи вместо результатов.
• Медленнее, чем ожидалось: резидентная маршрутизация добавляет задержку. Увеличьте тайм-аут до 30–45 секунд для резидентных или ISP-прокси и оставьте датацентровые прокси для задач, не связанных с выдачей.

Итог

Успех трекинга позиций зависит от гигиены прокси. Используйте резидентные или ISP-прокси IPRoyal со схемой socks5h для удалённого DNS, закрепляйте сессию за ключевым словом, когда нужна непрерывность, ротируйте, когда она не нужна, и ограничивайте параллелизм, чтобы пул продолжал работать. С повторными попытками, хранением сырых ответов и логом использованного прокси вы сможете отличить реальное изменение позиций от артефакта прокси.