logo

如何在 Python 中轮换代理

2024-06-21 23:51
本文介绍了如何定制属于我们自己的代理轮换器。

代理可以隐藏您的真实 IP 地址。但如果 IP 被禁止访问了,你就需要一个新的 IP。或者,您可以维护一个代理列表,并且在每次 Python 请求时轮换使用它们。

现在让我们来看看如何在 Pytho n 中构建我们自己的代理旋转器。我们将从一个普通的代理列表开始,检查并标记可用的代理,并提供简单的监控来从列表中移除失效的代理。虽然本教程使用的是 Python,但这个思路可以在任何你喜欢用于数据抓取的编程语言中实现。

让我们开始吧!

如何轮换我的 IP?

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

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

但并非所有代理都是一样的。

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

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

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

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

现在,我们将学习如何使用 Python 轮换静态代理。我们走吧!

如何在 Python 中轮换代理?

要按照本教程操作,您需要在设备上使用 python3。某些系统预装了它。之后,通过运行 pip install来安装所有需要的依赖。

pip install aiohttp

1. 获取代理列表

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

网上有几个免费代理列表。对于 demo,挑选其中的一个使用即可,将它们其中之一保存到本地(仅包含 URL 即可)(rotating_proxies_list.txt

从站点导出和复制代理。如果您想自动化该过程,该站点还提供了一个代理 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:端口的有效性进行任何校验。我们会尽量保持简单。

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

2.检查代理

假设我们希望大规模运行爬虫。这个演示是简化版,但我们的核心思路是将代理及其“健康状态”存储在一个可靠的介质中,例如数据库。我们将在这里使用内存数据结构,这些数据结构在每次运行后都会消失,但你能明白这个概念。这将是我们的代理池。

首先,让我们编写一个简单的函数来检查代理是否有效。为此,请调用 ident.me,这是一个将返回 IP 的 HTML 响应的网页。我们还将使用 Python requests库。

目前,这段代码从代理列表中取出一个代理并调用指定的 URL。大部分代码是模板代码,但很快就会派上用场。可能出现两种结果:

  • 😊 如果一切顺利,它会打印响应的内容和状态码(即 200),这可能就是代理的 IP。
  • 😔 如果由于超时或其他原因导致出错,错误信息会被打印出来。这通常意味着该代理不可用或无法处理请求。当使用免费代理时,这类情况会经常出现。
# rotate-proxies-python.py
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.将工作代理与失败的代理分开

我们将代理池分成三组:

  • 未检查:状态未知,待检查。
  • 有效:上次使用该代理服务器的请求成功。
  • 无效:上次使用该代理服务器的请求失败。

使用集合比数组更容易添加或移除项目,并且还有避免重复的优点。我们可以在列表之间移动代理,而不必担心出现重复项。如果代理已经存在,则不会被重复添加。这将简化我们的代码:从一个集合中移除一个项目并将其添加到另一个集合。为此,我们需要稍微修改代理的存储方式。

我们将有三个集合,每个集合对应一个组。初始集合(未检查)中将包含从文件中读取的代理。集合可以从数组初始化,这使我们创建它变得容易。

# rotate-proxies-python.py
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)
#...

现在,编写辅助函数来在不同状态之间移动代理。每个状态写一个函数。这些函数会将代理添加到一个集合中,并从其他两个集合中移除(如果存在的话)。这里使用集合非常方便,因为我们不需要担心检查代理是否存在或者遍历数组。调用 discard 来移除代理,如果代理存在就移除,否则忽略,并且不会引发任何异常。

例如,当一个请求成功时,我们会调用 set_working 函数。这个函数会将代理从未检查或无效的集合中移除,同时将其添加到有效的集合中。

# rotate-proxies-python.py
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

# rotate-proxies-python.py
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)

目前,在脚本末尾添加一些日志以查看其运行情况。由于我们已经运行了所有项目,未检查的集合应当是空的。而另外两个集合将会被填充。希望有效的集合不是空的 😅。当我们使用公共的免费代理列表时,可能会出现这种情况。

# rotate-proxies-python.py
#...
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 不能直接用于集合,因此我们需要将集合转换为元组。

最后,这里提供一个简单的 Python 代理轮换器版本,它从有效代理池中随机选取代理:

# rotate-proxies-python.py
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 函数,使其在没有提供代理时使用随机代理。代理参数现在是可选的。我们将使用该参数来检查初始代理,就像之前一样。但在那之后,我们可以忘记这个列表并直接调用 get 函数而不传递它。如果使用随机代理失败,那么该代理将被添加到 not_working 集合中。

由于我们现在需要获取实际内容,因此需要返回响应或引发异常。以下是最终版本

# rotate-proxies-python.py
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。你可以使用 Beautiful Soup 为你的目标网页创建一个自定义解析器。

我们的想法是从这里开始,并基于此框架构建一个实际应用的抓取工具。为了扩展它,可以将项目存储在持久性存储中,例如数据库(如 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 Head,特别是用户代理。请查阅我们关于这些和其他避免被封锁的抓取技巧的文章。

使用 Selenium 轮换代理

如果你要抓取动态内容,例如在 Selenium 或 Puppeteer 中轮换代理。在那里,你可以看到如何使用代理设置启动浏览器。将两者结合起来,你就可以在 Selenium 中实现一个代理轮换器。

# rotate-proxies-python.py
from selenium import webdriver
from selenium.webdriver.chrome.options import Options

# ...
proxy = "152.0.209.175:8080" # free proxy

options = Options()
options.add_argument("--proxy-server=%s" % proxy)

driver = webdriver.Chrome(options=options)
# ...

对于误报或一次性错误会发生什么?一旦我们将代理发送到 not_working 集合,它将永远保留在那里。没有回撤的办法。

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

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

无论如何,Python 允许我们设置定时器,"在经过一段时间后才运行的动作。" 有多种方法可以实现相同的目的,而这种方法足够简单,只需三行代码。

还记得 reset_proxy 函数吗?直到现在我们都没有使用它。我们将设置一个定时器,让它为每个被标记为不可用的代理运行。二十秒对于实际情况来说是一个很小的时间,但对于我们的测试来说足够了。我们将排除一个失败的代理,并在一段时间后将其移动回未检查状态。

这是在 get_random_proxy 函数中同时使用有效代理和未检查代理的原因!修改该函数以仅使用有效代理,可以使其用于更稳健的应用场景。然后,你可以定期运行 check_proxies,它会遍历未检查的元素——在这个例子中,也就是那些在等待区待了一段时间的失败代理。

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()

还有一个更稳健的系统选项,但我们将实现留给你自己完成。可以为每个代理存储分析和使用情况,例如,失败的次数以及最后一次失败的时间。利用这些信息,调整重新检查的时间——对多次失败的代理设置更长的时间。或者,如果工作代理的数量低于某个阈值,甚至可以设置一些警报。

结论

构建一个 Python 代理轮换器对于小型抓取脚本来说可能看起来是可行的,但它也可能变得痛苦起来。不过,嘿,你做到了!👏

需要注意的是,当抓取登录状态或任何其他类型的会话/ cookies 时,不要轮换 IP 地址。

让我们快速回顾一下你今天学到的内容:

  1. 如何获取轮换代理?
  2. 将代理列表存储为纯文本。
  3. 从文件中导入为数组。
  4. 检查每个代理。
  5. 分离出可用的代理。
  6. 在抓取过程中检查错误并从可用列表中移除它们。
  7. 不时地重新检查不可用的代理。
导航目录