如何在 Linux 上为 Python 价格监控设置 Smartproxy 数据中心代理

概述

Smartproxy 是一个对初学者友好的代理网络，提供可通过 HTTP 和 SOCKS5 访问的住宅和数据中心代理池，并按固定月费模式计费。本教程将其数据中心代理池用于 Linux 上一个具体且常见的任务：用 Python 进行定时价格监控。

价格监控非常适合数据中心 IP。请求速度快，目标是公开的产品页面，且工作负载以读取为主。如果某个特定零售商开始屏蔽数据中心 IP 段，同一个脚本可以改用住宅代理池运行——只需更改端点和凭据。

到本教程结束时，你将拥有：

• 存储在源代码之外的凭据
• 已从命令行验证的代理请求
• 一个通过代理抓取产品页面并带有重试和退避的 Python 脚本
• 将价格连同时间戳追加到 CSV 文件
• 一条按计划运行任务的 cron 条目

你不需要 GUI 来完成任何这些操作。Smartproxy 也提供浏览器扩展，但无头 Linux 任务只需要从仪表板获取的主机、端口、用户名和密码。

先决条件

• 一台 Linux 主机——以下命令适用于 Debian/Ubuntu 和 Fedora/RHEL 系列
• Python 3.9 或更新版本
• 一个具有数据中心代理访问权限的 Smartproxy 账户
• 在 Smartproxy 仪表板中列出的代理主机、端口、用户名和密码

数据中心还是住宅？

两种代理池在代码中的配置方式完全相同，因此决策取决于目标网站，而不是工具。

经验法则：

• 从数据中心开始，因为这是更便宜、更快速验证解析器是否工作的方式。
• 将单个顽固目标迁移到住宅代理，而不是迁移整个任务。
• 保持代理方案和身份验证在代理池之间完全相同，这样切换只是配置更改，而不是重写。

步骤

步骤 1 — 导出凭据并验证端点

将仪表板中的四个值设置为当前 shell 会话的环境变量。用你自己的值替换占位符。

export SMART_HOST="<host-from-dashboard>"
export SMART_PORT="<port-from-dashboard>"
export SMART_USER="<username>"
export SMART_PASS="<password>"

通过代理运行单个请求。使用 -U 传递凭据可避免密码中特殊字符引起的问题。

curl -sS --max-time 20 -x "http://${SMART_HOST}:${SMART_PORT}" -U "${SMART_USER}:${SMART_PASS}" https://api.ipify.org

响应应该是一个不是你服务器自身 IP 的 IP 地址。如果你看到的是 407 错误，请跳到末尾的故障排除表。

步骤 2 — 创建项目并将密钥移入文件

1. 创建一个目录和一个虚拟环境。
2. 安装 requests 以及 python-dotenv，后者会在运行时将 .env 文件加载到环境中。

mkdir -p ~/price-monitor && cd ~/price-monitor
python3 -m venv .venv && source .venv/bin/activate
pip install requests python-dotenv

创建密钥文件并锁定其权限。此文件绝不应被提交。

# ~/price-monitor/.env
SMART_HOST=<host-from-dashboard>
SMART_PORT=<port-from-dashboard>
SMART_USER=<username>
SMART_PASS=<password>

chmod 600 .env
cat > .gitignore <<'EOF'
.env
.venv/
EOF

步骤 3 — 在 Python 中构建代理 URL

构建一次 URL 并重复使用。使用 quote() 对用户名和密码进行引用编码，可防止 @、: 或 # 等字符破坏 URL 解析。

# ~/price-monitor/proxy_config.py
import os
from urllib.parse import quote
from dotenv import load_dotenv

load_dotenv()

def proxy_url(scheme='http'):
    user = quote(os.environ['SMART_USER'], safe='')
    password = quote(os.environ['SMART_PASS'], safe='')
    host = os.environ['SMART_HOST']
    port = os.environ['SMART_PORT']
    return f'{scheme}://{user}:{password}@{host}:{port}'

PROXIES = {'http': proxy_url(), 'https': proxy_url()}

如果你的端点对 HTTPS 流量使用不同的端口，请分别设置两个键，而不要假设端口是共享的。

步骤 4 — 通过代理抓取页面并带有重试

将请求包装在一个小型重试循环中。指数退避可防止你不断冲击正在对你进行速率限制的目标。

# ~/price-monitor/fetch_price.py
import time
import requests
from proxy_config import PROXIES

HEADERS = {
    'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 '
                  '(KHTML, like Gecko) Chrome/124.0 Safari/537.36',
    'Accept-Language': 'en-US,en;q=0.9',
}

