web-scraping

What this skill does

# Web Scraping for RAG

## Tool Comparison

| Tool | Type | JS Rendering | Main-Content | Scale | Best For |
|------|------|--------------|--------------|-------|----------|
| Firecrawl | API (paid) | Yes | Excellent (LLM) | Medium | Quick RAG ingestion, markdown out |
| Crawl4AI | OSS library | Yes (Playwright) | Very good | Medium | Self-hosted, LLM-friendly |
| trafilatura | OSS library | No | Excellent | High | Article/news extraction |
| BeautifulSoup + readability-lxml | OSS | No | Good | High | Custom HTML parsing |
| Scrapy | OSS framework | No (+ splash) | Manual | Very high | Multi-million page crawls |
| Playwright | Browser automation | Yes | Manual | Low-medium | SPA / logged-in scraping |

## Firecrawl API

```python
import os
from firecrawl import FirecrawlApp

app = FirecrawlApp(api_key=os.environ["FIRECRAWL_API_KEY"])

# Single page -> markdown
result = app.scrape_url(
    "https://docs.anthropic.com/claude/docs",
    params={
        "formats": ["markdown", "links", "html"],
        "onlyMainContent": True,
        "waitFor": 1500,
        "timeout": 30000,
    },
)
print(result["markdown"][:500])

# Full site crawl
job = app.crawl_url(
    "https://docs.anthropic.com",
    params={
        "limit": 500,
        "scrapeOptions": {"formats": ["markdown"], "onlyMainContent": True},
        "excludePaths": ["/blog/*", "/legal/*"],
        "maxDepth": 4,
    },
    wait_until_done=True,
)
for page in job["data"]:
    save_chunk(page["metadata"]["sourceURL"], page["markdown"])
```

## Crawl4AI — Open Source, LLM-Friendly

```python
import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig
from crawl4ai.extraction_strategy import LLMExtractionStrategy

async def crawl():
    browser = BrowserConfig(headless=True, viewport_width=1280)
    run_cfg = CrawlerRunConfig(
        word_count_threshold=50,
        exclude_external_links=True,
        remove_overlay_elements=True,
        wait_for="css:article",
        cache_mode="bypass",
    )
    async with AsyncWebCrawler(config=browser) as crawler:
        result = await crawler.arun(
            url="https://example.com/article",
            config=run_cfg,
        )
        return result.markdown.fit_markdown, result.links["internal"]

md, links = asyncio.run(crawl())
```

## trafilatura — Main-Content Extraction

```python
import trafilatura
from trafilatura.settings import use_config

cfg = use_config()
cfg.set("DEFAULT", "EXTRACTION_TIMEOUT", "30")

downloaded = trafilatura.fetch_url("https://news.ycombinator.com/item?id=1")
result = trafilatura.extract(
    downloaded,
    output_format="markdown",
    include_tables=True,
    include_links=True,
    include_comments=False,
    with_metadata=True,
    favor_precision=True,
    config=cfg,
)
print(result)

# Sitemap-driven ingestion
from trafilatura.sitemaps import sitemap_search
urls = sitemap_search("https://example.com")
for url in urls[:1000]:
    html = trafilatura.fetch_url(url)
    text = trafilatura.extract(html, output_format="markdown")
    if text:
        save_chunk(url, text)
```

## BeautifulSoup + readability-lxml

```python
import requests
from bs4 import BeautifulSoup
from readability import Document  # readability-lxml

html = requests.get("https://example.com/post", timeout=15).text

doc = Document(html)
title = doc.title()
cleaned_html = doc.summary(html_partial=True)

soup = BeautifulSoup(cleaned_html, "lxml")
for tag in soup(["script", "style", "nav", "footer", "aside"]):
    tag.decompose()
text = soup.get_text("\n", strip=True)
```

## Scrapy — Large-Scale Crawl

