web-scraper

Crawler/Data Collection Engineer Agent

Applicable Scenarios

When the user needs to:

• Collect structured or unstructured data from websites/APIs/Apps
• Build a maintainable crawler system (standalone or distributed)
• Solve anti-scraping problems (IP bans, CAPTCHAs, parameter encryption, font-based anti-scraping, etc.)
• Scrape dynamically rendered pages (SPA/JS rendering/WebSocket)
• Reverse engineer App request protocols (Protobuf/encrypted parameters)
• Data cleaning, deduplication, storage to databases/files
• Crawler monitoring and operations

Workflow

Phase 1: Requirements Analysis and Feasibility Assessment

1. Clarify the target: What data needs to be collected? Field list? Data volume? Update frequency?
2. Source analysis：
   • Web page → Analyze URL structure, pagination method, whether data is in the HTML
   • API → Capture packets to analyze request/response format, authentication method, parameter encryption
   • App → Determine packet capture approach (mitmproxy/Charles), protocol type (HTTP/WebSocket/gRPC)
3. Feasibility assessment：
   • Whether there are robots.txt restrictions
   • Anti-scraping strength (CAPTCHAs, rate limits, WAF, fingerprint detection)
   • Legal compliance (data usage boundaries, personal information protection)
4. Solution selection：
   • Simple static pages → requests + BeautifulSoup/lxml
   • Dynamically rendered pages → Playwright/Selenium/DrissionPage
   • Large-scale collection → Scrapy + distributed solution
   • App data → mitmproxy + reverse engineering analysis

Phase 2: Technical Solution Design

Select the tech stack based on the target:

Scenario	Recommended solution
Simple static pages (small volume)	requests + BeautifulSoup + lxml
Simple static pages (large volume)	Scrapy + parsel
Dynamically rendered pages	Playwright / DrissionPage
Large-scale distributed	Scrapy-Redis + Celery + Kafka
App packet capture	mitmproxy + Frida + jadx
High-performance asynchronous	aiohttp + BeautifulSoup + uvloop
Anti-scraping countermeasures	curl_cffi + proxy pool + browser fingerprint spoofing

Phase 2: Environment Setup

# 基础环境
pip install requests beautifulsoup4 lxml parsel httpx

# 爬虫框架
pip install scrapy scrapy-redis scrapy-splash

# 浏览器自动化
pip install playwright selenium
playwright install chromium

# 异步
pip install aiohttp aiofiles

# 数据存储
pip install pymongo redis pymysql psycopg2-binary

# 反爬工具
pip install curl_cffi fake-useragent

# App抓包与逆向
pip install mitmproxy
# Frida: pip install frida-tools (需配合手机端frida-server)

Phase 3: Implementation Steps

3.1 Static Page Scraping

import requests
from bs4 import BeautifulSoup
import json

headers = {
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) ...',
    'Accept': 'text/html,application/xhtml+xml,...',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}

# 基础请求
resp = requests.get(url, headers=headers, timeout=10)
resp.raise_for_status()
resp.encoding = resp.apparent_encoding  # 自动检测编码

# 解析
soup = BeautifulSoup(resp.text, 'lxml')
# 或使用 parsel (Scrapy核心解析器)
from parsel import Selector
sel = Selector(text=resp.text)

3.2 Dynamic Page Scraping (Playwright)

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    context = browser.new_context(
        user_agent='Mozilla/5.0 ...',
        viewport={'width': 1920, 'height': 1080},
        locale='zh-CN'
    )
    page = context.new_page()
    page.goto(url, wait_until='networkidle')
    
    # 等待元素出现
    page.wait_for_selector('.content-item', timeout=10000)
    
    # 滚动加载
    for _ in range(5):
        page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
        page.wait_for_timeout(2000)
    
    # 提取数据
    items = page.evaluate('''() => {
        return Array.from(document.querySelectorAll('.item')).map(el => ({
            title: el.querySelector('.title')?.innerText,
            url: el.querySelector('a')?.href
        }))
    }''')
    
    browser.close()

3.3 Using the Scrapy Framework

# 创建项目
scrapy startproject myproject
cd myproject
scrapy genspider example example.com

# spiders/example.py
import scrapy
from scrapy.http import HtmlResponse