def fetch(url, retries=3, timeout=25):
    last_error = None
    for attempt in range(1, retries + 1):
        try:
            response = requests.get(url, headers=HEADERS, proxies=PROXIES, timeout=timeout)
            if response.status_code == 200:
                return response.text
            last_error = 'HTTP %s' % response.status_code
        except requests.RequestException as exc:
            last_error = str(exc)
        time.sleep(2 ** attempt)
    raise RuntimeError('Failed to fetch %s: %s' % (url, last_error))

if __name__ == '__main__':
    html = fetch('https://example.com/product/123')
    print(html[:500])

步骤 5 — 提取价格并将其追加到 CSV

诸如 BeautifulSoup 或 selectolax 之类的 HTML 解析器可提供更具弹性的提取，但在你投入选择器之前，正则表达式足以验证管道。

# ~/price-monitor/parse_price.py
import csv
import datetime
import re
import sys
from fetch_price import fetch

def parse_price(html):
    # Replace this pattern with a selector that matches the target site's markup.
    match = re.search(r'data-price=([0-9.]+)', html)
    return float(match.group(1)) if match else None

def main(url):
    price = parse_price(fetch(url))
    with open('prices.csv', 'a', newline='', encoding='utf-8') as handle:
        writer = csv.writer(handle)
        writer.writerow([datetime.datetime.now(datetime.UTC).isoformat(), url, price])

if __name__ == '__main__':
    main(sys.argv[1] if len(sys.argv) > 1 else 'https://example.com/product/123')

步骤 6 — 决定轮换应如何为你的端点工作

轮换行为取决于分配给你的端点，因此在做任何假设之前请检查你的仪表板：

• 如果端口在每次请求时轮换出口 IP，只需再次调用 fetch() 即可——每个新连接都可以从不同的 IP 离开。
• 如果端口保持粘性会话，请循环使用仪表板中列出的端口，并为每个工作线程保留一个端口。
• 仅在需要粘性会话时才重复使用同一个 requests.Session()；每次请求使用新会话是监控更安全的默认做法。

要确认轮换，请连续多次请求 IP 端点。

def check_ip():
    return fetch('https://api.ipify.org').strip()

if __name__ == '__main__':
    for _ in range(5):
        print(check_ip())

步骤 7 — 添加适度的并发

线程池可缩短一长串 URL 的处理时间，但并发也是最快被封锁的方式。从小规模开始，仅在错误率保持平稳时才提高工作线程数。

from concurrent.futures import ThreadPoolExecutor, as_completed
from fetch_price import fetch

urls = [f'https://example.com/product/{sku}' for sku in range(1, 6)]

with ThreadPoolExecutor(max_workers=5) as pool:
    futures = {pool.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            print(url, len(future.result()))
        except Exception as exc:
            print(url, 'failed:', exc)

步骤 8 — 使用 cron 安排任务

编辑你的 crontab 并添加一行。cd 很重要，因为脚本依赖其 .env 文件位于工作目录中。

crontab -e

0 */6 * * * cd /home/<user>/price-monitor && .venv/bin/python parse_price.py https://example.com/product/123 >> run.log 2>&1

使用虚拟环境 Python 的绝对路径。在 cron 下直接使用 python 通常解析为系统解释器，并且看不到你安装的包。

使用 SOCKS5 代替 HTTP

Smartproxy 在 HTTP 之外还支持 SOCKS5，切换只需更改一行。请在仪表板中确认 SOCKS5 端口，因为它可能与 HTTP 端口不同。

pip install 'requests[socks]'

PROXIES = {'http': proxy_url('socks5'), 'https': proxy_url('socks5')}

当你还希望 DNS 解析通过代理进行时，或者管道中的另一个工具只支持 SOCKS5 时，SOCKS5 很有用。

故障排除

总结

• Smartproxy 的数据中心代理池是 Linux 上进行高流量、只读价格监控的实用默认选择。
• 将主机、端口、用户名和密码保存在具有 600 权限的 .env 文件中，并使用 python-dotenv 加载它们。
• 在将凭据放入代理 URL 之前对其进行引用编码，否则特殊字符会破坏解析。
• 始终设置请求超时，使用退避重试，并保持低并发，直到错误率证明可以安全增加。
• 检查你的仪表板，了解你的端点是按请求轮换还是保持粘性会话，然后围绕这一点设计运行。
• 将单个被封锁的目标切换到住宅代理池，而不是重建任务；仅在出于 DNS 或工具原因需要时使用 SOCKS5。