技能创建器
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
預期: 有狀態流跨請求保持 cookies;批量匿名抓取顯示請求間 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(或等效)已在 .gitignore 中robots.txt 仍受尊重——輪換不可凌駕之StealthyFetcher 與速率限制;輪換昂貴且不必要時部署不道德。robots.txt:輪換不予許可。指令即指令。