class ExampleSpider(scrapy.Spider):
    name = 'example'
    allowed_domains = ['example.com']
    start_urls = ['https://example.com/page/1']
    
    custom_settings = {
        'DOWNLOAD_DELAY': 1.5,
        'RANDOMIZE_DOWNLOAD_DELAY': True,
        'CONCURRENT_REQUESTS': 8,
        'DOWNLOADER_MIDDLEWARES': {
            'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,
            'scrapy_user_agents.middlewares.RandomUserAgentMiddleware': 400,
        },
        'ITEM_PIPELINES': {
            'myproject.pipelines.DuplicatesPipeline': 100,
            'myproject.pipelines.DatabasePipeline': 300,
        }
    }
    
    def parse(self, response: HtmlResponse):
        # 提取数据
        for item in response.css('.item'):
            yield {
                'title': item.css('.title::text').get(),
                'url': item.css('a::attr(href)').get(),
                'price': item.css('.price::text').re_first(r'[\d.]+'),
            }
        
        # 翻页
        next_page = response.css('.next::attr(href)').get()
        if next_page:
            yield scrapy.Request(url=response.urljoin(next_page))

3.4 Dynamic Pages (Playwright Complete Example)

from playwright.sync_api import sync_playwright
import json

def scrape_dynamic_page(url: str) -> list:
    """采集动态渲染页面，支持滚动加载和等待条件"""
    with sync_playwright() as p:
        browser = p.chromium.launch(
            headless=True,
            args=['--disable-blink-features=AutomationControlled']
        )
        context = browser.new_context(
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...',
            viewport={'width': 1920, 'height': 1080},
            locale='zh-CN',
            # 注入反检测脚本
            extra_http_headers={'Accept-Language': 'zh-CN,zh;q=0.9'}
        )
        
        # 注入反自动化检测
        context.add_init_script("""
            Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
            Object.defineProperty(navigator, 'plugins', {get: () => [1,2,3,4,5]});
        """)
        
        page = context.new_page()
        page.goto(url, wait_until='networkidle')
        
        # 等待数据加载
        page.wait_for_selector('.data-item', timeout=15000)
        
        # 滚动加载
        for _ in range(3):
            page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
            page.wait_for_timeout(2000)
        
        # 提取数据
        data = page.evaluate('''() => {
            return Array.from(document.querySelectorAll('.item')).map(el => ({
                title: el.querySelector('.title')?.innerText?.trim(),
                price: el.querySelector('.price')?.innerText?.trim(),
                link: el.querySelector('a')?.href
            }))
        }''')
        
        browser.close()
        return data

3.3 API Scraping

import requests
import time
import hashlib
import json

def fetch_api_data(base_url: str, params: dict, api_key: str = None):
    """通用API数据采集，支持分页和认证"""
    headers = {
        'User-Agent': 'Mozilla/5.0 ...',
        'Accept': 'application/json',
    }
    if api_key:
        headers['Authorization'] = f'Bearer {api_key}'
    
    all_data = []
    page = 1
    while True:
        params['page'] = page
        resp = requests.get(base_url, params=params, headers=headers, timeout=15)
        resp.raise_for_status()
        data = resp.json()
        
        # 提取数据
        items = data.get('data', data.get('items', data.get('results', [])))
        if not items:
            break
        all_data.extend(items)
        
        # 分页判断
        total = data.get('total', data.get('count', 0))
        if page * len(items) >= total:
            break
        page += 1
        time.sleep(0.5)
    
    return all_data

3.5 Anti-Scraping Countermeasures

Anti-scraping method	Countermeasure
IP rate limiting	Proxy pool rotation + randomized request intervals
User-Agent detection	fake-useragent / real UA pool
Cookie verification	Simulated login + Session maintenance + Cookie persistence
Font-based anti-scraping	Download font file → parse mapping relationships with fontTools
CSS offset	Analyze CSS styles to restore the real text order
Image CAPTCHA	ddddocr / PaddleOCR / captcha-solving platform
Slider verification	Playwright trajectory simulation / third-party captcha solving
WebDriver detection	Playwright stealth / undetected-chromedriver
Parameter signing	JS reverse engineering → rewrite signing algorithm in Python
WAF/Cloudflare	curl_cffi / cloudscraper / flaresolverr
Font anti-scraping	fontTools parses TTF/WOFF mapping tables

3.6 Data cleaning and storage

import pandas as pd
from pymongo import MongoClient
import json

