如何在Python中轮换代理IP地址

如何在Python中轮换代理IP地址

代理可以隐藏您的真实 IP 地址,但是当它被禁止时会发生什么?你需要一个新的IP。或者您可以维护它们的列表并使用 Python 为每个请求轮换代理。最后的选择是使用Smart Rotating Proxies,稍后会详细介绍。

现在,我们将专注于在 Python 中构建我们的自定义代理旋转器。我们将从常规代理列表开始,检查它们以标记工作的代理并提供简单的监控以从工作代理列表中删除失败的代理。本教程中提供的示例使用 Python,但这个想法适用于您用于抓取项目的任何语言。

如何轮换我的IP?

在构建用于 URL 和数据提取的爬虫时,防御系统阻止访问的最简单方法是禁止 IP。如果来自同一 IP 的大量请求在短时间内到达服务器,它们将标记该 IP 地址。

为避免这种情况,最简单的方法是使用不同的 IP 地址。但是你不能轻易改变它,对于服务器来说这几乎是不可能的。因此,为了轮换 IP,您需要通过代理服务器执行您的请求。这些将使您的原始请求保持不变,但目标服务器将看到他们的 IP,而不是您的。

但并非所有代理都是相同的。

静态和旋转代理有什么区别?

静态代理是那些使用相同出口 IP 的代理。这意味着您可以通过该代理更改您的 IP 地址,但最终 IP 将始终相同。

为了避免在每个请求中使用相同的 IP,静态代理通常会出现在列表中,因此您可以全部使用它们。例如,一段时间后更换它们或连续旋转它们。

另一方面,轮换代理会自动更改出口 IP 地址。某些轮换代理会在每次请求后切换 IP。其他人会在一段时间后更改它,例如 5 分钟。这种代理默认为您提供所需的 IP 轮换,但频率可能不够。

我们现在将了解如何使用静态代理在 Python 中轮换代理。我们走吧!

如何在 Python 中轮换代理?

要使代码正常工作,您需要安装 python3。有些系统已经预装了它。之后,通过运行安装所有必需的库pip install

pip install aiohttp

1.获取代理列表

您可能没有包含域+端口列表的代理提供商。别担心,我们会看看如何得到一个。

网上有几个免费代理列表。对于演示,获取其中一个并将其内容(仅 URL)保存在文本文件 ( rotating_proxies_list.txt) 中。或者使用下面的。他们不需要身份验证。

export-free-proxies

从站点导出和复制代理。如果你想自动化这个过程,该站点还提供了一个代理 API 来下载列表。

免费代理不可靠,下面的代理可能不适合您。它们通常是短暂的。对于生产抓取项目,我们建议使用数据中心或住宅代理。

167.71.230.124:8080 
192.155.107.211:1080 
77.238.79.111:8080 
167.71.5.83:3128 
195.189.123.213:3128 
8.210.83.33:80 
80.48.119.28:8080 
152.0.209.175:8080 
187.217.54.84:80 
169.57.1.85:8123

然后,我们将读取该文件并创建一个包含所有代理的数组。读取文件,去除空格,并拆分每一行。保存文件时要小心,因为我们不会对有效IP:port字符串执行任何健全性检查。我们会保持简单。

proxies_list = open("rotating_proxies_list.txt", "r").read().strip().split("n")

2.检查代理

假设我们想要大规模运行爬虫。该演示已简化,但想法是将代理及其“健康状态”存储在数据库等可靠介质中。我们将使用每次运行后消失的内存数据结构,但你明白了。那将是我们的代理池。

首先,让我们编写一个简单的函数来检查代理是否工作。为此,请调用ident.me,这是一个将返回带有 IP 的 HTML 响应的网页。这是一个适合我们用例的简单页面。我们将使用 Pythonrequests库,“一个简单而优雅的 HTTP 库”。

目前,它从代理列表中获取一个项目并调用提供的 URL。大多数代码都是样板,很快就会证明是有用的。有两种可能的结果:

  • 😊 如果一切顺利,它会打印响应的内容和状态代码(即 200),这可能是代理的 IP。
  • 😔 由于超时或其他原因打印错误。这通常意味着代理不可用或无法处理请求。使用免费代理时,其中许多会出现。
    import requests 
     
    proxies_list = open("rotating_proxies_list.txt", "r").read().strip().split("n") 
     
    def get(url, proxy): 
        try: 
            # Send proxy requests to the final URL 
            response = requests.get(url, proxies={'http': f"http://{proxy}"}, timeout=30) 
            print(response.status_code, response.text) 
        except Exception as e: 
            print(e) 
     
    def check_proxies(): 
        proxy = proxies_list.pop() 
        get("http://ident.me/", proxy) 
     
    check_proxies()

我们有意使用 HTTP 而不是 HTTPS,因为许多免费代理不支持 SSL。

将此内容保存在文件中,例如 rotate-proxies-python.py。然后从命令行运行它:

python rotate-proxies-python.py

