Web Crawler

Web Crawler Skill Guide

When to Use

• User provides a URL and wants a Python crawler/scraper script.
• Scraping Vue/React/SPA/AJAX-driven sites, dynamic pages, JSON APIs.
• Handling JavaScript-rendered content, pagination, login, infinite scroll.
• Reverse engineering encrypted parameters, signatures, tokens.
• Bypassing anti-bot systems (Cloudflare, Akamai, reCAPTCHA, etc.).
• Need complete, runnable, production-grade Python script.

---

Phase 1: Reconnaissance (站点侦察)

Before writing any code, analyze the target:

1.1 Network Analysis

• Open DevTools → Network tab → filter by Fetch/XHR
• Identify data API endpoints (often return JSON)
• Check request methods: GET / POST / PUT
• Inspect query params, body, headers, cookies
• Look for: api., /api/, /graphql, .json, ajax, json patterns

1.2 Page Type Detection

1.3 Anti-Bot Detection

• Check for Cloudflare (cf-ray header, challenge page)
• Akamai BMP (_abck cookie)
• reCAPTCHA / hCaptcha / Turnstile
• Device fingerprinting (canvas, webgl, navigator properties)
• Behavioral analysis (mouse movement, typing patterns)
• Use https://bot.sannysoft.com to test headless detection

---

Phase 2: Library Selection (技术选型)

Decision Tree

Is there a clean JSON API?
├── Yes → requests / httpx (fastest, simplest)
└── No → Is content server-rendered?
    ├── Yes → requests + BeautifulSoup/lxml
    └── No → Need JS rendering?
        ├── Light JS → requests-html (pyppeteer backend)
        ├── Heavy JS / SPA → playwright (recommended) or selenium
        └── Need to bypass detection → DrissionPage / undetected-chromedriver

Library Comparison

---

Phase 3: Core Implementation Patterns

3.1 Static/API Crawling (requests)

import requests
import pandas as pd
import time
import random

# Config
BASE_URL = "https://api.example.com/data"
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept": "application/json",
    "Referer": "https://example.com/",
}
COOKIES = {"session": "xxx"}
PROXY = {"http": "http://proxy:8080", "https": "http://proxy:8080"}

def fetch_page(page=1):
    params = {"page": page, "size": 20}
    for attempt in range(3):
        try:
            resp = requests.get(BASE_URL, headers=HEADERS, params=params,
                                cookies=COOKIES, proxies=PROXY, timeout=15)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as e:
            print(f"Attempt {attempt+1} failed: {e}")
            time.sleep(2 ** attempt)
    return None

def crawl_all(max_pages=100):
    all_data = []
    for page in range(1, max_pages + 1):
        data = fetch_page(page)
        if not data or not data.get("items"):
            break
        all_data.extend(data["items"])
        print(f"Page {page}: {len(data['items'])} items")
        time.sleep(random.uniform(1, 3))  # Polite delay
    return all_data

if __name__ == "__main__":
    results = crawl_all()
    pd.DataFrame(results).to_csv("output.csv", index=False, encoding="utf-8-sig")
    print(f"Saved {len(results)} records to output.csv")

3.2 Dynamic SPA Crawling (Playwright)

from playwright.sync_api import sync_playwright
import pandas as pd
import time

def crawl_spa():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            viewport={"width": 1920, "height": 1080},
            locale="zh-CN",
        )
        page = context.new_page()

        # Intercept API responses (preferred over parsing DOM)
        api_data = []
        def handle_response(response):
            if "/api/list" in response.url and response.status == 200:
                try:
                    api_data.append(response.json())
                except:
                    pass
        page.on("response", handle_response)

        page.goto("https://example.com/list", wait_until="networkidle")

        # Infinite scroll
        last_height = page.evaluate("document.body.scrollHeight")
        while True:
            page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            page.wait_for_timeout(2000)
            new_height = page.evaluate("document.body.scrollHeight")
            if new_height == last_height:
                break
            last_height = new_height

        browser.close()
        return api_data

if __name__ == "__main__":
    data = crawl_spa()
    pd.DataFrame(data).to_excel("output.xlsx", index=False)

3.3 Anti-Detection Crawling (DrissionPage)

from DrissionPage import ChromiumPage, ChromiumOptions
import time

def crawl_stealth():
    co = ChromiumOptions()
    co.set_argument("--no-sandbox")
    co.set_argument("--disable-blink-features=AutomationControlled")
    co.set_user_agent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36")

    page = ChromiumPage(co)
    page.get("https://example.com")

    # DrissionPage bypasses many fingerprint checks automatically
    items = page.eles('css:.item-class')
    results = []
    for item in items:
        results.append({
            "title": item.ele('css:.title').text,
            "link": item.ele('css:a').attr('href'),
            "price": item.ele('css:.price').text,
        })

    page.quit()
    return results

3.4 Scrapy Framework (Large-Scale)

# spider.py
import scrapy
from scrapy.crawler import CrawlerProcess