def clean_and_store(raw_data: list, collection_name: str):
    """数据清洗与入库"""
    df = pd.DataFrame(raw_data)
    
    # 去重
    df = df.drop_duplicates(subset=['id', 'url'])
    
    # 清洗
    df['price'] = pd.to_numeric(df['price'], errors='coerce')
    df['title'] = df['title'].str.strip()
    df['created_at'] = pd.to_datetime(df['created_at'], errors='coerce')
    
    # 过滤无效数据
    df = df.dropna(subset=['title'])
    
    # 入库
    client = MongoClient('mongodb://localhost:27017')
    db = client['scraped_data']
    collection = db[collection_name]
    
    records = df.to_dict('records')
    for record in records:
        collection.update_one(
            {'_id': record.get('id', record.get('url'))},
            {'$set': record},
            upsert=True
        )
    
    return len(records)

Stage 4: Anti-scraping countermeasures in practice

4.1 IP proxy pool

import random
import requests
from typing import List

class ProxyPool:
    """简易代理池"""
    def __init__(self):
        self.proxies: List[dict] = []
        self._load_proxies()
    
    def _load_proxies(self):
        """从代理源加载（示例：免费代理源）"""
        sources = [
            'https://raw.githubusercontent.com/proxifly/free-proxy-list/main/proxies/proxies.json',
            # 也可从数据库/文件加载
        ]
        for source in sources:
            try:
                resp = requests.get(source, timeout=5)
                data = resp.json()
                for p in data:
                    self.proxies.append({
                        'http': f'http://{p["ip"]}:{p["port"]}',
                        'https': f'http://{p["ip"]}:{p["port"]}',
                    })
            except Exception:
                continue
    
    def get_random(self) -> dict:
        return random.choice(self.proxies) if self.proxies else {}
    
    def test_proxy(self, proxy: dict) -> bool:
        """测试代理可用性"""
        try:
            resp = requests.get('http://httpbin.org/ip', proxies=proxy, timeout=5)
            return resp.status_code == 200
        except:
            return False

4.2 Browser fingerprint spoofing (Playwright)

# 反检测初始化脚本
STEALTH_SCRIPT = """
// 隐藏WebDriver
Object.defineProperty(navigator, 'webdriver', {get: () => undefined});

// 伪装Chrome
window.chrome = { runtime: {} };

// 覆盖权限查询
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
    parameters.name === 'notifications' ?
        Promise.resolve({state: Notification.permission}) :
        originalQuery(parameters)
);

// 覆盖plugins
Object.defineProperty(navigator, 'plugins', {
    get: () => [1, 2, 3, 4, 5]
});

// 覆盖languages
Object.defineProperty(navigator, 'languages', {
    get: () => ['zh-CN', 'zh', 'en']
});
"""

4.3 CAPTCHA handling

# ddddocr - 轻量级OCR验证码
import ddddocr

ocr = ddddocr.DdddOcr()
with open('captcha.png', 'rb') as f:
    result = ocr.classification(f.read())
print(f'验证码识别结果: {result}')

# 滑块验证码 - 使用Playwright模拟轨迹
import random

def slide_verify(page, slider_selector: str, gap_selector: str):
    """模拟人类滑块验证"""
    slider = page.locator(slider_selector)
    gap = page.locator(gap_selector)
    
    slider_box = slider.bounding_box()
    gap_box = gap.bounding_box()
    
    start_x = slider_box['x'] + slider_box['width'] / 2
    start_y = slider_box['y'] + slider_box['height'] / 2
    target_x = gap_box['x'] + gap_box['width'] / 2
    
    distance = target_x - start_x
    
    # 模拟人类轨迹：先快后慢 + 抖动
    tracks = []
    current = 0
    mid = distance * 0.7
    while current < distance:
        if current < mid:
            move = random.randint(3, 8)
        else:
            move = random.randint(1, 3)
        current += move
        tracks.append(current)
    
    page.mouse.move(start_x, start_y)
    page.mouse.down()
    for x in tracks:
        page.mouse.move(start_x + x, start_y + random.randint(-2, 2))
        page.wait_for_timeout(random.randint(10, 30))
    page.mouse.up()

Stage 5: Data cleaning and storage

import pandas as pd
from pymongo import MongoClient, UpdateOne
import json

def clean_scraped_data(raw_data: list) -> pd.DataFrame:
    """通用数据清洗流程"""
    df = pd.DataFrame(raw_data)
    
    # 去重
    if 'url' in df.columns:
        df = df.drop_duplicates(subset=['url'])
    if 'id' in df.columns:
        df = df.drop_duplicates(subset=['id'])
    
    # 文本清洗
    text_cols = [c for c in df.columns if df[c].dtype == 'object']
    for col in text_cols:
        df[col] = df[col].str.strip().str.replace(r'\s+', ' ', regex=True)
    
    # 数值清洗
    for col in ['price', 'amount', 'count', 'score']:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')
    
    # 日期标准化
    for col in ['date', 'created_at', 'updated_at']:
        if col in df.columns:
            df[col] = pd.to_datetime(df[col], errors='coerce')
    
    return df

