自豪地由 Swiftproxy 赞助
高性能住宅代理,拥有 超过 1.8 亿 IP 池,覆盖 195+ 个国家/地区 —— 在大规模数据采集和浏览器自动化过程中提高请求成功率,减少 IP 被封禁。非常适合与
requests-ip-rotator、Playwright、Selenium 及其他爬虫工具搭配使用。
一个 Python 库,利用 AWS API Gateway 庞大的 IP 池作为代理,为网络爬虫和暴力破解生成几乎无限的 IP 地址。
该库允许用户绕过基于 IP 的网站和服务速率限制。
除非指定,否则 X-Forwarded-For 头部会被自动随机化。这是因为否则 AWS 会在该头部中发送客户端的真实 IP 地址。
AWS 的 ApiGateway 从任何可用 IP 发送请求 —— 由于 AWS 基础设施极其庞大,几乎每次请求的 IP 都不同。通过将 ApiGateway 用作代理,我们可以利用这一点从不同的 IP 发送请求。请注意,这些请求很容易被识别和阻止,因为它们会携带独特的 AWS 头部(例如 X-Amzn-Trace-Id)。
本包已上传至 PyPI,您可以通过以下任一方式安装:
pip3 install requests-ip-rotatorpython3 -m pip install requests-ip-rotator
import requests
from requests_ip_rotator import ApiGateway
# 创建网关对象并在 AWS 中初始化
gateway = ApiGateway("https://site.com")
gateway.start()
# 将网关分配给会话
session = requests.Session()
session.mount("https://site.com", gateway)
# 发送请求(IP 将被随机化)
response = session.get("https://site.com/index.php", params={"theme": "light"})
print(response.status_code)
# 删除网关
gateway.shutdown()
import requests
from requests_ip_rotator import ApiGateway
with ApiGateway("https://site.com") as g:
session = requests.Session()
session.mount("https://site.com", g)
response = session.get("https://site.com/index.php")
print(response.status_code)
请记住,如果使用方法 #1 而未通过 shutdown() 方法关闭网关,您将来可能会被收费。
API Gateway 在每个区域的前一百万次请求免费,因此对于大多数使用场景,这应该是完全免费的。
在撰写本文时,AWS 在超出免费额度后每百万次请求收费约 3 美元。
如果您的请求涉及数据流,AWS 将按每 GB 0.09 美元收取数据传输费。
建议通过环境变量设置认证。使用 awscli 运行 aws configure 即可完成,或者您也可以按照此帖子中的步骤自行设置 AWS_ACCESS_KEY_ID 和 AWS_SECRET_ACCESS_KEY 变量。
您可以按照 AWS 官方教程找到您的访问密钥 ID 和密钥。
ApiGateway 类可以通过以下可选参数创建:
from ip_rotator import ApiGateway, EXTRA_REGIONS, ALL_REGIONS
# 仅为两个区域创建到出站 HTTP IP 和端口的网关
gateway_1 = ApiGateway("http://1.1.1.1:8080", regions=["eu-west-1", "eu-west-2"])
# 为额外区域包创建到 HTTPS Google 的网关,并指定访问密钥对
gateway_2 = ApiGateway("https://www.google.com", regions=EXTRA_REGIONS, access_key_id="ID", access_key_secret="SECRET")
随后必须使用 start 方法启动 ApiGateway 对象。
默认情况下,如果该站点已存在 ApiGateway,则会使用现有端点而不创建新的。
该方法不需要参数,但可接受以下参数:
# 为站点启动新的 ApiGateway 实例,如果已有端点则定位现有端点
gateway_1.start()
# 即使某些端点已存在,也强制启动新的 ApiGateway 实例
gateway_2.start(force=True)
通过将 ApiGateway 对象附加到 requests Session 对象来发送请求。
mount 中指定的站点必须与 ApiGateway 构造函数中传递的站点一致。
import requests
# 向 gateway_1 创建的站点发送 POST 请求。将从随机 IP 发送。
session_1 = requests.Session()
session_1.mount("http://1.1.1.1:8080", gateway_1)
session_1.post("http://1.1.1.1:8080/update.php", headers={"Hello": "World"})
# 在出站请求中将 X-Forwarded-For 头部设置为 127.0.0.1(否则 X-Forwarded-For 会被随机化)。
session_1.post("http://1.1.1.1:8080/update.php", headers={"X-Forwarded-For", "127.0.0.1"})
# 从随机 IP 执行 Google 搜索查询
session_2 = requests.Session()
session_2.mount("https://www.google.com", gateway_2)
session_2.get("https://www.google.com/search?q=test")
完成使用后,务必关闭 ApiGateway 资源,以防止悬空的公共端点导致账户产生额外费用。
通过 ApiGateway 对象的 shutdown 方法实现。该方法将关闭构造函数中指定区域的所有资源。
# 这将关闭 "http://1.1.1.1:8080" 在 "eu-west-1" 和 "eu-west-2" 的所有网关代理
gateway_1.shutdown()
# 这将关闭 "https://www.google.com" 在 ip_rotator.EXTRA_REGIONS 中所有区域的网关代理
gateway_2.shutdown()
或者,如果需要,也可以有选择地关闭特定端点。只需向 shutdown() 方法传入端点数组即可,例如:
# 这将强制启动一个新网关(即即使该区域已存在端点也创建新端点),然后仅删除其中的前三个。
gateway_3 = ApiGateway("http://1.1.1.1:8082", regions=ALL_REGIONS)
endpoints = gateway_3.start(force=True)
gateway_3.shutdown(endpoints[:3])
请注意,任何以 require_manual_deletion 参数设置为 True 启动的网关都不会通过 shutdown 方法删除,必须通过 AWS CLI 或网站手动删除。
核心网关创建和组织代码改编自 RhinoSecurityLabs 的 IPRotate Burp 扩展。
X-My-X-Forwarded-For 头部转发概念最初由 ustayready 在其 fireprox 代理中提出。
| 名称 | 描述 | 是否必须 | 默认值 |
|---|
| site | 请求发送到的站点(不含路径) | 是 | |
| regions | 要设置网关的 AWS 区域 数组 | 否 | ip_rotator.DEFAULT_REGIONS |
| access_key_id | AWS 访问密钥 ID(将覆盖环境变量) | 否 | 依赖环境变量 |
| access_key_secret | AWS 访问密钥 Secret(将覆盖环境变量) | 否 | 依赖环境变量 |
| verbose | 包含状态和错误信息 | 否 | True |
| 名称 | 描述 | 是否必须 | 默认值 |
|---|
| endpoints | 预先存在的端点数组(例如来自之前的会话) | 否 | |
| force | 创建一组新端点,即使某些端点已存在 | 否 | False |
| require_manual_deletion | 布尔值,指定 ApiGateway 是否应在 shutdown() 调用后持久保留 | 否 | False |