```python
# spider.py
import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule

class DocsSpider(CrawlSpider):
    name = "docs"
    allowed_domains = ["example.com"]
    start_urls = ["https://example.com/sitemap.xml"]

    custom_settings = {
        "ROBOTSTXT_OBEY": True,
        "CONCURRENT_REQUESTS": 16,
        "DOWNLOAD_DELAY": 0.5,
        "AUTOTHROTTLE_ENABLED": True,
        "AUTOTHROTTLE_TARGET_CONCURRENCY": 4.0,
        "USER_AGENT": "RAGBot/1.0 (+https://myorg.com/bot)",
        "HTTPCACHE_ENABLED": True,
        "DEPTH_LIMIT": 5,
    }

    rules = (
        Rule(LinkExtractor(allow=r"/docs/"), callback="parse_page", follow=True),
    )

    def parse_page(self, response):
        yield {
            "url": response.url,
            "title": response.css("h1::text").get(),
            "text": " ".join(response.css("article ::text").getall()),
        }
```

Run: `scrapy runspider spider.py -O out.jsonl`.

## Playwright — JS-Rendered Pages

```python
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    ctx = browser.new_context(
        user_agent="RAGBot/1.0",
        viewport={"width": 1280, "height": 900},
    )
    page = ctx.new_page()
    page.goto("https://spa.example.com", wait_until="networkidle")
    page.wait_for_selector("article")

    # Auto-scroll for lazy-loaded content
    page.evaluate("""
      async () => {
        for (let y = 0; y < document.body.scrollHeight; y += 500) {
          window.scrollTo(0, y);
          await new Promise(r => setTimeout(r, 200));
        }
      }
    """)

    html = page.content()
    browser.close()
```

## Sitemap.xml Discovery

```python
import requests
import xml.etree.ElementTree as ET

def discover_urls(sitemap_url: str) -> list[str]:
    resp = requests.get(sitemap_url, timeout=15)
    resp.raise_for_status()
    ns = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}
    root = ET.fromstring(resp.content)
    # Sitemap index vs urlset
    if root.tag.endswith("sitemapindex"):
        urls = []
        for sm in root.findall("sm:sitemap", ns):
            loc = sm.find("sm:loc", ns).text
            urls.extend(discover_urls(loc))
        return urls
    return [u.find("sm:loc", ns).text for u in root.findall("sm:url", ns)]
```

## robots.txt Respect

```python
from urllib.robotparser import RobotFileParser

rp = RobotFileParser()
rp.set_url("https://example.com/robots.txt")
rp.read()

if not rp.can_fetch("RAGBot/1.0", "https://example.com/private/"):
    raise PermissionError("Disallowed by robots.txt")

crawl_delay = rp.crawl_delay("RAGBot/1.0") or 1.0
```

## Anti-Bot Handling

| Challenge | Mitigation |
|---|---|
| Cloudflare / hCaptcha | Use Playwright with `playwright-stealth`, or a paid unblocker (Bright Data, ZenRows) |
| Rate limiting (429) | Exponential backoff + `Retry-After` header |
| IP blocks | Rotating residential proxies; lower `CONCURRENT_REQUESTS_PER_DOMAIN` |
| UA blocklists | Set a real-looking UA and `Accept-Language` |
| JS fingerprinting | `playwright-stealth` patches, `crawl4ai` stealth mode |

```python
# Polite backoff
import time, random, requests

def fetch(url, max_tries=5):
    for attempt in range(max_tries):
        r = requests.get(url, timeout=20, headers={"User-Agent": "RAGBot/1.0"})
        if r.status_code == 200:
            return r.text
        if r.status_code == 429:
            delay = int(r.headers.get("Retry-After", 2 ** attempt))
            time.sleep(delay + random.random())
            continue
        r.raise_for_status()
    raise RuntimeError(f"failed: {url}")
```

## Content Deduplication

```python
import hashlib
from datasketch import MinHash, MinHashLSH

def content_hash(text: str) -> str:
    norm = " ".join(text.split()).lower()
    return hashlib.sha256(norm.encode()).hexdigest()

def minhash(text: str, num_perm=128) -> MinHash:
    m = MinHash(num_perm=num_perm)
    for shingle in {text[i:i+5] for i in range(len(text) - 4)}:
        m.update(shingle.encode())
    return m

lsh = MinHashLSH(threshold=0.85, num_perm=128)
seen_hashes: set[str] = set()

def is_duplicate(url: str, text: str) -> bool:
    h = content_hash(text)
    if h in seen_hashes:
        return True
    mh = minhash(text)
    if lsh.query(mh):
        return True
    seen_hashes.add(h)
    lsh.insert(u