网络数据侦探：抓包工具在爬虫开发中的艺术与科学

# 示例：分析抓包得到的API请求结构

import requests

import json

def analyze\_api\_request():

 # 从抓包工具中获取的请求信息

    api\_url = "https://api.example.com/data"

    headers = {

 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",

 "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9",

 "Content-Type": "application/json"

    }

    params = {

 "page": 1,

 "limit": 20,

 "timestamp": 1635678900

    }

 # 发送请求并分析响应

    response = requests.get(api\_url, headers=headers, params=params)

    data = response.json()

 # 关键信息提取

 print(f"状态码: {response.status\_code}")

 print(f"响应头: {dict(response.headers)}")

 print(f"数据量: {len(data.get('items', []))}条")

 return data

# 关键行点评：

# 第8行：User-Agent伪装是绕过基础反爬的重要措施

# 第9行：Authorization头通常包含认证令牌，需要动态获取

# 第13-16行：参数结构分析有助于理解API的分页机制

# 示例：基于抓包数据构建可重用的爬虫请求

import time

import hashlib

class RequestBuilder:

 def \_\_init\_\_(self, base\_url):

 self.base\_url = base\_url

 self.session = requests.Session()

 def build\_headers(self, referer=None):

 """构建请求头，模拟真实浏览器"""

        headers = {

 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",

 "Accept": "application/json, text/plain, \*/\*",

 "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",

 "Connection": "keep-alive"

        }

 if referer:

            headers["Referer"] = referer

 return headers

 def add\_signature(self, params):

 """添加签名参数（常见反爬机制）"""

        timestamp = str(int(time.time()))

        params['timestamp'] = timestamp

 # 模拟签名算法（从抓包分析得出）

        sign\_str = f"key1={params.get('key1', '')}&timestamp={timestamp}&secret=example"

        params['sign'] = hashlib.md5(sign\_str.encode()).hexdigest()

 return params

# 示例：电商网站API逆向分析

import json

import re

from urllib.parse import urlencode

class EcommerceCrawler:

 def \_\_init\_\_(self):

 self.session = requests.Session()

 self.base\_headers = {

 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",

 "Accept": "application/json, text/plain, \*/\*",

 "X-Requested-With": "XMLHttpRequest"

        }

 def extract\_api\_endpoints(self, html\_content):

 """从页面HTML中提取API端点（通过抓包发现）"""

 # 使用正则表达式匹配API URL模式

        api\_patterns = [

 r'api\.example\.com\/v\d+\/products',

 r'\/ajax\/product\/list',

 r'window\.API\_URL\s\*=\s\*["\']([^"\']+)["\']'

        ]

        endpoints = []

 for pattern in api\_patterns:

            matches = re.findall(pattern, html\_content)

            endpoints.extend(matches)

 return endpoints

 def analyze\_product\_api(self, api\_url, page=1):

 """分析商品列表API"""

 # 构建请求参数（基于抓包分析）

        params = {

 "page": page,

 "pageSize": 40,

 "sort": "default",

 "categoryId": 1000,

 "timestamp": int(time.time() \* 1000)

        }

 # 添加签名（如果存在）

 if self.has\_signature\_requirement(api\_url):

            params = self.add\_api\_signature(params)

        response = self.session.get(api\_url, params=params, headers=self.base\_headers)

 return self.parse\_api\_response(response.json())

# 示例：反爬机制识别与绕过

class AntiAntiCrawler:

 def detect\_anti\_crawler\_measures(self, response):

 """检测反爬虫措施"""

        indicators = {

 "cloudflare": "cf-ray" in response.headers,

 "rate\_limit": response.status\_code == 429,

 "captcha": "captcha" in response.text.lower(),

 "blocked": response.status\_code == 403,

 "js\_challenge": "window.location" in response.text

        }

 return {k: v for k, v in indicators.items() if v}

 def bypass\_cloudflare(self, url):

 """绕过Cloudflare防护"""

 # 使用selenium模拟真实浏览器

 from selenium import webdriver

 from selenium.webdriver.chrome.options import Options

        options = Options()

        options.add\_argument("--headless")

        driver = webdriver.Chrome(options=options)

        driver.get(url)

        time.sleep(5)  # 等待JS执行完成

 # 获取经过JS处理后的页面内容

        page\_source = driver.page\_source

        driver.quit()

 return page\_source

# 示例：基于抓包数据的性能优化

import asyncio

import aiohttp

from concurrent.futures import ThreadPoolExecutor

class OptimizedCrawler:

 def \_\_init\_\_(self, max\_concurrent=10):

 self.max\_concurrent = max\_concurrent

 self.session = None

 async def batch\_crawl(self, urls):

 """批量异步爬取"""

        connector = aiohttp.TCPConnector(limit=self.max\_concurrent)

        timeout = aiohttp.ClientTimeout(total=30)

 async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:

            tasks = [self.fetch\_url(session, url) for url in urls]

            results = await asyncio.gather(\*tasks, return\_exceptions=True)

 return results

 async def fetch\_url(self, session, url):

 """单个URL爬取"""

 async with session.get(url) as response:

 # 基于抓包分析设置合理的超时和重试策略

 if response.status == 200:

 return await response.text()

 else:

 raise Exception(f"请求失败: {response.status}")

# 示例：数据验证与清洗

import pandas as pd

from datetime import datetime

class DataValidator:

 def \_\_init\_\_(self):

 self.validation\_rules = {

 "price": lambda x: x > 0,

 "title": lambda x: len(x) > 0 and len(x) < 200,

 "timestamp": lambda x: datetime.fromisoformat(x) < datetime.now()

        }

 def validate\_product\_data(self, products):

 """验证商品数据质量"""

        valid\_products = []

        validation\_errors = []

 for product in products:

            is\_valid = True

            errors = []

 for field, rule in self.validation\_rules.items():

 if field in product:

 try:

 if not rule(product[field]):

                            is\_valid = False

                            errors.append(f"{field}验证失败")

 except Exception as e:

                        is\_valid = False

                        errors.append(f"{field}验证异常: {str(e)}")

 if is\_valid:

                valid\_products.append(product)

 else:

                validation\_errors.append({

 "product": product.get("id", "unknown"),

 "errors": errors

                })

 return valid\_products, validation\_errors