brainstorming
brainstorming
You MUST use this before any creative work - creating features, building components, adding functionality, modifying behavior, or planning anything new, in software or out of it (a talk, a business, a renovation).
rotate-scraping-proxies当抓取任务被封锁时,使用与服务商无关的代理轮换方案进行升级处置——在数据中心、 住宅和移动代理池之间做出选择,将轮换机制与 scrapling 集成,为有状态流程配置 会话粘滞,监控成本与健康度,并始终在法律与伦理边界之内行事。仅当 `headless-web-scraping` 客户端隐匿手段(StealthyFetcher、限速、robots.txt) 已不足以应对、且流量本身合法时,才将其作为下一步。
Independent GitHub snapshot · Automated classification · Original documentation governs
headless-web-scraping 客户端隐匿手段(StealthyFetcher、限速、robots.txt)
已不足以应对、且流量本身合法时,才将其作为下一步。
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
locale: zh-CN
source_locale: en
source_commit: 75ded7a1
translator: "Claude + human review"
translation_date: "2026-04-14"当客户端隐匿手段已经用尽时,在网络层面对抓取任务进行升级处置。代理轮换是最后 的手段,而非默认选项——它成本昂贵、涉及伦理争议,且极易被滥用。本技能不仅讲授 如何用好轮换,也同样重视说明何时不该使用它。
headless-web-scraping(Fetcher → StealthyFetcher → DynamicFetcher)都
已尝试过,目标站点依然返回 403/429 或进行地理封锁robots.txt 允许访问该路径python-requests)不适用场景:存在公开 API(请使用它)、站点的服务条款禁止自动化访问、 抓取行为会绕过地域授权、目的是欺诈 / 撞库 / 抢购机器人 / 内容盗版。
整个工作流的闸门是一份有书面记录的法律与伦理审查。跳过这一步是最大的危害 来源。
Inputs to confirm before writing any code:
Expected: 每个问题都有可辩护的书面答案。出现第一个"否"或"未知" 即停止流程,直到问题得到解决。
On failure:
不同类型的代理池在成本、被检测难度和伦理属性上各有差异。选择能切实解决你 所遇封锁的最低价位层级。
| 代理池类型 | 可检测性 | 成本 | 最适用于 |
|---|---|---|---|
| 数据中心 | 高(容易被 Cloudflare/Akamai 封禁) | $ | 不设真正反爬措施、仅需地理切换的站点 |
| 住宅 | 低(真实的 ISP IP) | $$ | 屏蔽数据中心 ASN 的站点 |
| 移动 | 极低(运营商级 NAT,与数千人共享) | $$ | 连住宅代理都封禁的站点(少见) |
住宅与移动代理的伦理警示:这些代理池会让你的流量途经真实消费者的网络 连接。代理池运营商的同意模型参差不齐——有些会向用户付费,有些则把出口 节点同意条款捆绑在用户根本不会阅读的"免费 VPN"最终用户许可协议里。请 优先选择那些经过审计、采用明确选择加入(opt-in)同意模型的服务商。 如果你不愿意让一个陌生人把抓取流量通过你自家路由器转发,那就不要把你的 流量通过别人的路由器转发。
Expected: 有书面记录的选择,选择了最低价位的可行层级,并简要说明为何 拒绝更高层级(或为何需要更高层级)。
On failure:
把代理接入 scrapling 的 fetcher。凭证从环境变量读取——绝不硬编码,绝不把
.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)
Expected: 请求成功,且出口 IP 在不同调用间发生变化。在执行真正的抓取
前,先调用一个 IP 回显接口(例如 https://api.ipify.org)进行确认。
On failure:
407 Proxy Authentication Required——凭证错误,或密码的 URL 编码出错
(重新对特殊字符进行编码)-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
Expected: 有状态流程在多个请求之间保留 cookie;批量匿名抓取在多个 请求之间呈现 IP 变化;失效的代理会被跳过,而不是陷入循环。
On failure:
代理流量同时计费到每 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
Expected: 预算上限在成本失控前就先触发。日志中记录了每个代理的成功 率,便于识别并排除行为异常的出口 IP。
On failure:
gateway.、proxy=、服务商主机名).env(或同类文件)已列入 .gitignorerobots.txt 仍被遵守——轮换不能凌驾于它之上StealthyFetcher 和限速;轮换代价昂贵,在非必要情况下部署它也不
符合伦理。robots.txt:轮换不授予任何许可。
禁止访问的指令依然是禁止访问的指令。