技能创建器
skill-creator
创建、改进和评估自定义技能,把专业流程转化为可重复使用的 AI 能力。
rotate-scraping-proxies【数据与研究】Escalate blocked scraping campaigns with provider-neutral proxy rotation — decide between datacenter, residential, and mobile pools, integrate rotation with scrapling, configure session stickiness for stateful flows, monitor cost and health, and stay inside legal and ethical boundaries. Use as the next step after `headless-web-scraping` client-side s…
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
headless-web-scraping client-side stealth (StealthyFetcher,
rate limiting, robots.txt) is insufficient and traffic is legitimate.
license: MIT
allowed-tools: Bash Read Write Grep Glob
metadata:
author: Philipp Thoss
version: "1.0"
domain: web-scraping
complexity: intermediate
language: Python
tags: web-scraping, proxies, rotation, residential, scrapling, networking爬蟲被阻而客戶端隱身已盡,乃以網層之代理輪轉升而圖之。輪轉者,末路之計,非默用也——其費高、其德有疵、易為惡用。此技教其不用之時,與其善用之法。
headless-web-scraping(Fetcher → StealthyFetcher → DynamicFetcher)已試,目標仍返 403/429/地阻乃用robots.txt 許其徑乃用python-requests)乃用勿用之時:有公 API(用之)、站之 ToS 禁自動之訪、欲越地之授、欲行詐/憑據塞/球鞋搶/盜版者。
全程繫於書面之法德審。略此者,害之最大源也。
Inputs to confirm before writing any code:
得:每問皆有可辯之書答。一遇「否」或「未知」則止,俟其解。
敗則:
各池類之費、可察、德之屬皆異。擇能解阻之最廉者。
| 池類 | 可察 | 費 | 宜於 |
|---|---|---|---|
| 數據中心 | 高(Cloudflare/Akamai 易阻) | $ | 無真正反爬之站,唯地遷 |
| 住宅 | 低(真 ISP 之 IP) | $$ | 阻數據中心 ASN 之站 |
| 行動 | 極低(運營商級 NAT,與千者共) | $$ | 連住宅亦阻(罕) |
住宅與行動之德戒:此池借真消費者之網以行汝之流量。池運營者之同意之模有異——或付用者,或將出口節點之同意捆於「免費 VPN」之 EULA 中,用者未嘗讀也。宜擇有審計、明選之同意之供者。若汝不願陌人由汝家路由送其爬流,則勿令汝之流經他人之路由。
得:書面之擇——可行之最廉等,附簡注述高等何以拒(或何以需高等)。
敗則:
代理裝入 scrapling 之取者。憑據自環境變量讀——勿硬編,勿提交 .env 入 git。
import os
import random
from scrapling import Fetcher, StealthyFetcher
# Pattern A: provider-managed rotating endpoint (one URL, provider rotates per request)
PROXY_URL = os.environ["SCRAPING_PROXY_URL"] # http://user:pass@gateway.example:7777
fetcher = StealthyFetcher()
fetcher.configure(
headless=True,
timeout=60,
network_idle=True,
proxy=PROXY_URL,
)
# Pattern B: explicit pool, rotate yourself
POOL = os.environ["SCRAPING_PROXY_POOL"].split(",") # comma-separated URLs
def fetch_with_rotation(url):
proxy = random.choice(POOL)
fetcher = StealthyFetcher()
fetcher.configure(headless=True, timeout=60, proxy=proxy)
return fetcher.get(url)
得:諸請皆成,出口之 IP 每呼有變。實爬之前,擊一 IP 回響之點(如 https://api.ipify.org)以驗。
敗則:
-rotating 或每請之旗依所為擇輪轉之粒度,後守池之健。
# Sticky session for stateful flows (login, multi-page checkout-like crawls)
# Most providers expose a session ID via the username:
# user-session-abc123:pass@gateway.example:7777
# All requests with the same session ID exit through the same IP for ~10 min.
# Per-request rotation for anonymous bulk scraping (default)
# Pool health check — call before bulk run
def check_pool(pool, sample_size=5):
sample = random.sample(pool, min(sample_size, len(pool)))
alive = []
for proxy in sample:
try:
r = StealthyFetcher().configure(proxy=proxy, timeout=10).get(
"https://api.ipify.org"
)
if r.status == 200:
alive.append(proxy)
except Exception:
pass
return alive
# Backoff on transient proxy failures
def fetch_with_backoff(url, max_attempts=3):
for attempt in range(max_attempts):
try:
r = fetch_with_rotation(url)
if r.status not in (407, 502, 503):
return r
except Exception:
pass
time.sleep(2 ** attempt)
return None
得:有狀之流跨請保 cookie;匿名之大爬諸請示 IP 之變;死代理略而不循環。
敗則:
代理之流量有每 GB 之費與每請之費。失控之爬,致失控之賬。必含限與斷。
import time
class ScrapeBudget:
def __init__(self, max_requests, max_duration_seconds, max_failures):
self.max_requests = max_requests
self.max_duration = max_duration_seconds
self.max_failures = max_failures
self.requests = 0
self.failures = 0
self.start = time.monotonic()
def allow(self):
if self.requests >= self.max_requests:
return False, "request cap reached"
if time.monotonic() - self.start >= self.max_duration:
return False, "time cap reached"
if self.failures >= self.max_failures:
return False, "failure cap reached (circuit breaker)"
return True, None
def record(self, success):
self.requests += 1
if not success:
self.failures += 1
budget = ScrapeBudget(max_requests=1000, max_duration_seconds=3600, max_failures=20)
for url in target_urls:
ok, reason = budget.allow()
if not ok:
print(f"Aborting: {reason}")
break
response = fetch_with_backoff(url)
budget.record(success=response is not None)
time.sleep(1) # rate limiting still applies even with rotation
得:預算之限觸於失控之費前。日誌示每代理之成率,惡之出口 IP 可識而排。
敗則:
gateway.、proxy=、供者主機名).env(或等者)入 .gitignorerobots.txt 仍守——輪轉不能凌之StealthyFetcher 與速限;輪轉費高而德疵,無故勿施robots.txt:輪轉不授許可。其令即令也