输出应该是这样的:

200 167.71.230.124 ### status code and the proxy's IP

3. 添加更多检查以验证结果

异常意味着请求失败,但我们还应该检查其他选项,例如状态代码。我们将只考虑有效的特定代码并将其余代码标记为错误。该列表并不详尽,请根据您的需要进行调整。例如,您可能认为 404“未找到”无效并将其标记为在一段时间后重试。

我们还可以添加其他检查,例如验证响应是否包含 IP 地址。

VALID_STATUSES = [200, 301, 302, 307, 404] 
 
def get(url, proxy): 
    try: 
        response = requests.get(url, proxies={'http': f"http://{proxy}"}, timeout=30) 
        if response.status_code in VALID_STATUSES: # valid proxy 
            print(response.status_code, response.text) 
    except Exception as e: 
        print("Exception: ", type(e)) 
 
        VALID_STATUSES = [200, 301, 302, 307, 404]

4.遍历所有代理

伟大的!我们现在需要对数组中的每个代理运行检查。get我们将像以前一样遍历代理调用列表。为了简单起见,我们将按顺序执行,但我们可以使用aiohttpasyncio.gather启动所有请求并等待它们完成。异步使代码更复杂,但它加快了网络抓取的速度

为了安全起见,该列表被硬编码为最多包含 10 个项目,以避免数百个非自愿请求。

session = requests.Session() 
# ... 
        response = session.get(url, proxies={'http': f"http://{proxy}"}, timeout=30) 
# ... 
def check_proxies(): 
    proxies = proxies_list[0:10] # limited to 10 to avoid too many requests 
    for proxy in proxies: 
        get("http://ident.me/", proxy)

5. 将工作代理与失败代理分开

检查输出日志远非理想,不是吗?我们应该为代理池保留一个内部状态。我们将它们分为三组:

  • unchecked:未知状态,待检查。
  • working: 使用此代理服务器的最后一次调用成功。
  • 不工作:最后一个请求失败。

从 s 中添加或删除项比数组更容易set,并且它们具有避免重复的优点。我们可以在列表之间移动代理而不用担心相同的代理两次。如果它存在,它就不会被添加。这将简化我们的代码:从集合中删除一个项目并将其添加到另一个。为此,我们需要稍微修改代理存储。

将存在三组,上面看到的每组一组。第一个 ,unchecked将包含文件中的代理。一个集合可以从一个数组中初始化,这让我们很容易创建它。

proxies_list = open("rotating_proxies_list.txt", "r").read().strip().split("n") 
unchecked = set(proxies_list[0:10]) # limited to 10 to avoid too many requests 
# unchecked = set(proxies_list) 
working = set() 
not_working = set() 
 
# ... 
def check_proxies(): 
    for proxy in list(unchecked): 
        get("http://ident.me/", proxy) 
#...

现在,编写辅助函数来在状态之间移动代理。每个州一名助手。他们会将代理添加到一个集合中,然后将其从其他两个集合中删除(如果存在)。这是集合派上用场的地方,因为我们不需要担心检查代理是否存在或遍历数组。如果存在或被忽略,请调用“丢弃”以删除,但不会引发异常。

例如,我们将set_working在请求成功时调用。并且该函数将在将代理添加到工作集中时从未检查或未工作集中删除代理。

def reset_proxy(proxy): 
    unchecked.add(proxy) 
    working.discard(proxy) 
    not_working.discard(proxy) 
 
def set_working(proxy): 
    unchecked.discard(proxy) 
    working.add(proxy) 
    not_working.discard(proxy) 
 
def set_not_working(proxy): 
    unchecked.discard(proxy) 
    working.discard(proxy) 
    not_working.add(proxy)

我们错过了关键部分!我们需要编辑get以在每次请求后调用这些函数。set_working对于成功者和set_not_working其他人。

def get(url, proxy): 
    try: 
        response = session.get(url, proxies={'http': f"http://{proxy}"}, timeout=30) 
        if response.status_code in VALID_STATUSES: 
            set_working(proxy) 
        else: 
            set_not_working(proxy) 
    except Exception as e: 
        set_not_working(proxy)

目前,在脚本末尾添加一些痕迹,看看它是否运行良好。unchecked由于我们运行了所有项目,因此该集合应该是空的。这些项目将填充其他两组。希望它working不是空的 – 当我们使用来自公共免费列表的代理时,它可能会发生。

#... 
check_proxies() 
 
print("unchecked ->", unchecked) # unchecked -> set() 
print("working ->", working) # working -> {"152.0.209.175:8080", ...} 
print("not_working ->", not_working) # not_working -> {"167.71.5.83:3128", ...}

6. 工作代理的使用

这是一种检查代理的直接方法,但还不是真正有用。我们现在需要一种方法来获取工作代理并出于真正原因使用它们:网络抓取实际内容。我们将创建一个函数来选择一个随机代理

我们在示例中包括了工作代理和未检查代理,如果适合您的需要,请随意使用工作代理。稍后我们会看到为什么未选中的也存在。