Stage 6: Distributed crawler architecture

# Scrapy-Redis 配置
# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = 'redis://localhost:6379'
SCHEDULER_PERSIST = True  # 爬虫停止后保留请求队列
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'

Stage 7: App packet capture and reverse engineering

# mitmproxy 脚本示例 - 拦截App请求
# mitmproxy_script.py
from mitmproxy import http
import json

def request(flow: http.HTTPFlow):
    """拦截并记录请求"""
    url = flow.request.pretty_url
    if 'api.target.com' in url:
        print(f"[REQ] {url}")
        print(f"  Headers: {dict(flow.request.headers)}")
        if flow.request.content:
            print(f"  Body: {flow.request.content[:500]}")

def response(flow: http.HTTPFlow):
    """拦截并解析响应"""
    url = flow.request.pretty_url
    if 'api.target.com' in url:
        print(f"[RES] {url}")
        if 'application/json' in flow.response.headers.get('content-type', ''):
            data = json.loads(flow.response.text)
            # 保存到文件
            with open(f'data/{int(time.time())}.json', 'w') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)

Stage 8: Troubleshooting common issues

Problem	Cause	Solution
403 Forbidden	Request blocked by WAF/Cloudflare	Use curl_cffi / cloudscraper / real browser fingerprints
429 Too Many Requests	Request frequency too high	Increase intervals, use proxy pools, distribute
Empty data/structure changes	Website redesign	Check page structure, update selectors, add structure-change alerts
Garbled Chinese characters	Encoding detection error	resp.encoding = resp.apparent_encoding / chardet detection
Memory overflow	Data volume too large	Use iterators, paginated writes, limit concurrency
Connection timeout	Network/proxy issues	Set up retry mechanisms, timeouts, backup proxies
CAPTCHA popup	Risk control triggered	Reduce frequency, change IP, use captcha-solving services

Stage 9: Law and compliance

1. robots.txt: Comply with the website's crawling rules statement
2. Frequency control: Do not put pressure on the target server (recommended interval ≥ 1 second)
3. Data usage: Do not collect personal private information, do not use for competitive commercial purposes
4. Copyright notice: Pay attention to data copyright ownership and usage licenses
5. Anti-scraping countermeasure boundaries: Do not perform destructive operations (DDoS, vulnerability exploitation)

---

Template files

1. Scrapy project template

# scrapy_spider_template.py
"""
Scrapy爬虫模板 - 使用方式:
1. scrapy startproject myproject
2. 将本模板放入 spiders/ 目录
3. 修改 settings.py 配置中间件和管道
"""
import scrapy
from scrapy.http import HtmlResponse
from urllib.parse import urljoin
import json

class BaseSpider(scrapy.Spider):
    """基础爬虫模板，继承后重写 parse_item 即可"""
    name = 'base'
    allowed_domains = []
    start_urls = []
    
    custom_settings = {
        'DOWNLOAD_DELAY': 1.0,
        'RANDOMIZE_DOWNLOAD_DELAY': True,
        'CONCURRENT_REQUESTS': 8,
        'CONCURRENT_REQUESTS_PER_DOMAIN': 4,
        'RETRY_TIMES': 3,
        'RETRY_HTTP_CODES': [429, 500, 502, 503, 504],
        'DOWNLOAD_TIMEOUT': 15,
        'ROBOTSTXT_OBEY': False,
        'DEFAULT_REQUEST_HEADERS': {
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        }
    }
    
    def parse(self, response):
        """主解析方法 - 子类重写"""
        raise NotImplementedError
    
    def parse_item(self, response):
        """解析单个详情页 - 子类重写"""
        raise NotImplementedError
    
    def closed(self, reason):
        """爬虫关闭时的清理工作"""
        self.logger.info(f"Spider closed: {reason}")

Stage 7: App packet capture and reverse engineering

7.1 mitmproxy interception script

# mitmproxy_addon.py
# 运行: mitmproxy -s mitmproxy_addon.py
from mitmproxy import http
import json
import time
import os

# 配置目标域名
TARGET_DOMAINS = ['api.target.com', 'data.target.com']
OUTPUT_DIR = 'captured_data'

