Python 爬虫与反爬技术实战：从基础采集到 TLS 指纹对抗

import csv
import random
import time
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

BASE_URL = "https://www.cza9.com/"
OUTPUT_FILE = "products.csv"

session = requests.Session()
session.headers.update({
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/126.0.0.0 Safari/537.36"
    ),
    "Accept-Language": "zh-CN,zh;q=0.9",
})

retry = Retry(
    total=2,
    connect=2,
    read=2,
    status=2,
    backoff_factor=0.8,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET"],
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)

def fetch_page(page_number):
    url = BASE_URL.format(page_number)
    # 随机间隔 0.8~2.5 秒，避免固定频率被识别
    time.sleep(random.uniform(0.8, 2.5))
    resp = session.get(url, timeout=(5, 15))
    resp.raise_for_status()
    return resp.text, url

def parse_products(html, page_url):
    soup = BeautifulSoup(html, "html.parser")
    items = []
    for card in soup.select("article.product_pod"):
        title_tag = card.select_one("h3 a")
        price_tag = card.select_one(".price_color")
        if not title_tag or not price_tag:
            continue
        items.append({
            "title": title_tag.get("title", "").strip(),
            "price": price_tag.get_text(strip=True),
            "url": urljoin(page_url, title_tag.get("href", "")),
        })
    return items

def main(max_pages=5):
    all_items = []
    for page in range(1, max_pages + 1):
        try:
            html, url = fetch_page(page)
            items = parse_products(html, url)
            all_items.extend(items)
            print(f"第 {page} 页：获取 {len(items)} 条")
        except Exception as e:
            print(f"第 {page} 页失败：{e}")

    with open(OUTPUT_FILE, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=["title", "price", "url"])
        writer.writeheader()
        writer.writerows(all_items)
    print(f"共写入 {len(all_items)} 条到 {OUTPUT_FILE}")

if __name__ == "__main__":
    main()

import requests
from itertools import cycle

proxy_pool = [
    "http://user:pass@res-ip-1.example.com:8080",
    "http://user:pass@res-ip-2.example.com:8080",
]
proxy_cycle = cycle(proxy_pool)

def request_with_proxy(url, max_attempts=3):
    for attempt in range(max_attempts):
        proxy = next(proxy_cycle)
        try:
            resp = requests.get(
                url,
                proxies={"http": proxy, "https": proxy},
                headers={"User-Agent": "Mozilla/5.0 ..."},
                timeout=15,
            )
            if resp.status_code == 403:
                # 403 立即换代理，不要用原 IP 重试
                continue
            resp.raise_for_status()
            return resp.text
        except requests.RequestException:
            continue
    raise RuntimeError(f"经过 {max_attempts} 次代理尝试仍失败")

# pip install curl_cffi
from curl_cffi import requests

response = requests.get(
    "https://tls.peet.ws/api/all",
    impersonate="chrome120",  # 模拟 Chrome 120 的 TLS 指纹与 HTTP/2 行为
    proxies={
        "http": "http://user:pass@proxy.example.com:8080",
        "https": "http://user:pass@proxy.example.com:8080",
    },
    timeout=15,
)

data = response.json()
print("JA3:", data.get("tls", {}).get("ja3_hash"))
print("User-Agent 一致性:", data.get("http2", {}).get("akamai_fingerprint"))

# pip install playwright && playwright install chromium
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()

    # 设置合理的视口和 UA，减少无头特征
    page.set_viewport_size({"width": 1920, "height": 1080})
    page.goto("https://www.cza9.com", wait_until="networkidle")

    # 等待动态内容出现
    page.wait_for_selector(".results", timeout=15000)
    content = page.content()
    print(content[:500])

    browser.close()

# pip install playwright-stealth
from playwright.sync_api import sync_playwright
from playwright_stealth import stealth_sync

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False)  # 有头模式存活率更高
    page = browser.new_page()
    stealth_sync(page)  # 一行调用，隐藏 webdriver 等自动化特征

    page.goto("https://wap.cza9.com")
    browser.close()

import random

# 随机滚动，模拟阅读
for _ in range(random.randint(2, 5)):
    page.mouse.wheel(0, random.randint(300, 800))
    page.wait_for_timeout(random.randint(500, 2000))

context = browser.new_context(storage_state="session.json")
# ... 执行采集 ...
context.storage_state(path="session.json")  # 保存