FerroxLabs Web Scraper
Core Philosophy
Web scraping is the programmatic extraction of data from websites. It sits at the intersection of engineering, ethics, and law. A responsible scraper respects rate limits, follows robots.txt, identifies itself honestly, and minimizes impact on the target server. The best scraper is one that never gets blocked because it behaves like a respectful automated visitor.
HTTP Client Selection
Decision Tree
Do you need to render JavaScript?
YES -> Browser automation (Playwright, Puppeteer)
NO -> Is the site API-based (XHR/Get requests)?
YES -> HTTP client + reverse-engineer API calls
NO -> HTTP client + HTML parser (BeautifulSoup, Cheerio)
Python: httpx (Recommended)
import httpx
from selectolax.parser import HTMLParser
# Async HTTP client with connection pooling
async def scrape_product_pages(urls: list[str]) -> list[dict]:
results = []
async with httpx.AsyncClient(
timeout=30.0,
follow_redirects=True,
headers={
"User-Agent": "MyBot/1.0 (+[reference URL])",
"Accept": "text/html",
"Accept-Language": "en-US,en;q=0.9",
},
limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
) as client:
for url in urls:
response = await client.get(url)
response.raise_for_status()
data = parse_product_page(response.text)
results.append(data)
await asyncio.sleep(1) # Rate limiting
return results
def parse_product_page(html: str) -> dict:
tree = HTMLParser(html)
return {
"title": tree.css_first("h1.product-title").text(strip=True),
"price": tree.css_first("[data-testid='price']").text(strip=True),
"description": tree.css_first(".product-description").text(strip=True),
"images": [img.attrs["src"] for img in tree.css("img.product-image")],
}
Node.js: undici/get + Cheerio
import * as cheerio from 'cheerio';
async function scrapeProductPage(url) {
const response = await get(url, {
headers: {
'User-Agent': 'MyBot/1.0 (+[reference URL])',
'Accept': 'text/html',
},
});
if (!response.ok) throw new Error(`HTTP ${response.status}: ${url}`);
const html = await response.text();
const $ = cheerio.load(html);
return {
title: $('h1.product-title').text().trim(),
price: $('[data-testid="price"]').text().trim(),
description: $('.product-description').text().trim(),
images: $('img.product-image').map((_, el) => $(el).attr('src')).get(),
specs: $('table.specs tr').map((_, row) => ({
key: $(row).find('th').text().trim(),
value: $(row).find('td').text().trim(),
})).get(),
};
}
HTML Parsing
BeautifulSoup (Python)
from bs4 import BeautifulSoup
def parse_listing_page(html: str) -> list[dict]:
soup = BeautifulSoup(html, 'lxml') # lxml is fastest
products = []
for card in soup.select('.product-card'):
product = {
'name': card.select_one('.product-name').get_text(strip=True),
'price': card.select_one('.price').get_text(strip=True),
'url': card.select_one('a.product-link')['href'],
'rating': float(card.select_one('[data-rating]').get('data-rating', 0)),
'in_stock': 'out-of-stock' not in card.get('class', []),
}
# Handle optional elements
review_el = card.select_one('.review-count')
product['reviews'] = int(review_el.get_text(strip=True).replace(',', '')) if review_el else 0
products.append(product)
return products
def extract_structured_data(html: str) -> dict:
"""Extract JSON-LD structured data from page."""
soup = BeautifulSoup(html, 'lxml')
scripts = soup.select('script[type="application/ld+json"]')
for script in scripts:
data = json.loads(script.string)
if data.get('@type') == 'Product':
return data
return {}
Cheerio (JavaScript)
import * as cheerio from 'cheerio';
function parseListingPage(html) {
const $ = cheerio.load(html);
return $('.product-card').map((_, card) => {
const $card = $(card);
return {
name: $card.find('.product-name').text().trim(),
price: $card.find('.price').text().trim(),
url: $card.find('a.product-link').attr('href'),
rating: parseFloat($card.find('[data-rating]').attr('data-rating') || '0'),
inStock: !$card.hasClass('out-of-stock'),
reviews: parseInt($card.find('.review-count').text().replace(/,/g, '') || '0'),
};
}).get();
}
Browser Automation for Dynamic Sites
Playwright (Recommended)
from playwright.async_api import async_playwright
async def scrape_spa(url: str) -> dict:
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context(
user_agent='MyBot/1.0 (+[reference URL])',
viewport={'width': 1920, 'height': 1080},
)
page = await context.new_page()
# Block unnecessary resources for speed
await page.route('**/*.{png,jpg,jpeg,gif,svg,ico,woff,woff2}',
lambda route: route.abort())
await page.route('**/analytics*', lambda route: route.abort())
await page.route('**/ads*', lambda route: route.abort())
await page.goto(url, wait_until='networkidle')
# Wait for dynamic content
await page.wait_for_selector('.product-grid', state='visible')
# Interact: load all items via infinite scroll
previous_count = 0
while True:
await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
await page.wait_for_timeout(1000)
current_count = await page.locator('.product-card').count()
if current_count == previous_count:
break
previous_count = current_count
# Extract data after all content is loaded
products = await page.evaluate('''() => {
return Array.from(document.querySelectorAll('.product-card')).map(card => ({
name: card.querySelector('.product-name')?.textContent?.trim(),
price: card.querySelector('.price')?.textContent?.trim(),
url: card.querySelector('a')?.href,
}));
}''')
await browser.close()
return products
Intercepting API Calls
async def scrape_via_api_interception(url: str) -> list[dict]:
"""Instead of parsing HTML, intercept the XHR/get API calls."""
collected_data = []
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
# Intercept API responses
async def handle_response(response):
if '/api/products' in response.url and response.status == 200:
data = await response.json()
collected_data.extend(data.get('items', []))
page.on('response', handle_response)
await page.goto(url, wait_until='networkidle')
# Trigger pagination to collect all data
while True:
next_button = page.locator('button[aria-label="Next page"]')
if not await next_button.is_visible() or not await next_button.is_enabled():
break
await next_button.click()
await page.wait_for_timeout(1000)
await browser.close()
return collected_data
Rate Limiting
import asyncio
import time
from collections import deque
class RateLimiter:
"""Sliding window rate limiter."""
def __init__(self, max_requests: int, time_window: float):
self.max_requests = max_requests
self.time_window = time_window
self.timestamps = deque()
async def acquire(self):
now = time.monotonic()
# Remove timestamps outside the window
while self.timestamps and (now - self.timestamps[0]) > self.time_window:
self.timestamps.popleft()
if len(self.timestamps) >= self.max_requests:
sleep_time = self.time_window - (now - self.timestamps[0])
if sleep_time > 0:
await asyncio.sleep(sleep_time)
self.timestamps.append(time.monotonic())
# Usage
limiter = RateLimiter(max_requests=10, time_window=60) # 10 req/min
async def fetch_with_rate_limit(client, url):
await limiter.acquire()
return await client.get(url)
Robots.txt Compliance
from urllib.robotparser import RobotFileParser
class RobotsTxtChecker:
def __init__(self, base_url: str, user_agent: str = 'MyBot'):
self.parser = RobotFileParser()
self.parser.set_url(f'{base_url}/robots.txt')
self.parser.read()
self.user_agent = user_agent
def can_fetch(self, url: str) -> bool:
return self.parser.can_fetch(self.user_agent, url)
def crawl_delay(self) -> float | None:
delay = self.parser.crawl_delay(self.user_agent)
return delay
# Usage
robots = RobotsTxtChecker('[reference URL]', 'MyBot')
if robots.can_fetch('[reference URL]'):
delay = robots.crawl_delay() or 1.0
# Proceed with delay
Session and Cookie Handling
async def scrape_with_session(login_url: str, target_url: str):
async with httpx.AsyncClient(follow_redirects=True) as client:
# Login to get session cookie
login_response = await client.post(login_url, data={
'username': 'user',
'password': 'pass',
'csrf_token': await get_csrf_token(client, login_url),
})
# Cookies are automatically stored in the client session
# Access protected pages
response = await client.get(target_url)
return response.text
async def get_csrf_token(client, login_page_url):
response = await client.get(login_page_url)
soup = BeautifulSoup(response.text, 'lxml')
return soup.select_one('input[name="csrf_token"]')['value']
Anti-Detection Techniques
# Rotate User-Agents
import random
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 Safari/605.1.15',
'Mozilla/5.0 (X11; Linux x86_64; rv:121.0) Gecko/20100101 Firefox/121.0',
]
headers = {'User-Agent': random.choice(USER_AGENTS)}
# Random delays between requests
await asyncio.sleep(random.uniform(1.0, 3.0))
# Playwright stealth mode
from playwright_stealth import stealth_async
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
await stealth_async(page) # Apply stealth patches
Pagination Handling
async def scrape_all_pages(base_url: str) -> list[dict]:
all_items = []
page_num = 1
async with httpx.AsyncClient() as client:
while True:
response = await client.get(f"{base_url}?page={page_num}")
items = parse_listing_page(response.text)
if not items:
break
all_items.extend(items)
# Check for next page
soup = BeautifulSoup(response.text, 'lxml')
next_link = soup.select_one('a[rel="next"]')
if not next_link:
break
page_num += 1
await asyncio.sleep(1)
return all_items
# Cursor-based pagination (API)
async def scrape_api_cursor(api_url: str) -> list[dict]:
all_items = []
cursor = None
async with httpx.AsyncClient() as client:
while True:
params = {'limit': 100}
if cursor:
params['cursor'] = cursor
response = await client.get(api_url, params=params)
data = response.json()
all_items.extend(data['items'])
cursor = data.get('next_cursor')
if not cursor:
break
await asyncio.sleep(0.5)
return all_items
Data Extraction Patterns
CSS Selector Patterns
# Common extraction patterns
def extract_data(soup):
return {
# Text content
'title': soup.select_one('h1').get_text(strip=True),
# Attribute value
'image': soup.select_one('img.main')['src'],
# Data attribute
'product_id': soup.select_one('[data-product-id]').get('data-product-id'),
# Nested text with separator
'breadcrumbs': ' > '.join(
a.get_text(strip=True) for a in soup.select('.breadcrumb a')
),
# Table extraction
'specs': {
row.select_one('th').get_text(strip=True): row.select_one('td').get_text(strip=True)
for row in soup.select('table.specs tr')
if row.select_one('th') and row.select_one('td')
},
# List extraction
'features': [
li.get_text(strip=True) for li in soup.select('ul.features li')
],
# Price parsing
'price': parse_price(soup.select_one('.price').get_text()),
}
def parse_price(text: str) -> float | None:
"""Extract numeric price from text like '$1,234.56' or 'EUR 99.99'."""
import re
match = re.search(r'[\d,]+\.?\d*', text.replace(',', ''))
return float(match.group()) if match else None
Legal Considerations
ALWAYS:
- Check and respect robots.txt
- Include identifying User-Agent with contact info
- Respect rate limits and Crawl-delay directives
- Check Terms of Service before scraping
- Only scrape publicly available data
- Store and use data in compliance with privacy laws (GDPR, CCPA)
NEVER:
- Circumvent authentication without permission
- Scrape personal data without legal basis
- Overload servers with aggressive request rates
- Ignore cease-and-desist requests
- Scrape behind paywalls you haven't paid for
- Claim scraped data as your own copyrighted work
CONSIDER:
- Is there a public API? (Always prefer official APIs)
- Does the site offer data exports or feeds?
- Can you partner with the data provider?
- Is the data facts (generally OK) or creative expression (copyright)?
Error Handling and Resilience
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=30),
retry=retry_if_exception_type((httpx.TimeoutException, httpx.HTTPStatusError)),
)
async def fetch_with_retry(client: httpx.AsyncClient, url: str) -> str:
response = await client.get(url)
if response.status_code == 429: # Too Many Requests
retry_after = int(response.headers.get('Retry-After', 60))
await asyncio.sleep(retry_after)
raise httpx.HTTPStatusError("Rate limited", request=response.request, response=response)
response.raise_for_status()
return response.text
When to Use
Use this skill when:
- Designing or implementing web scraper solutions
- Reviewing or improving existing web scraper approaches
- Making architectural or implementation decisions about web scraper
- Learning web scraper patterns and best practices
- Troubleshooting web scraper-related issues
Do NOT use this skill when:
- The question is about a fundamentally different technology domain
- A more specific sibling skill covers the exact topic needed
- The user needs a complete hands-on tutorial rather than expert guidance
Output Format
# Web Scraper Analysis
## Context Assessment
[Situation summary and constraints]
## Recommended Approach
[Primary recommendation with rationale]
## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]
## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]
## Next Steps
- [Immediate action item]
- [Follow-up action item]
Example
Input: "Help me implement web scraper for a medium-scale production application"
Output: A structured analysis covering current state assessment, recommended web scraper approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.
Edge Cases
- Legacy system integration: When web scraper must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
- Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
- Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
- Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities