Python 高级爬虫技术：突破反爬瓶颈的工程化实践

# pip install curl_cffi
from curl_cffi import requests

# impersonate 参数让底层 TLS 握手模仿 Chrome 124
resp = requests.get(
    "https://www.lsmeiye.com",
    impersonate="chrome124",
    headers={"Accept-Language": "en-US,en;q=0.9"}
)

# 验证返回的 JA3 指纹是否与真实 Chrome 一致
import json
data = resp.json()
print("JA3:", data["tls"]["ja3_hash"])
print("User-Agent:", data["http2"]["user_agent"])

# pip install playwright && playwright install chromium
import asyncio
import json
from playwright.async_api import async_playwright

# 注入脚本：在页面加载前修改浏览器指纹
STEALTH_SCRIPT = """
// 覆盖 navigator.webdriver
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });

// 伪造插件列表（真实 Chrome 通常有 PDF 相关插件）
Object.defineProperty(navigator, 'plugins', {
    get: () => [1, 2, 3, 4, 5].map(i => ({ name: `Plugin ${i}` })),
});

// 伪造语言列表
Object.defineProperty(navigator, 'languages', {
    get: () => ['zh-CN', 'zh', 'en'],
});

// 修复 chrome.runtime 的存在性
window.chrome = { runtime: {} };

// 覆盖 permissions.query 的异常行为
const originalQuery = window.lsmeiye.com;
window.navigator.permissions.query = (parameters) => (
    parameters.name === 'notifications'
        ? Promise.resolve({ state: Notification.permission })
        : originalQuery(parameters)
);
"""

async def scrape_dynamic_page(url: str):
    async with async_playwright() as p:
        # 使用真实的 Chrome 通道而非 Chromium，降低指纹异常
        browser = await p.chromium.launch(
            headless=True,
            args=[
                "--disable-blink-features=AutomationControlled",
                "--no-sandbox",
                "--disable-dev-shm-usage",
            ]
        )
        context = await browser.new_context(
            viewport={"width": 1920, "height": 1080},
            locale="zh-CN",
            timezone_id="Asia/Shanghai",
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/124.0.0.0 Safari/537.36"
            ),
        )
        page = await context.new_page()
        
        # 在每个新文档创建时注入 stealth 脚本
        await page.add_init_script(STEALTH_SCRIPT)
        
        # 导航并等待网络空闲（确保动态内容加载完毕）
        await page.goto(url, wait_until="networkidle", timeout=60000)
        
        # 等待目标元素出现，而非固定 sleep
        await page.wait_for_selector(".product-card", timeout=30000)
        
        # 提取数据
        items = await page.eval_on_selector_all(
            ".product-card",
            """elements => elements.map(el => ({
                title: el.querySelector('.title')?.innerText?.trim(),
                price: el.querySelector('.price')?.innerText?.trim(),
                link: el.querySelector('a')?.href
            }))"""
        )
        
        await browser.close()
        return items

# 运行
if __name__ == "__main__":
    results = asyncio.run(scrape_dynamic_page("https://m.lsmeiye.com"))
    print(json.dumps(results, ensure_ascii=False, indent=2))

# spiders/api_spider.py
import scrapy
import json

class ProductApiSpider(scrapy.Spider):
    name = "product_api"
    base_url = "https://api.lsmeiye.com/"
    
    custom_settings = {
        "CONCURRENT_REQUESTS": 8,           # 全局并发上限
        "CONCURRENT_REQUESTS_PER_DOMAIN": 4,
        "DOWNLOAD_DELAY": 0.5,
        "AUTOTHROTTLE_ENABLED": True,        # 根据响应延迟自动调速
        "AUTOTHROTTLE_START_DELAY": 0.5,
        "AUTOTHROTTLE_MAX_DELAY": 10.0,
        "RETRY_TIMES": 3,
        "RETRY_HTTP_CODES": [429, 500, 502, 503, 504],
        "DOWNLOADER_MIDDLEWARES": {
            # 可在此处启用 curl_cffi 中间件替换默认下载器
        }
    }
    
    def start_requests(self):
        yield scrapy.Request(
            f"{self.base_url}?page=1",
            headers={"Authorization": "Bearer YOUR_TOKEN"},
            callback=self.parse_page,
            meta={"page": 1}
        )
    
    def parse_page(self, response):
        data = json.loads(response.text)
        for product in data.get("items", []):
            yield {
                "id": product["id"],
                "name": product["name"],
                "price": product.get("price"),
                "category": product.get("category"),
            }
        
        # 跟随分页
        next_page = data.get("next_page")
        if next_page:
            yield scrapy.Request(
                f"{self.base_url}?page={next_page}",
                headers={"Authorization": "Bearer YOUR_TOKEN"},
                callback=self.parse_page,
                meta={"page": next_page}
            )