class ExampleSpider(scrapy.Spider):
    name = "example"
    start_urls = ["https://example.com/page/1"]

    def parse(self, response):
        for item in response.css(".item"):
            yield {
                "title": item.css(".title::text").get(),
                "url": item.css("a::attr(href)").get(),
            }
        next_page = response.css(".next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, self.parse)

# Run
process = CrawlerProcess(settings={
    "FEEDS": {"output.json": {"format": "json"}},
    "USER_AGENT": "Mozilla/5.0",
    "DOWNLOAD_DELAY": 2,
    "AUTOTHROTTLE_ENABLED": True,
})
process.crawl(ExampleSpider)
process.start()

---

Phase 4: Advanced Techniques (高级技术)

4.1 JS Reverse Engineering (JS逆向)

When parameters are encrypted/signed:

1. Locate the encryption function: Search source for param name, set breakpoints
2. Common patterns:
   • sign: md5(timestamp + secret + params)
   • token: base64(aes_encrypt(data, key))
   • Custom webpack modules
3. Tools: Chrome DevTools, Fiddler, Charles, mitmproxy, PyExecJS, Node.js subprocess
4. Approaches:
   • Replay: Extract algorithm and reimplement in Python
   • ExecJS: Run obfuscated JS directly via execjs or node subprocess
   • Hook injection: Override functions to dump intermediate values

# Example: calling JS encryption from Python
import execjs

ctx = execjs.compile("""
function sign(params, timestamp, secret) {
    // ... extracted from target site
    return CryptoJS.MD5(timestamp + secret + JSON.stringify(params)).toString();
}
""")
signature = ctx.call("sign", {"page": 1}, "1700000000", "secret_key")

4.2 Login & Session Management

import requests

session = requests.Session()
# Login
login_resp = session.post("https://example.com/api/login",
                           json={"username": "user", "password": "pwd"})
# Session maintains cookies automatically
data = session.get("https://example.com/api/protected").json()

4.3 Pagination Strategies

4.4 Proxy & Fingerprint Rotation

import random
import requests
from fake_useragent import UserAgent

ua = UserAgent()
PROXIES = [
    "http://user:pass@proxy1:8080",
    "http://user:pass@proxy2:8080",
]

def fetch(url):
    headers = {"User-Agent": ua.random}
    proxy = {"http": random.choice(PROXIES), "https": random.choice(PROXIES)}
    return requests.get(url, headers=headers, proxies=proxy, timeout=10)

4.5 Captcha Handling

---

Phase 5: Data Storage (数据存储)

5.1 Storage Options

5.2 Image Download Pattern

import os
import requests
from pathlib import Path

def download_images(urls, folder="images"):
    Path(folder).mkdir(exist_ok=True)
    for i, url in enumerate(urls):
        try:
            resp = requests.get(url, timeout=10)
            ext = url.split(".")[-1][:4]  # crude extension detection
            filename = f"{folder}/img_{i:04d}.{ext}"
            with open(filename, "wb") as f:
                f.write(resp.content)
        except Exception as e:
            print(f"Failed {url}: {e}")

5.3 Database Pattern

import sqlite3

conn = sqlite3.connect("data.db")
conn.execute("""CREATE TABLE IF NOT EXISTS items
                (id INTEGER PRIMARY KEY, title TEXT, url TEXT, price REAL)""")

def save(item):
    conn.execute("INSERT OR IGNORE INTO items (title, url, price) VALUES (?,?,?)",
                 (item["title"], item["url"], item.get("price")))
    conn.commit()

---

Phase 6: Distributed & Scaled Crawling (分布式)

6.1 Scrapy-Redis (Distributed Scrapy)

# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
SCHEDULER_PERSIST = True
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://localhost:6379/0"

6.2 Custom Redis Queue

import redis
import json

r = redis.Redis()
QUEUE = "crawl:urls"

def push_urls(urls):
    for url in urls:
        r.lpush(QUEUE, json.dumps({"url": url, "retry": 0}))

def pop_url():
    return json.loads(r.brpop(QUEUE)[1])

6.3 Async Crawling (asyncio + aiohttp)

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def crawl(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

results = asyncio.run(crawl(url_list))

---

Phase 7: Anti-Detection Best Practices (反反爬)

7.1 Request Level

• Rotate User-Agent (fake-useragent library)
• Random delays: time.sleep(random.uniform(1, 5))
• Respect Retry-After header
• Use residential proxies (datacenter IPs get blocked)
• Rotate cookies/sessions per batch

7.2 Browser Level

• undetected-chromedriver or DrissionPage for fingerprint spoofing
• Set realistic viewport, locale, timezone
• Disable webdriver flag: navigator.webdriver = undefined
• Mock canvas/webgl fingerprints
• Simulate human behavior: mouse movement, scroll, random clicks

7.3 Behavioral

• Don't crawl too fast — match human browsing speed
• Visit unrelated pages occasionally (decoy requests)
• Use realistic referer chains
• Handle 403/429 gracefully with exponential backoff

---

Phase 8: Compliance & Ethics (合规与道德)

8.1 Legal Considerations

• Always check robots.txt — respect disallow rules
• Review Terms of Service — some prohibit scraping
• Personal data: comply with GDPR / PIPL (China's 个人信息保护法)
• Copyright: don't republish copyrighted content
• Rate limits: don't overload target servers

8.2 Ethical Guidelines

• Identify your crawler in UA when appropriate
• Crawl during off-peak hours
• Cache responses to avoid redundant requests
• Only collect what you need
• Consider asking for API access first

---

Phase 9: Interactive Workflow (交互流程)

When user requests a crawler, follow this dialogue:

Step 1: Clarify Requirements

Ask the user (only if not provided):

1. Target URL — specific page or entire site?
2. Data fields — what to extract? (title, price, images, full text?)
3. Scope — how many pages/items?
4. Login required? — if yes, credentials or existing cookies?
5. Output format — CSV / XLSX / JSON / DB / images?
6. Anti-bot observed? — Cloudflare, captcha, etc.?
7. Frequency — one-time or scheduled/recurring?
8. Environment — local machine, server, cloud?

Step 2: Analyze & Recommend

• Based on URL, recommend the best approach (API vs. browser)
• List required libraries with pip install commands
• Warn about potential anti-bot challenges
• Estimate complexity and time

Step 3: Generate Script

• Complete, runnable Python script
• Configuration section at top (URL, headers, output path)
• Modular functions: fetch(), parse(), save(), main()
• Error handling with retries
• Progress logging
• Comments explaining non-obvious logic
• Type hints for clarity

Step 4: Usage Instructions

• pip install commands for all dependencies
• How to run: python crawler.py
• Configuration changes needed
• Output location and format
• Troubleshooting common issues

Step 5: Offer Enhancements

• Add proxy support
• Add scheduling (cron / APScheduler)
• Add monitoring/alerting
• Convert to Scrapy project for scale
• Add data cleaning/processing pipeline

---

Phase 10: Output Template (输出模板)

Every generated script should follow this structure:

"""
Crawler: [Site Name]
Description: [What it crawls]
Author: Generated by Aura
Date: [auto]
Dependencies: pip install requests beautifulsoup4 pandas
"""

import os
import sys
import time
import random
import logging
from pathlib import Path

# ===== Configuration =====
TARGET_URL = "https://example.com"
OUTPUT_DIR = Path("./output")
OUTPUT_FORMAT = "csv"  # csv / xlsx / json / sqlite
MAX_PAGES = 100
DELAY_RANGE = (1, 3)  # random delay between requests
TIMEOUT = 15
MAX_RETRIES = 3

HEADERS = {
    "User-Agent": "Mozilla/5.0 ...",
    "Accept": "text/html,application/xhtml+xml,...",
}

# ===== Logging =====
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(OUTPUT_DIR / "crawler.log", encoding="utf-8"),
        logging.StreamHandler(),
    ]
)
logger = logging.getLogger(__name__)

# ===== Core Functions =====
def fetch(url, **kwargs):
    """Fetch a single URL with retry logic."""
    ...

def parse(html):
    """Parse HTML/JSON and extract target data."""
    ...

def save(data):
    """Save extracted data to configured output."""
    ...

def main():
    """Main entry point."""
    OUTPUT_DIR.mkdir(exist_ok=True)
    # ... crawl logic
    logger.info(f"Crawling complete. {len(results)} items saved to {OUTPUT_DIR}")

if __name__ == "__main__":
    main()

---

Quick Reference (速查表)

Common Selectors

Common Headers for Anti-Detection

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Accept-Encoding": "gzip, deflate, br",
    "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",
}

