概述

排名跟踪器反复做一件事：从特定位置查询搜索引擎，并记录某个域名出现的位置。没有代理，这种模式很快就会失效。搜索引擎会根据请求者的 IP 本地化结果，并且会对大量查询它们的 IP 进行限流或封禁。

本教程展示如何通过 SOCKS5 使用 IPRoyal 住宅代理运行 Python 排名跟踪器：构建代理 URL，在需要分页时保持会话粘性，不需要时进行轮换，并处理你在生产环境中实际会遇到的故障。

IPRoyal 通过 HTTP、SOCKS5 和 Shadowsocks 风格隧道提供住宅、ISP 和数据中心代理，宣称拥有 3200 万+ IP 池，并按即用即付方式对住宅流量计费，部分产品还提供固定月付套餐。网关主机、端口、用户名和密码位于你的 IPRoyal 仪表板中。

哪种代理类型适合排名跟踪？

代理类型	最适合排名跟踪的场景	权衡
住宅（轮换）	国家和城市级 SERP；3200 万+ 的大型 IP 池分散请求量	响应较慢；每次请求间出口 IP 会变化
ISP（静态住宅）	从同一地址进行可重复检查	位置覆盖范围比住宅代理小
数据中心	轻量、高流量且不是 SERP 抓取的工作	快速且便宜，但容易被搜索引擎指纹识别

轮换与会话粘性

每个关键词只需要一个结果页时，每次请求都轮换。
当你跟踪分页、展开相关问题块，或需要在一系列请求中使用同一出口 IP 时，保持粘性会话（几分钟到大约 30 分钟）。
定向到与你报告的语言环境相匹配的国家/地区。通过美国出口 IP 进行的 de-DE 排名检查不是德国排名检查。

会话和国家/地区定向通过代理用户名或网关子域控制。请从你的 IPRoyal 仪表板复制确切模式，而不是猜测；下面的代码使用常见的 -session- 和 -country- 占位符。

先决条件

Python 3.9 或更高版本
Requests 的 SOCKS 扩展：pip install "requests[socks]"
来自你的仪表板的 IPRoyal 网关主机、端口、用户名和密码
关键词文件，每个关键词一行：查询、国家/地区代码和语言环境

步骤

导出你的凭据，使它们永远不会留在源文件中。

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 = "..."。

为每个请求构建代理 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:
    """Short, unique token used to pin a sticky session."""
    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}'

用指数退避将请求包装在重试中。一次失败的检查不应中止 2,000 个关键词的运行。

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 {attempt} failed for {query!r}: {exc}')
            time.sleep(2 ** attempt)

    return None

将 URL 指向你被允许查询的搜索接口，例如官方 API 或你自己的获准数据源。抓取搜索引擎可能违反其服务条款，因此在扩大工作流规模前请先确认。

为每个关键词分配一个粘性会话，以便分页和后续请求复用同一出口 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',
    }

使用你已有的解析器提取排名位置。

def rank_position(html: str | None, domain: str) -> int | None:
    if not html:
        return None
    # Parse with lxml, selectolax, or BeautifulSoup, match result URLs
    # against `domain`, and return the 1-based index of the first match.
    ...

限制并发。住宅代理池很大但并非无限，同时从许多 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 个 worker 开始，仅当成功率保持在大约 95% 以上时才提高数量。
在重试之间添加少量随机抖动，使重试不会同步到达。

将原始 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)

安排运行计划，并记录每个结果由哪个代理配置产生。当排名变化时，你需要知道是 SERP 变了还是代理变了。

# Example cron entry: run the tracker at 06:00 and 18:00 local time
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。省略会话组件，让提供商进行轮换。
整个国家/地区突然被封锁：降低该国家/地区的并发数，更积极地轮换，并检查该端点是否开始返回同意页面或验证码页面而不是结果。
比预期慢：住宅路由会增加延迟。对于住宅或 ISP 代理，将超时提高到 30-45 秒，并将数据中心代理保留给非 SERP 工作。

总结

排名跟踪的成败在于代理卫生。使用 IPRoyal 住宅或 ISP 代理，并采用 socks5h 方案实现远程 DNS；当你需要连续性时，为每个关键词固定一个会话；不需要时则轮换；并限制并发，使代理池持续可用。借助重试、原始响应存储和所用代理的日志，你可以区分真正的排名变化和代理造成的伪影。