技能创建器
skill-creator
创建、改进和评估自定义技能,把专业流程转化为可重复使用的 AI 能力。
rotate-scraping-proxies【数据与研究】Escalate blocked scrape → proxy rotation. Datacenter|residential|mobile pool, scrapling integ, sticky session, cost+health monitor, legal/ethical bounds. Use → after `headless-web-scraping` client stealth fails + traffic legit.
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
headless-web-scraping client stealth fails + traffic legit.
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, networkingNetwork-layer escalation when client stealth exhausted. Last resort, not default — expensive, ethically charged, easily misused. Skill teaches when not to use as much as how.
headless-web-scraping (Fetcher → StealthyFetcher → DynamicFetcher) tried + still 403/429/geo-blockrobots.txt permitspython-requests)Don't use → public API exists, ToS forbids automation, geo-license circumvention, fraud|cred-stuff|sneaker-bot|piracy.
Gate workflow on documented review. Skip = biggest harm source.
Inputs to confirm before writing any code:
→ Every Q has defensible written ans. First "no"|"unknown" stops proc.
If err:
Diff cost, detect, ethics. Cheapest tier solving block.
| Pool type | Detectability | Cost | Best for |
|---|---|---|---|
| Datacenter | High (easily blocked by Cloudflare/Akamai) | $ | Sites with no real anti-bot, geo-shifting only |
| Residential | Low (real ISP IPs) | $$ | Sites that block datacenter ASNs |
| Mobile | Very low (carrier-grade NAT, shared with thousands) | $$ | Sites that even block residential (rare) |
Ethical caveat residential+mobile: routes via real consumer connections. Provider consent varies — some pay, some bundle exit-node consent into "free VPN" EULAs unread. Prefer audited opt-in. If wouldn't be comfortable w/ stranger sending scrape via your home router → don't send via theirs.
→ Documented choice + cheapest viable + brief why higher rejected (or needed).
If err:
Wire proxy → scrapling fetcher. Read creds from env, never hard-code, never commit .env.
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)
→ Reqs succeed + egress IP varies. Confirm via IP echo (https://api.ipify.org) before real scrape.
If err:
-rotating or per-req flagGranularity per workload + keep pool healthy.
# 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
→ Stateful preserves cookies; bulk shows IP variance; dead proxies skipped not looped.
If err:
Per-GB + per-req cost. Runaway scrape → runaway invoice. Always cap+abort.
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
→ Caps trigger before runaway. Logs show per-proxy success → identify+exclude bad IP.
If err:
gateway., proxy=, hostname).env in .gitignorerobots.txt respected — rotation doesn't overrideStealthyFetcher+rate first.robots.txt because rotation: Doesn't grant permission. Directive=directive.