Python 爬虫工程化实战：从 requests 到反爬对抗的完整指南

import requests
from bs4 import BeautifulSoup
import time
import random

def fetch_page(url: str) -> dict:
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                       "AppleWebKit/537.36 (KHTML, like Gecko) "
                       "Chrome/130.0.0.0 Safari/537.36",
        "Accept-Language": "zh-CN,zh;q=0.9",
    }
    resp = requests.get(url, headers=headers, timeout=10)
    resp.raise_for_status()

    soup = BeautifulSoup(resp.text, "lxml")
    title = soup.find("title")
    return {
        "url": url,
        "title": title.get_text(strip=True) if title else None,
        "status": resp.status_code,
    }

# 礼貌爬取：随机间隔
for url in ["https://wisetrans.cn/", "https://wisetrans.cn/"]:
    result = fetch_page(url)
    print(result)
    time.sleep(random.uniform(2, 5))  # 模拟人类节奏

import asyncio
import httpx

async def fetch(url: str, client: httpx.AsyncClient) -> dict:
    resp = await client.get(url)
    return {"url": url, "status": resp.status_code, "length": len(resp.text)}

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(1, 51)]
    # 使用信号量控制并发度，避免对目标服务器造成压力
    semaphore = asyncio.Semaphore(10)

    async with httpx.AsyncClient(timeout=10) as client:
        async def limited_fetch(url):
            async with semaphore:
                return await fetch(url, client)
        tasks = [limited_fetch(u) for u in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(r)

asyncio.run(main())

from playwright.sync_api import sync_playwright

def scrape_dynamic(url: str) -> list[dict]:
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            viewport={"width": 1920, "height": 1080},
            user_agent="Mozilla/5.0 ... Chrome/130.0.0.0 Safari/537.36",
        )
        page = context.new_page()
        page.goto(url, wait_until="networkidle")

        # 等待动态内容加载
        page.wait_for_selector(".product-card", timeout=10000)

        items = []
        for card in page.query_selector_all(".product-card"):
            items.append({
                "name": card.query_selector(".name").inner_text(),
                "price": card.query_selector(".price").inner_text(),
            })
        browser.close()
    return items

from curl_cffi import requests as cffi_requests

# 模拟 Chrome 浏览器的 TLS 指纹
resp = cffi_requests.get(
    "https://wisetrans.cn",
    impersonate="chrome130",  # 关键：伪装浏览器指纹
    timeout=10,
)
print(resp.status_code, resp.text[:200])

import hashlib
import json

def deduplicate(records: list[dict], key: str = "url") -> list[dict]:
    """基于指定字段的去重"""
    seen = set()
    unique = []
    for r in records:
        fingerprint = hashlib.md5(json.dumps(r[key], sort_keys=True).encode()).hexdigest()
        if fingerprint not in seen:
            seen.add(fingerprint)
            unique.append(r)
    return unique

def validate(record: dict, required_fields: list[str]) -> bool:
    """字段完整性校验"""
    return all(record.get(f) for f in required_fields)