代码示例
以下示例展示如何用 Python 获取并使用短效代理 IP。
写法 A:直接使用复制来的链接
Section titled “写法 A:直接使用复制来的链接”最短路径。在用户中心【短效代理】→【生成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']})")写法 B:在代码里拼接参数
Section titled “写法 B:在代码里拼接参数”需要根据业务动态调整 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 提取数时会被静默截断,代理池不足时也可能少给——不要假设返回条数一定等于请求条数。
使用代理 IP 发送请求
Section titled “使用代理 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())这一步连不上,通常是白名单问题,见 白名单配置。
轮询使用多个代理
Section titled “轮询使用多个代理”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