random不适用于集合,因此我们将使用tuple.

最后,这是一个简单的 Python 代理旋转器版本,它使用从工作代理池中随机挑选的方法:

import random 
 
def get_random_proxy(): 
    # create a tuple from unchecked and working sets 
    available_proxies = tuple(unchecked.union(working)) 
    if not available_proxies: 
        raise Exception("no proxies available") 
    return random.choice(available_proxies)

接下来,如果不存在代理,我们可以编辑get函数以使用随机代理。该proxy参数现在是可选的。我们将使用该参数来检查初始代理,就像我们之前所做的那样。但是在那之后,我们可以忘记代理列表并get在没有它的情况下进行调用。如果失败,将使用一个随机的并添加到集合中。not_working

由于我们现在想要获取实际内容,因此我们需要返回响应或引发异常。在这里找到最终版本:

def get(url, proxy = None): 
    if not proxy: 
        proxy = get_random_proxy() 
 
    try: 
        response = session.get(url, proxies={'http': f"http://{proxy}"}, timeout=30) 
        if response.status_code in VALID_STATUSES: 
            set_working(proxy) 
        else: 
            set_not_working(proxy) 
 
        return response 
    except Exception as e: 
        set_not_working(proxy) 
        raise e # raise exception

在脚本下方包含您要抓取的内容。我们只会为演示再次调用相同的测试 URL。您可以使用 Beautifulsoup 为您的目标网页创建自定义解析器。

想法是从这里开始,基于这个主干构建一个真实世界的爬虫。为了扩展它,将项目存储在持久性存储中,例如数据库(即 Redis)。

在这里,您可以实际看到有关如何在 Python 中轮换代理的最终脚本。保存并运行文件。输出应该类似于评论中的输出。

# rest of the rotating proxy script 
 
check_proxies() 
 
# real scraping part comes here 
def main(): 
    result = get("http://ident.me/") 
    print(result.status_code) # 200 
    print(result.text) # 152.0.209.175 
 
main()

还要考虑到 IP 只是防御系统检查的因素之一。随之而来的是 HTTP 标头,尤其是用户代理。我们在Scraping Without Getting Blocked上写了一个包含这些和其他技巧的列表。

使用 Selenium 旋转代理

例如,如果要抓取动态内容,则需要在 Selenium 或 Puppeteer 中轮换代理。我们已经写了一篇关于使用 Python 使用 Selenium 进行网页抓取的指南。在那里您可以看到如何使用代理设置启动浏览器。将这两个内容结合在一起,您将在 Selenium 上拥有一个代理旋转器

from selenium import webdriver 
 
# ... 
proxy = "152.0.209.175:8080" # free proxy 
options.add_argument("--proxy-server=%s" % proxy) 
with webdriver.Chrome(options=options) as driver: 
    # ...

假阴性或一次性错误会怎样?一旦我们向集合发送代理not_working,它将永远保留在那里。没有退路了。

7. 重新检查不工作的代理

我们应该不时地重新检查失败的代理。原因有很多:失败是由于网络问题、错误或代理提供商修复了它。

在任何情况下,Python 都允许我们设置Timers“仅在经过一定时间后才应运行的操作”。有不同的方法可以达到相同的目的,这很简单,只需使用三行代码即可运行它。

还记得reset_proxy函数吗?直到现在我们都没有使用它。我们将设置一个Timer为每个标记为不工作的代理运行该函数。二十秒对于真实案例来说是一个很小的数字,但对于我们的测试来说已经足够了。我们排除了失败的代理并在一段时间后将其移回未检查状态。

这就是在get_random_proxy. 修改该函数以仅使用工作代理来实现更强大的用例。然后,您可以check_proxies定期运行,这将遍历未检查的元素 – 在这种情况下,失败的代理在 sin bin 中保留了一段时间。

from threading import Timer 
 
def set_not_working(proxy): 
    unchecked.discard(proxy) 
    working.discard(proxy) 
    not_working.add(proxy) 
 
    # move to unchecked after a certain time (20s in the example) 
    Timer(20.0, reset_proxy, [proxy]).start()

还有一个更强大的系统的最终选择,但我们将把实施留给您。存储每个代理的分析和使用情况,例如失败的次数以及最后一次失败的时间。使用该信息,调整重新检查的时间 – 多次失败的代理需要更长的时间。或者甚至在工作代理的数量低于阈值时设置一些警报。

对于最终版本,请查看我们的GitHub 要点。要测试的最终 URL 在第 61 行硬编码;为您要测试的任何地址、端点或 API 更改它。由于所有调用都是通过请求库进行的,因此它将处理内容类型,无论是 HTML、JSON 还是浏览器可以接收的任何其他类型。

结论

构建一个 Python 代理旋转器对于小型抓取脚本来说似乎是可行的,请注意,在抓取登录或任何其他类型的会话/cookie 时不要轮换 IP 地址

类似文章