Python Web Scraping 2026: Anti-Bot Bypass and Production Crawler Guide
编程语言
The Challenges of Web Scraping in 2026
The Web of 2026 is far from the era of simple static pages. Modern websites deploy multi-layered anti-scraping mechanisms that make data collection increasingly difficult:
- Anti-Bot Detection: Commercial solutions like Cloudflare Turnstile, Akamai Bot Manager, and PerimeterX use behavioral analysis, TLS fingerprinting, and Canvas fingerprinting to precisely identify automated requests
- CAPTCHA Evolution: From simple image captchas to hCaptcha, reCAPTCHA v3 invisible verification, and GeeTest slider captchas, some even requiring AI-assisted solving
- Dynamic Rendering: SPA applications built with React/Vue/Svelte generate page content dynamically via JavaScript, rendering traditional requests + BeautifulSoup approaches completely ineffective
- Rate Limiting: Multi-dimensional throttling based on IP + User-Agent + Cookie, where simple time-interval controls can no longer bypass
- Data Encryption: Critical data protected through Webpack obfuscation, API signature verification, and encrypted response payloads
Facing these challenges, we need a systematic scraping approach. This article builds a production-grade scraping system step by step, starting from core concepts.
Core Concepts Reference
| Concept | Description | Typical Tools |
|---|---|---|
| Static Scraping | Direct HTML request + DOM parsing | requests + BeautifulSoup |
| Dynamic Rendering | Simulate browser JS execution | Playwright, Selenium |
| Anti-Bot Bypass | Disguise request characteristics | Proxy pools, fingerprint masking, CAPTCHA services |
| Distributed Scraping | Multi-node concurrent crawling | Scrapy + Redis, Celery |
| Data Pipeline | Clean, transform, store scraped results | Pandas, SQLAlchemy, Item Pipeline |
| Incremental Scraping | Only fetch new/changed content | URL dedup, content hash comparison |
| Rate Limiting | Control request frequency to avoid bans | asyncio.Semaphore, Scrapy AutoThrottle |
Five Core Challenges
Challenge 1: TLS Fingerprint Detection
Modern anti-bot systems identify request origins through TLS handshake characteristics (JA3/JA4 fingerprints). Python's requests and aiohttp default TLS fingerprints differ dramatically from browsers, making them trivially detectable.
Challenge 2: Browser Fingerprint Tracking
Return values from browser APIs like Canvas, WebGL, AudioContext, and font lists form unique fingerprints. Headless browsers have detectable differences from real browsers.
Challenge 3: JavaScript Dynamic Rendering
Core data in SPA applications loads asynchronously via AJAX/Fetch. The initial HTML does not contain target data; you must wait for JS execution to complete before extracting content.
Challenge 4: CAPTCHA Interception
Key operations like login, search, and pagination trigger captchas. Traditional OCR approaches perform poorly against modern captchas, requiring third-party solving services or AI models.
Challenge 5: IP Bans and Rate Limiting
High-frequency requests trigger IP bans. Single proxies fail easily, requiring a high-availability proxy pool with intelligent rotation strategies.---
Pattern 1: Scrapy Framework for Large-Scale Crawling
Scrapy is Python's most mature scraping framework, ideal for structured, large-scale data collection tasks.
Project Initialization
pip install scrapy scrapy-playwright
scrapy startproject news_crawler
cd news_crawler
scrapy genspider tech_news example.com
Complete Spider Example
# news_crawler/spiders/tech_news.py
import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule
from ..items import NewsItem
class TechNewsSpider(CrawlSpider):
name = "tech_news"
allowed_domains = ["example.com"]
start_urls = ["https://example.com/tech"]
custom_settings = {
"CONCURRENT_REQUESTS": 16,
"DOWNLOAD_DELAY": 1.5,
"AUTOTHROTTLE_ENABLED": True,
"AUTOTHROTTLE_TARGET_CONCURRENCY": 8,
"AUTOTHROTTLE_MAX_DELAY": 10,
"ROBOTSTXT_OBEY": True,
"USER_AGENT": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/126.0.0.0 Safari/537.36"
),
"DEFAULT_REQUEST_HEADERS": {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
},
"FEEDS": {
"output/tech_news.json": {
"format": "json",
"encoding": "utf-8",
"indent": 2,
},
},
"ITEM_PIPELINES": {
"news_crawler.pipelines.CleanHtmlPipeline": 100,
"news_crawler.pipelines.DeduplicatePipeline": 200,
"news_crawler.pipelines.SqlitePipeline": 300,
},
}
rules = (
Rule(
LinkExtractor(allow=r"/tech/article/\d+"),
callback="parse_article",
follow=True,
),
Rule(
LinkExtractor(allow=r"/tech/page/\d+"),
follow=True,
),
)
def parse_article(self, response):
item = NewsItem()
item["title"] = response.css("h1.article-title::text").get("").strip()
item["author"] = response.css(".author-name::text").get("").strip()
item["publish_time"] = response.css(
"time.article-time::attr(datetime)"
).get("")
item["content"] = response.css(".article-content").get("")
item["tags"] = response.css(".tag-item::text").getall()
item["url"] = response.url
yield item
Item Definition
# news_crawler/items.py
import scrapy
from itemloaders.processors import TakeFirst, MapCompose, Join
def strip_text(value):
return value.strip() if value else ""
class NewsItem(scrapy.Item):
title = scrapy.Field(
input_processor=MapCompose(strip_text),
output_processor=TakeFirst(),
)
author = scrapy.Field(
input_processor=MapCompose(strip_text),
output_processor=TakeFirst(),
)
publish_time = scrapy.Field(output_processor=TakeFirst())
content = scrapy.Field(
input_processor=MapCompose(strip_text),
output_processor=Join(""),
)
tags = scrapy.Field()
url = scrapy.Field(output_processor=TakeFirst())
Pipeline Data Processing
# news_crawler/pipelines.py
import hashlib
import sqlite3
from html.parser import HTMLParser
from itemadapter import ItemAdapter
class HtmlStripper(HTMLParser):
def __init__(self):
super().__init__()
self.reset()
self.fed = []
def handle_data(self, d):
self.fed.append(d)
def get_data(self):
return "".join(self.fed)
class CleanHtmlPipeline:
"""Strip HTML tags from content"""
def process_item(self, item, spider):
adapter = ItemAdapter(item)
content = adapter.get("content", "")
if content:
stripper = HtmlStripper()
stripper.feed(content)
adapter["content"] = stripper.get_data().strip()
return item
class DeduplicatePipeline:
"""Deduplicate based on content hash"""
def __init__(self):
self.seen_hashes = set()
def process_item(self, item, spider):
adapter = ItemAdapter(item)
content = adapter.get("content", "")
content_hash = hashlib.md5(content.encode()).hexdigest()
if content_hash in self.seen_hashes:
spider.logger.info(f"Duplicate content, skipping: {adapter.get('url')}")
raise scrapy.exceptions.DropItem("Duplicate content")
self.seen_hashes.add(content_hash)
return item
class SqlitePipeline:
"""Store to SQLite"""
def open_spider(self, spider):
self.conn = sqlite3.connect("news_data.db")
self.cursor = self.conn.cursor()
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT,
author TEXT,
publish_time TEXT,
content TEXT,
tags TEXT,
url TEXT UNIQUE,
crawl_time TEXT
)
""")
self.conn.commit()
def close_spider(self, spider):
self.conn.close()
def process_item(self, item, spider):
adapter = ItemAdapter(item)
self.cursor.execute(
"""INSERT OR IGNORE INTO articles
(title, author, publish_time, content, tags, url)
VALUES (?, ?, ?, ?, ?, ?)""",
(
adapter.get("title"),
adapter.get("author"),
adapter.get("publish_time"),
adapter.get("content"),
",".join(adapter.get("tags", [])),
adapter.get("url"),
),
)
self.conn.commit()
return item
Running the Spider
scrapy crawl tech_news -s LOG_LEVEL=INFO
```---
## Pattern 2: Playwright for Dynamic SPA Scraping
For JavaScript-rendered SPA applications, Playwright is the top choice in 2026, faster and more stable than Selenium.
### Installation
```bash
pip install playwright
playwright install chromium
Basic Dynamic Page Scraping
import asyncio
from playwright.async_api import async_playwright
async def scrape_spa():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/126.0.0.0 Safari/537.36"
),
locale="en-US",
)
page = await context.new_page()
await page.goto(
"https://spa-example.com/products",
wait_until="networkidle",
)
await page.wait_for_selector(".product-card", timeout=10000)
for _ in range(3):
await page.evaluate(
"window.scrollTo(0, document.body.scrollHeight)"
)
await page.wait_for_timeout(1500)
products = await page.evaluate("""() => {
const cards = document.querySelectorAll('.product-card');
return Array.from(cards).map(card => ({
name: card.querySelector('.product-name')?.textContent?.trim() || '',
price: card.querySelector('.price')?.textContent?.trim() || '',
rating: card.querySelector('.rating')?.textContent?.trim() || '',
image: card.querySelector('img')?.src || '',
}));
}""")
print(f"Scraped {len(products)} products")
for product in products[:5]:
print(f" {product['name']} - {product['price']}")
await browser.close()
return products
asyncio.run(scrape_spa())
Handling Infinite Scroll + Pagination
import asyncio
from playwright.async_api import async_playwright
async def scrape_infinite_scroll():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
await page.goto(
"https://example.com/feed", wait_until="networkidle"
)
all_items = []
previous_count = 0
max_scrolls = 20
for scroll_round in range(max_scrolls):
await page.evaluate(
"window.scrollTo(0, document.body.scrollHeight)"
)
try:
await page.wait_for_function(
"document.querySelectorAll('.feed-item').length > arguments[0]",
previous_count,
timeout=5000,
)
except Exception:
print(f"Round {scroll_round + 1}: no new content, stopping")
break
current_items = await page.evaluate("""() =>
Array.from(document.querySelectorAll('.feed-item')).map(
el => el.textContent.trim()
)
""")
all_items = current_items
previous_count = len(current_items)
print(f"Round {scroll_round + 1}: {len(current_items)} items total")
await page.wait_for_timeout(
int(asyncio.get_event_loop().time() % 2000 + 1000)
)
await browser.close()
return all_items
asyncio.run(scrape_infinite_scroll())
Intercepting Network Requests to Extract API Data
import asyncio
import json
from playwright.async_api import async_playwright
async def intercept_api_data():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
api_responses = []
async def handle_response(response):
if "/api/v2/products" in response.url and response.status == 200:
try:
data = await response.json()
api_responses.append(data)
print(f"Intercepted API data: {response.url}")
except Exception:
pass
page.on("response", handle_response)
await page.goto(
"https://example.com/shop", wait_until="networkidle"
)
await page.wait_for_timeout(3000)
all_products = []
for resp in api_responses:
items = resp.get("data", {}).get("items", [])
all_products.extend(items)
print(f"Retrieved {len(all_products)} items via API interception")
with open("api_data.json", "w", encoding="utf-8") as f:
json.dump(all_products, f, ensure_ascii=False, indent=2)
await browser.close()
return all_products
asyncio.run(intercept_api_data())
```---
## Pattern 3: Anti-Bot Bypass Techniques
### Header Spoofing
```python
import random
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0",
]
ACCEPT_LANGUAGES = [
"en-US,en;q=0.9",
"en-GB,en;q=0.9,en-US;q=0.8",
"en-US,en;q=0.9,fr;q=0.8",
]
def get_random_headers():
"""Generate randomized request headers"""
return {
"User-Agent": random.choice(USER_AGENTS),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": random.choice(ACCEPT_LANGUAGES),
"Accept-Encoding": "gzip, deflate, br",
"DNT": "1",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Cache-Control": "max-age=0",
}
Proxy Pool Rotation
import asyncio
import aiohttp
import random
from dataclasses import dataclass, field
@dataclass
class ProxyPool:
"""High-availability proxy pool"""
proxies: list[str] = field(default_factory=list)
failed_count: dict[str, int] = field(default_factory=dict)
max_failures: int = 3
def add_proxy(self, proxy: str):
self.proxies.append(proxy)
self.failed_count[proxy] = 0
def get_proxy(self) -> str | None:
available = [
p for p in self.proxies
if self.failed_count.get(p, 0) < self.max_failures
]
if not available:
return None
return random.choice(available)
def mark_success(self, proxy: str):
self.failed_count[proxy] = 0
def mark_failure(self, proxy: str):
self.failed_count[proxy] = self.failed_count.get(proxy, 0) + 1
if self.failed_count[proxy] >= self.max_failures:
if proxy in self.proxies:
self.proxies.remove(proxy)
print(f"Proxy {proxy} removed (too many failures)")
async def fetch_with_proxy(pool: ProxyPool, url: str, max_retries: int = 3):
"""Fetch with proxy pool rotation"""
for attempt in range(max_retries):
proxy = pool.get_proxy()
if not proxy:
print("Proxy pool exhausted, using direct connection")
proxy = None
try:
async with aiohttp.ClientSession() as session:
async with session.get(
url,
proxy=f"http://{proxy}" if proxy else None,
headers=get_random_headers(),
timeout=aiohttp.ClientTimeout(total=15),
ssl=False,
) as response:
if response.status == 200:
if proxy:
pool.mark_success(proxy)
return await response.text()
elif response.status in (403, 429):
if proxy:
pool.mark_failure(proxy)
print(f"Status {response.status}, switching proxy (attempt {attempt + 1})")
await asyncio.sleep(random.uniform(2, 5))
except Exception as e:
if proxy:
pool.mark_failure(proxy)
print(f"Request error: {e} (attempt {attempt + 1})")
await asyncio.sleep(random.uniform(1, 3))
return None
Playwright Fingerprint Masking
import asyncio
from playwright.async_api import async_playwright
async def stealth_scrape():
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=True,
args=[
"--disable-blink-features=AutomationControlled",
"--disable-features=IsolateOrigins,site-per-process",
"--disable-dev-shm-usage",
"--no-sandbox",
],
)
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
screen={"width": 1920, "height": 1080},
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/126.0.0.0 Safari/537.36"
),
locale="en-US",
timezone_id="America/New_York",
geolocation={"latitude": 40.7128, "longitude": -74.0060},
permissions=["geolocation"],
color_scheme="light",
)
page = await context.new_page()
await page.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});
window.chrome = {
runtime: {}, loadTimes: function() {},
csi: function() {}, app: {}
};
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
parameters.name === 'notifications' ?
Promise.resolve({ state: Notification.permission }) :
originalQuery(parameters)
);
Object.defineProperty(navigator, 'plugins', {
get: () => [1, 2, 3, 4, 5]
});
Object.defineProperty(navigator, 'languages', {
get: () => ['en-US', 'en']
});
""")
await page.goto("https://example.com", wait_until="networkidle")
content = await page.content()
print(f"Page length: {len(content)}")
await browser.close()
return content
asyncio.run(stealth_scrape())
```---
## Pattern 4: Async Scraping with aiohttp + BeautifulSoup
For scenarios that do not require JS rendering, async scraping is the most efficient choice.
```python
import asyncio
import aiohttp
from bs4 import BeautifulSoup
from dataclasses import dataclass, field
import json
import time
@dataclass
class AsyncScraper:
"""Async scraper core class"""
base_url: str
max_concurrency: int = 10
request_delay: float = 0.5
timeout: int = 15
results: list = field(default_factory=list)
visited: set = field(default_factory=set)
async def fetch_page(self, session: aiohttp.ClientSession, url: str) -> str | None:
"""Fetch a single page"""
try:
async with session.get(
url,
headers=get_random_headers(),
timeout=aiohttp.ClientTimeout(total=self.timeout),
ssl=False,
) as response:
if response.status == 200:
return await response.text()
print(f"[{response.status}] {url}")
return None
except asyncio.TimeoutError:
print(f"[Timeout] {url}")
return None
except Exception as e:
print(f"[Error] {url}: {e}")
return None
def parse_page(self, html: str, url: str) -> dict | None:
"""Parse page content"""
soup = BeautifulSoup(html, "lxml")
title = soup.select_one("h1.article-title")
if not title:
return None
content_paragraphs = soup.select(".article-content p")
content = "\n".join(p.get_text(strip=True) for p in content_paragraphs)
return {
"title": title.get_text(strip=True),
"content": content,
"url": url,
}
async def scrape_urls(self, urls: list[str]):
"""Concurrently scrape multiple URLs"""
semaphore = asyncio.Semaphore(self.max_concurrency)
async def limited_fetch(session, url):
async with semaphore:
result = await self.fetch_page(session, url)
await asyncio.sleep(self.request_delay)
return url, result
async with aiohttp.ClientSession() as session:
tasks = [
limited_fetch(session, url)
for url in urls if url not in self.visited
]
responses = await asyncio.gather(*tasks, return_exceptions=True)
for resp in responses:
if isinstance(resp, Exception):
continue
url, html = resp
if html:
self.visited.add(url)
parsed = self.parse_page(html, url)
if parsed:
self.results.append(parsed)
return self.results
def save_to_json(self, filename: str = "scraped_data.json"):
"""Save results to JSON"""
with open(filename, "w", encoding="utf-8") as f:
json.dump(self.results, f, ensure_ascii=False, indent=2)
print(f"Saved {len(self.results)} items to {filename}")
async def main():
scraper = AsyncScraper(
base_url="https://example.com", max_concurrency=8
)
urls = [f"https://example.com/article/{i}" for i in range(1, 51)]
start = time.perf_counter()
results = await scraper.scrape_urls(urls)
elapsed = time.perf_counter() - start
print(f"Scraping complete: {len(results)} items in {elapsed:.2f}s")
scraper.save_to_json()
asyncio.run(main())
Pattern 5: Data Pipeline and Storage
CSV Export
import csv
from dataclasses import dataclass
@dataclass
class CsvExporter:
filename: str
fieldnames: list[str]
def export(self, data: list[dict]):
with open(self.filename, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=self.fieldnames)
writer.writeheader()
writer.writerows(data)
print(f"Exported {len(data)} items to {self.filename}")
exporter = CsvExporter("products.csv", ["name", "price", "rating", "url"])
exporter.export(products_data)
JSON Export
import json
from datetime import datetime
class JsonExporter:
def __init__(self, filename: str, indent: int = 2):
self.filename = filename
self.indent = indent
def export(self, data: list[dict]):
output = {
"metadata": {
"total": len(data),
"export_time": datetime.now().isoformat(),
"version": "1.0",
},
"data": data,
}
with open(self.filename, "w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=self.indent)
print(f"Exported {len(data)} items to {self.filename}")
SQLite Storage
import sqlite3
from contextlib import contextmanager
class SqliteStorage:
def __init__(self, db_path: str = "scraped.db"):
self.db_path = db_path
self._init_db()
@contextmanager
def get_connection(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def _init_db(self):
with self.get_connection() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS scraped_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
content TEXT,
url TEXT UNIQUE,
tags TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_url ON scraped_data(url)
""")
conn.commit()
def insert(self, item: dict):
with self.get_connection() as conn:
conn.execute(
"""INSERT OR IGNORE INTO scraped_data
(title, content, url, tags) VALUES (?, ?, ?, ?)""",
(
item.get("title"),
item.get("content"),
item.get("url"),
",".join(item.get("tags", [])),
),
)
conn.commit()
def query(self, keyword: str, limit: int = 50) -> list[dict]:
with self.get_connection() as conn:
rows = conn.execute(
"""SELECT * FROM scraped_data
WHERE title LIKE ? OR content LIKE ?
ORDER BY created_at DESC LIMIT ?""",
(f"%{keyword}%", f"%{keyword}%", limit),
).fetchall()
return [dict(row) for row in rows]
```---
## Pitfall Guide
### Pitfall 1: Ignoring robots.txt
❌ Brute-force scraping with no regard for site crawling rules
✅ Check robots.txt first, respect `Disallow` rules; Scrapy enables `ROBOTSTXT_OBEY` by default
### Pitfall 2: Fixed Request Intervals
❌ Using fixed `time.sleep(2)` delays, creating an easily detectable pattern
✅ Use randomized delays `await asyncio.sleep(random.uniform(1.5, 4.0))` to simulate human browsing behavior
### Pitfall 3: Skipping Error Retries
❌ Skipping failed requests entirely, causing significant data loss
✅ Implement exponential backoff retry logic, distinguishing retryable errors (429, 503) from non-retryable ones (404, 403)
### Pitfall 4: Accumulating All Data in Memory
❌ Storing all results in a list, causing OOM with millions of records
✅ Use streaming writes, flushing to file/database every N items and clearing the buffer
### Pitfall 5: Hardcoded Selectors
❌ Hardcoding CSS/XPath selectors that break on every site redesign
✅ Use multiple selector strategies as fallbacks, or extract based on data characteristics (e.g., JSON-LD structured data)
---
## Error Troubleshooting Table
| Error Symptom | Possible Cause | Solution |
|--------------|----------------|----------|
| HTTP 403 Forbidden | Missing headers or identified as bot | Add complete browser headers, use fingerprint masking |
| HTTP 429 Too Many Requests | Request frequency too high | Reduce concurrency, add random delays, use proxy rotation |
| Connection TimeoutError | Unstable network or server rate limiting | Increase timeout, implement retries, switch proxy |
| SSL Verification Failed | Proxy MITM certificate or TLS fingerprint detection | Set `ssl=False` (dev only), or use curl_cffi |
| Empty Page Content | JS dynamic rendering incomplete | Switch to Playwright, wait for element loading |
| CAPTCHA Interception | Anti-bot rules triggered | Reduce request frequency, use CAPTCHA solving service, use cookie pool |
| Duplicate Data Scraping | Missing URL deduplication logic | Use Bloom Filter or Redis Set for dedup |
| Memory Overflow OOM | Large data accumulation in memory | Stream to disk, batch processing, use generators |
| Encoding Error UnicodeDecodeError | Response encoding mismatch | Use `response.encoding = response.apparent_encoding` |
| ElementNotInteractable | Element obscured or not yet interactive | Wait for element visibility `wait_for_selector`, scroll to element |
---
## Advanced Optimization Tips
### 1. Bloom Filter Deduplication
```python
from pybloom_live import ScalableBloomFilter
class UrlDeduplicator:
def __init__(self):
self.bloom = ScalableBloomFilter(
initial_capacity=100000, error_rate=0.001,
)
def is_duplicate(self, url: str) -> bool:
if url in self.bloom:
return True
self.bloom.add(url)
return False
2. Smart Rate Limiting with AutoThrottle
# Scrapy settings
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 1.0
AUTOTHROTTLE_MAX_DELAY = 30.0
AUTOTHROTTLE_TARGET_CONCURRENCY = 8.0
AUTOTHROTTLE_DEBUG = True
3. Distributed Scraping (Scrapy + Redis)
pip install scrapy-redis
# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://localhost:6379/0"
SCHEDULER_PERSIST = True
4. curl_cffi for TLS Fingerprint Bypass
from curl_cffi import requests as curl_requests
response = curl_requests.get(
"https://tls-protected.example.com",
impersonate="chrome126",
timeout=15,
)
print(response.status_code)
5. Cookie Pool Management
import json
import random
from pathlib import Path
class CookiePool:
def __init__(self, cookie_file: str = "cookies.json"):
self.cookie_file = Path(cookie_file)
self.cookies: list[dict] = []
self._load()
def _load(self):
if self.cookie_file.exists():
self.cookies = json.loads(
self.cookie_file.read_text(encoding="utf-8")
)
def _save(self):
self.cookie_file.write_text(
json.dumps(self.cookies, ensure_ascii=False, indent=2),
encoding="utf-8",
)
def add_cookie(self, cookie: dict):
self.cookies.append(cookie)
self._save()
def get_random_cookie(self) -> dict | None:
if not self.cookies:
return None
return random.choice(self.cookies)
def remove_cookie(self, cookie: dict):
self.cookies = [c for c in self.cookies if c != cookie]
self._save()
Framework Comparison
| Dimension | Scrapy | Playwright | Selenium | BeautifulSoup + requests |
|---|---|---|---|---|
| Best For | Large-scale structured scraping | Dynamic SPA pages | Legacy system compatibility | Small-scale static pages |
| JS Rendering | Requires scrapy-playwright | Native support | Native support | Not supported |
| Performance | High (async + Twisted) | Medium (browser instances) | Low (WebDriver overhead) | High (lightweight HTTP) |
| Anti-Bot Capability | Medium (needs extra config) | High (real browser) | Medium (easily detected) | Low (bare requests) |
| Learning Curve | Steep | Moderate | Easy | Easy |
| Distributed | Native (scrapy-redis) | DIY | DIY | DIY |
| Data Pipeline | Built-in Pipeline | DIY | DIY | DIY |
| Debugging Tools | Scrapy Shell | Playwright Inspector | Selenium IDE | Browser DevTools |
| Resource Usage | Low | High (browser process) | High (WebDriver) | Very Low |
| Recommendation | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐ |
Selection Guide:
- Large-scale structured data collection → Scrapy
- Dynamic SPA / interactive operations → Playwright
- Quick scraping of simple static pages → BeautifulSoup + requests
- Existing Selenium test infrastructure → Selenium (but consider migrating to Playwright)
Recommended Tools
These ToolsKu tools can help in your web scraping practice:
- JSON Formatter — Format API responses and scraped results to quickly spot data structure issues
- Hash Calculator — Generate content hashes for deduplication and URL fingerprinting
- cURL to Code — Convert browser cURL requests to Python code in one click, quickly building request templates
Web scraping is fundamentally a "game of cat and mouse with anti-bot systems." There is no silver bullet — only by choosing the right tech stack based on target site characteristics and balancing efficiency with stealth can you succeed. Respecting robots.txt, controlling request frequency, and honoring data copyrights are the baseline for every scraping developer.
Try these browser-local tools — no sign-up required →
#Python爬虫#反爬虫#Scrapy#Playwright爬虫#2026#编程语言