Installation Commands

# Basic
pip install requests beautifulsoup4 lxml pandas openpyxl

# Dynamic
pip install playwright
playwright install chromium

# Anti-detection
pip install DrissionPage undetected-chromedriver fake-useragent

# Framework
pip install scrapy scrapy-redis

# Async
pip install aiohttp httpx

# JS execution
pip install PyExecJS
# Node.js required for PyExecJS

# Image processing
pip install Pillow

# OCR
pip install ddddocr  # Chinese captcha OCR

---

Error Handling Patterns

# Retry with exponential backoff
import time

def fetch_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = requests.get(url, timeout=10)
            if resp.status_code == 200:
                return resp
            elif resp.status_code == 429:
                wait = int(resp.headers.get("Retry-After", 60))
                logger.warning(f"Rate limited, waiting {wait}s")
                time.sleep(wait)
            elif resp.status_code == 403:
                logger.error("Forbidden — may need cookies/proxy")
                break
            else:
                resp.raise_for_status()
        except requests.RequestException as e:
            wait = 2 ** attempt
            logger.warning(f"Attempt {attempt+1} failed: {e}, retry in {wait}s")
            time.sleep(wait)
    return None

---

This skill ensures every generated crawler is robust, production-ready, anti-detection-aware, and tailored to modern web architectures (Vue/React/SPA/SSR/API).