跳转到内容

代码示例

以下示例展示如何用 Python 获取并使用短效代理 IP。

最短路径。在用户中心【短效代理】→【生成API】→【复制链接】拿到完整地址,直接请求:

import requests
# 从用户中心【复制链接】得到的完整地址
TARGET_URL = "https://api.beesproxy.com/api/v1/proxies/?order_id=...&format=json&ip_num=10&user_token=..."
resp = requests.get(TARGET_URL, timeout=10)
data = resp.json()
# 出错时 HTTP 状态码同样是 200,必须检查 code 字段
if data.get("code") != 0:
raise RuntimeError(f"获取代理失败:[{data.get('code')}] {data.get('message')}")
for p in data["result"]:
print(f"{p['host']}:{p['port']} ({p['city_cn']})")

需要根据业务动态调整 ip_num 时用这种写法:

import requests
API_URL = "https://api.beesproxy.com/api/v1/proxies/"
ORDER_ID = "YOUR_ORDER_ID" # 用户中心【短效代理】页
USER_TOKEN = "YOUR_USER_TOKEN" # 用户中心【总览】页
def get_proxies(ip_num=10):
params = {
"order_id": ORDER_ID,
"user_token": USER_TOKEN,
"format": "json",
"ip_num": ip_num,
}
resp = requests.get(API_URL, params=params, timeout=10)
# 不要用 resp.raise_for_status():接口出错时状态码仍是 200
data = resp.json()
if data.get("code") != 0:
raise RuntimeError(f"获取代理失败:[{data.get('code')}] {data.get('message')}")
return data["result"]
proxies = get_proxies(ip_num=5)
for p in proxies:
print(f"{p['host']}:{p['port']} ({p['city_cn']})")

ip_num 超过订单的 IP 提取数时会被静默截断,代理池不足时也可能少给——不要假设返回条数一定等于请求条数

import requests
def use_proxy(proxy_host, proxy_port, target_url):
proxy_url = f"http://{proxy_host}:{proxy_port}"
proxies = {
"http": proxy_url,
"https": proxy_url,
}
return requests.get(target_url, proxies=proxies, timeout=10)
# 获取代理列表后,取第一个代理发送请求
proxies = get_proxies(ip_num=1)
proxy = proxies[0]
resp = use_proxy(proxy["host"], proxy["port"], "https://httpbin.org/ip")
print(resp.json())

这一步连不上,通常是白名单问题,见 白名单配置

import requests
def fetch_with_rotation(urls, ip_num=10):
"""轮询使用多个代理 IP 抓取多个 URL"""
proxies = get_proxies(ip_num=ip_num)
if not proxies:
raise RuntimeError("代理池暂时没有可用 IP,请稍后重试")
results = []
for i, url in enumerate(urls):
proxy = proxies[i % len(proxies)]
proxy_url = f"http://{proxy['host']}:{proxy['port']}"
try:
resp = requests.get(url, proxies={"http": proxy_url, "https": proxy_url}, timeout=10)
results.append({"url": url, "status": resp.status_code})
except Exception as e:
results.append({"url": url, "error": str(e)})
return results