Python爬虫进阶：反爬机制突破与数据存储实战指南

import requests
from proxy_pool import TunnelProxy  # 假设的隧道代理库

# 初始化隧道代理（以站大爷为例）
proxy = TunnelProxy(
    api_url="https://tunnel.zhandaye.com/api",
    api_key="YOUR_KEY",
    rotate_strategy="per_request"  # 每请求切换IP
)

def fetch_data(url):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    try:
        proxies = {"http": proxy.get_proxy(), "https": proxy.get_proxy()}
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        if response.status_code == 403:
            proxy.rotate_now()  # 立即切换IP
            return fetch_data(url)  # 重试
        return response.text
    except Exception as e:
        proxy.rotate_now()
        return fetch_data(url)

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

options = Options()
options.add_argument("--headless")  # 无头模式
options.add_argument("--disable-gpu")
driver = webdriver.Chrome(options=options)

driver.get("https://example.com")
# 等待JS渲染完成
driver.implicitly_wait(5)
content = driver.page_source  # 获取渲染后的HTML
driver.quit()

# Redis缓存最新价格（内存数据库）
import redis
r = redis.Redis(host='localhost', port=6379)
r.set("product_123_price", "299.00", ex=3600)  # 1小时过期

# MongoDB存储历史数据（文档数据库）
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017/")
db = client["price_monitor"]
collection = db["history"]
collection.insert_one({
    "product_id": "123",
    "price": 299.00,
    "timestamp": datetime.now()
})

# PostgreSQL分析数据（关系型数据库）
import psycopg2
conn = psycopg2.connect("dbname=monitor user=postgres")
cur = conn.cursor()
cur.execute("""
    CREATE TABLE IF NOT EXISTS price_trend (
        id SERIAL PRIMARY KEY,
        product_id VARCHAR(50),
        price DECIMAL(10,2),
        record_time TIMESTAMP
    )
""")

# 同步版本
import requests
def sync_crawl(urls):
    results = []
    for url in urls:
        res = requests.get(url)
        results.append(res.text)
    return results

# 异步版本（aiohttp）
import aiohttp
import asyncio
async def async_crawl(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [await r.text() for r in responses]

# 实测结果（100个URL）：
# 同步：12.4s
# 异步：3.1s

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))
def robust_request(url):
    response = requests.get(url, timeout=10)
    if response.status_code != 200:
        response.raise_for_status()
    return response