def request(flow: http.HTTPFlow):
    """拦截请求"""
    for domain in TARGET_DOMAINS:
        if domain in flow.request.pretty_host:
            print(f"[REQ] {flow.request.method} {flow.request.pretty_url}")
            if flow.request.headers.get('Content-Type', '').startswith('application/json'):
                try:
                    body = json.loads(flow.request.content)
                    print(f"  Body: {json.dumps(body, ensure_ascii=False, indent=2)[:500]}")
                except:
                    pass

def response(flow: http.HTTPFlow):
    """拦截响应"""
    for domain in TARGET_DOMAINS:
        if domain in flow.request.pretty_host:
            os.makedirs(OUTPUT_DIR, exist_ok=True)
            fname = f"{OUTPUT_DIR}/{int(time.time())}_{flow.request.path.replace('/', '_')}.json"
            with open(fname, 'w') as f:
                f.write(flow.response.text)
            print(f"[SAVED] {fname}")

Stage 8: Data quality assurance

def validate_scraped_data(df: pd.DataFrame, rules: dict) -> dict:
    """
    数据质量校验
    rules: {
        'title': {'required': True, 'min_length': 1, 'max_length': 200},
        'price': {'required': True, 'type': 'numeric', 'min': 0},
        'url': {'required': True, 'pattern': r'^https?://'},
    }
    """
    report = {'total': len(df), 'passed': 0, 'failed': 0, 'errors': []}
    
    for idx, row in df.iterrows():
        row_errors = []
        for field, rule in rules.items():
            val = row.get(field)
            if rule.get('required') and (pd.isna(val) or val == ''):
                row_errors.append(f"{field}: 必填字段为空")
            if rule.get('type') == 'numeric' and val:
                try:
                    float(val)
                except (ValueError, TypeError):
                    row_errors.append(f"{field}: 非数值类型")
            if rule.get('pattern') and val:
                import re
                if not re.match(rule['pattern'], str(val)):
                    row_errors.append(f"{field}: 格式不匹配")
        
        if row_errors:
            report['failed'] += 1
            report['errors'].append({'row': idx, 'errors': row_errors})
        else:
            report['passed'] += 1
    
    return report

Stage 10: Monitoring and operations

# 爬虫健康检查脚本
import requests
import time
from datetime import datetime

def health_check(spider_name: str, expected_count: int, timeout: int = 300):
    """监控爬虫运行状态"""
    start = time.time()
    while time.time() - start < timeout:
        # 检查数据库记录数
        count = get_db_count(spider_name)
        if count >= expected_count:
            return {'status': 'success', 'count': count, 'time': time.time() - start}
        
        # 检查爬虫进程
        import psutil
        spider_running = any('scrapy' in p.name() for p in psutil.process_iter())
        if not spider_running:
            return {'status': 'crashed', 'count': count}
        
        time.sleep(10)
    
    return {'status': 'timeout', 'count': count}

Stage 10: Legal and compliance checklist

• Check whether robots.txt allows collection
• Confirm whether the data contains personal private information (name, phone, address, ID card, etc.)
• Confirm the purpose of data usage (research/commercial/personal)
• Set reasonable request intervals so as not to put pressure on the target server
• Do not perform destructive operations such as password cracking or vulnerability exploitation
• Do not use the collected data for direct competition or illegal purposes
• Comply with the target website's terms of service
• Pay attention to legal requirements for cross-border data transfer

---

Common pitfalls and precautions

1. Do not hard-code selectors: Website structures change; use relatively stable attributes (data-*, id) or configurable selectors
2. Always handle exceptions: Network timeouts, parsing failures, and structural changes all need fallbacks
3. Logs are the lifeline: Record each request's URL, status code, duration, and data volume
4. Incremental crawling: Record the last crawl position to avoid repeating full crawls
5. Data versioning: Save raw responses (HTML/JSON) for later re-parsing
6. Proxy pool maintenance: Regularly check proxy availability and remove failed proxies
7. Rate control: Do not use fixed intervals; use random intervals of ±30%
8. Encoding issues: Always specify encoding, and prefer apparent_encoding
9. Error retries: Retry network errors 3 times with exponential backoff
10. Local cache: Do not repeatedly request pages that have already been downloaded

Verification methods

1. Single-page test: First use requests/curl to test whether a single URL can retrieve data normally
2. Parsing verification: Compare extracted data field by field with the actual page content
3. Batch verification: Crawl 100 sample records and manually spot-check the accuracy rate
4. Stress testing: Gradually increase concurrency and observe the target server's response
5. Stability testing: Run continuously for 1 hour and check for interruptions or exceptions
6. Data integrity: Check for missing fields, duplicate records, and abnormal formats