Python 爬虫

网络爬虫与数据采集技术，从基础请求到框架应用。

爬虫基础概念

什么是爬虫

网络爬虫（Web Crawler）是一种自动化程序，通过模拟浏览器发送 HTTP 请求，获取网页内容并从中提取所需数据。

爬虫的分类

通用爬虫：搜索引擎的核心，爬取整个互联网（如 Google、百度）
聚焦爬虫：针对特定主题进行爬取
增量爬虫：只爬取新增或更新的内容
深网爬虫：爬取需要表单提交或登录后才能访问的页面

法律与道德

python

from urllib.robotparser import RobotFileParser

rp = RobotFileParser()
rp.set_url("https://www.example.com/robots.txt")
rp.read()
print(f"允许爬取: {rp.can_fetch('MyBot/1.0', 'https://www.example.com/page')}")

from urllib.robotparser import RobotFileParser

rp = RobotFileParser()
rp.set_url("https://www.example.com/robots.txt")
rp.read()
print(f"允许爬取: {rp.can_fetch('MyBot/1.0', 'https://www.example.com/page')}")

robots.txt：告知爬虫哪些路径不可爬取，虽非强制但应遵守
请求频率控制：合理设置间隔，避免对服务器造成压力
数据使用：不得侵犯版权、隐私，不得用于非法用途

网络请求

requests 库（重点）

bash

pip install requests

pip install requests

python

import requests

# GET / POST
resp = requests.get("https://httpbin.org/get")
resp = requests.post("https://httpbin.org/post", json={"key": "value"})

# headers / params
resp = requests.get("https://httpbin.org/headers", headers={
    "User-Agent": "Mozilla/5.0 Chrome/120.0.0.0",
})
resp = requests.get("https://httpbin.org/get", params={"page": 1})

# Session（自动管理 Cookie）
session = requests.Session()
session.post("https://httpbin.org/post", data={"user": "admin"})
resp = session.get("https://httpbin.org/cookies")

# 超时与重试
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(3, backoff_factor=1, status_forcelist=[500])))
resp = session.get("https://httpbin.org/status/500", timeout=5)

# 代理
resp = requests.get("https://httpbin.org/ip", proxies={"http": "http://127.0.0.1:7890"})

# 跳过 SSL 验证
import urllib3
urllib3.disable_warnings()
resp = requests.get("https://self-signed.badssl.com", verify=False)

import requests

# GET / POST
resp = requests.get("https://httpbin.org/get")
resp = requests.post("https://httpbin.org/post", json={"key": "value"})

# headers / params
resp = requests.get("https://httpbin.org/headers", headers={
    "User-Agent": "Mozilla/5.0 Chrome/120.0.0.0",
})
resp = requests.get("https://httpbin.org/get", params={"page": 1})

# Session（自动管理 Cookie）
session = requests.Session()
session.post("https://httpbin.org/post", data={"user": "admin"})
resp = session.get("https://httpbin.org/cookies")

# 超时与重试
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(3, backoff_factor=1, status_forcelist=[500])))
resp = session.get("https://httpbin.org/status/500", timeout=5)

# 代理
resp = requests.get("https://httpbin.org/ip", proxies={"http": "http://127.0.0.1:7890"})

# 跳过 SSL 验证
import urllib3
urllib3.disable_warnings()
resp = requests.get("https://self-signed.badssl.com", verify=False)

httpx（现代替代）

bash

pip install httpx

pip install httpx

python

import httpx

# 同步
with httpx.Client() as client:
    resp = client.get("https://httpbin.org/get", params={"q": "python"})
    print(resp.status_code, resp.json())

# 异步（性能优势）
import asyncio

async def fetch_all():
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"https://httpbin.org/delay/{i}") for i in range(1, 4)]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(f"Status: {r.status_code}")

asyncio.run(fetch_all())

import httpx

# 同步
with httpx.Client() as client:
    resp = client.get("https://httpbin.org/get", params={"q": "python"})
    print(resp.status_code, resp.json())

# 异步（性能优势）
import asyncio

async def fetch_all():
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"https://httpbin.org/delay/{i}") for i in range(1, 4)]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(f"Status: {r.status_code}")

asyncio.run(fetch_all())

对比 requests：httpx 原生支持异步和 HTTP/2，API 高度兼容；requests 生态更成熟。

HTML 解析

BeautifulSoup

bash

pip install beautifulsoup4 lxml

pip install beautifulsoup4 lxml

python

from bs4 import BeautifulSoup

html = '<html><body><h1 id="title">Hello</h1><div class="content"><p>第一段</p><p class="highlight">第二段</p><a href="https://example.com/1">链接1</a></div></body></html>'
soup = BeautifulSoup(html, "lxml")

from bs4 import BeautifulSoup

html = '<html><body><h1 id="title">Hello</h1><div class="content"><p>第一段</p><p class="highlight">第二段</p><a href="https://example.com/1">链接1</a></div></body></html>'
soup = BeautifulSoup(html, "lxml")

四大对象

Tag：标签对象，如 soup.h1，可访问属性 tag["id"]
NavigableString：标签内的文本，如 h1.string 返回 "Hello"
BeautifulSoup：整个文档对象
Comment：注释内容

find / find_all / CSS 选择器

python

# find（单个）
soup.find("h1")
soup.find(id="title")

# find_all（列表）
soup.find_all("a")
soup.find_all(["h1", "p"])

# CSS 选择器
soup.select_one("#title")
soup.select("div.content a")

# 获取属性与文本
link = soup.find("a")
print(link["href"])           # 属性值
print(link.text)              # 文本内容
print(link.get_text(strip=True))

# find（单个）
soup.find("h1")
soup.find(id="title")

# find_all（列表）
soup.find_all("a")
soup.find_all(["h1", "p"])

# CSS 选择器
soup.select_one("#title")
soup.select("div.content a")

# 获取属性与文本
link = soup.find("a")
print(link["href"])           # 属性值
print(link.text)              # 文本内容
print(link.get_text(strip=True))

lxml 与 XPath

python

from lxml import etree

tree = etree.HTML(html)
items = tree.xpath("//div[@class='content']/a")
for item in items:
    print(item.text, item.get("href"))

# 常用 XPath
tree.xpath("//a[contains(@href, 'example')]")   # href 包含 example
tree.xpath("//p[position()=1]")                  # 第一个 p
tree.xpath("//p/text()")                         # 所有 p 文本

from lxml import etree

tree = etree.HTML(html)
items = tree.xpath("//div[@class='content']/a")
for item in items:
    print(item.text, item.get("href"))

# 常用 XPath
tree.xpath("//a[contains(@href, 'example')]")   # href 包含 example
tree.xpath("//p[position()=1]")                  # 第一个 p
tree.xpath("//p/text()")                         # 所有 p 文本

对比：BeautifulSoup 更 Pythonic，lxml XPath 更强大性能更好。

Scrapy 框架

bash

pip install scrapy

pip install scrapy

架构

核心组件：Engine（协调中心）-> Scheduler（请求队列）-> Downloader（下载器）-> Spider（解析器）-> Item Pipeline（数据处理）。

创建项目

bash

scrapy startproject my_spider
cd my_spider
scrapy genspider quotes quotes.toscrape.com

scrapy startproject my_spider
cd my_spider
scrapy genspider quotes quotes.toscrape.com

Spider 编写

python

# my_spider/spiders/quotes.py
import scrapy

class QuotesSpider(scrapy.Spider):
    name = "quotes"
    allowed_domains = ["quotes.toscrape.com"]
    start_urls = ["https://quotes.toscrape.com/"]

    def parse(self, response):
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(),
                "author": quote.css("small.author::text").get(),
                "tags": quote.css("div.tags a.tag::text").getall(),
            }
        # 翻页
        next_page = response.css("li.next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

# my_spider/spiders/quotes.py
import scrapy

class QuotesSpider(scrapy.Spider):
    name = "quotes"
    allowed_domains = ["quotes.toscrape.com"]
    start_urls = ["https://quotes.toscrape.com/"]

    def parse(self, response):
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(),
                "author": quote.css("small.author::text").get(),
                "tags": quote.css("div.tags a.tag::text").getall(),
            }
        # 翻页
        next_page = response.css("li.next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

Item 与 Pipeline

python

# items.py
import scrapy
class QuoteItem(scrapy.Item):
    text = scrapy.Field()
    author = scrapy.Field()
    tags = scrapy.Field()

# pipelines.py
import json
class JsonPipeline:
    def open_spider(self, spider):
        self.file = open("quotes.json", "w", encoding="utf-8")
        self.file.write("[")
        self.first = True

    def close_spider(self, spider):
        self.file.write("]")
        self.file.close()

    def process_item(self, item, spider):
        line = json.dumps(dict(item), ensure_ascii=False)
        self.file.write(f"{',' if not self.first else ''}{line}")
        self.first = False
        return item

# settings.py 中启用
ITEM_PIPELINES = {"my_spider.pipelines.JsonPipeline": 300}

# items.py
import scrapy
class QuoteItem(scrapy.Item):
    text = scrapy.Field()
    author = scrapy.Field()
    tags = scrapy.Field()

# pipelines.py
import json
class JsonPipeline:
    def open_spider(self, spider):
        self.file = open("quotes.json", "w", encoding="utf-8")
        self.file.write("[")
        self.first = True

    def close_spider(self, spider):
        self.file.write("]")
        self.file.close()

    def process_item(self, item, spider):
        line = json.dumps(dict(item), ensure_ascii=False)
        self.file.write(f"{',' if not self.first else ''}{line}")
        self.first = False
        return item

# settings.py 中启用
ITEM_PIPELINES = {"my_spider.pipelines.JsonPipeline": 300}

bash

scrapy crawl quotes -o quotes.json   # 也可直接输出

scrapy crawl quotes -o quotes.json   # 也可直接输出

CrawlSpider（规则爬取）

python

from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor

class BookSpider(CrawlSpider):
    name = "books"
    allowed_domains = ["books.toscrape.com"]
    start_urls = ["https://books.toscrape.com/"]
    rules = (
        Rule(LinkExtractor(allow=r"catalogue/page-\d+\.html"), follow=True),
        Rule(LinkExtractor(allow=r"catalogue/.*_\d+/index\.html"), callback="parse_book"),
    )

    def parse_book(self, response):
        yield {
            "title": response.css("h1::text").get(),
            "price": response.css("p.price_color::text").get(),
        }

from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor

class BookSpider(CrawlSpider):
    name = "books"
    allowed_domains = ["books.toscrape.com"]
    start_urls = ["https://books.toscrape.com/"]
    rules = (
        Rule(LinkExtractor(allow=r"catalogue/page-\d+\.html"), follow=True),
        Rule(LinkExtractor(allow=r"catalogue/.*_\d+/index\.html"), callback="parse_book"),
    )

    def parse_book(self, response):
        yield {
            "title": response.css("h1::text").get(),
            "price": response.css("p.price_color::text").get(),
        }

分布式爬虫（Scrapy-Redis）

bash

pip install scrapy-redis

pip install scrapy-redis

python

# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://localhost:6379"

# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://localhost:6379"

多个爬虫实例共享 Redis 请求队列，实现分布式爬取。

浏览器自动化

Selenium

bash

pip install selenium

pip install selenium

需要下载 ChromeDriver 并放入 PATH。

python

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

driver = webdriver.Chrome()
driver.implicitly_wait(10)
driver.get("https://quotes.toscrape.com/js/")

# 显式等待
wait = WebDriverWait(driver, 10)
quotes = wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, "div.quote")))
for q in quotes:
    print(q.find_element(By.CSS_SELECTOR, "span.text").text)

driver.find_element(By.NAME, "search").send_keys("Python")
driver.save_screenshot("page.png")
driver.quit()

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

driver = webdriver.Chrome()
driver.implicitly_wait(10)
driver.get("https://quotes.toscrape.com/js/")

# 显式等待
wait = WebDriverWait(driver, 10)
quotes = wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, "div.quote")))
for q in quotes:
    print(q.find_element(By.CSS_SELECTOR, "span.text").text)

driver.find_element(By.NAME, "search").send_keys("Python")
driver.save_screenshot("page.png")
driver.quit()

Playwright（推荐）

bash

pip install playwright
playwright install

pip install playwright
playwright install

比 Selenium 更快更稳定，支持自动等待。

python

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto("https://quotes.toscrape.com/js/")
    # 自动等待元素出现
    quotes = page.query_selector_all("div.quote")
    for q in quotes:
        print(q.query_selector("span.text").inner_text())
    page.screenshot(path="page.png")
    browser.close()

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto("https://quotes.toscrape.com/js/")
    # 自动等待元素出现
    quotes = page.query_selector_all("div.quote")
    for q in quotes:
        print(q.query_selector("span.text").inner_text())
    page.screenshot(path="page.png")
    browser.close()

上下文与网络拦截

python

with sync_playwright() as p:
    browser = p.chromium.launch()
    # 独立上下文（隔离 Cookie、存储、UA）
    ctx = browser.new_context(
        viewport={"width": 1920, "height": 1080},
        user_agent="Mozilla/5.0 Chrome/120.0.0.0",
    )
    page = ctx.new_page()

    # 拦截图片请求以提升速度
    page.route("**/*", lambda route, req: route.abort() if req.resource_type == "image" else route.continue_())
    page.goto("https://example.com")
    browser.close()

# 录制脚本: playwright codegen https://quotes.toscrape.com

with sync_playwright() as p:
    browser = p.chromium.launch()
    # 独立上下文（隔离 Cookie、存储、UA）
    ctx = browser.new_context(
        viewport={"width": 1920, "height": 1080},
        user_agent="Mozilla/5.0 Chrome/120.0.0.0",
    )
    page = ctx.new_page()

    # 拦截图片请求以提升速度
    page.route("**/*", lambda route, req: route.abort() if req.resource_type == "image" else route.continue_())
    page.goto("https://example.com")
    browser.close()

# 录制脚本: playwright codegen https://quotes.toscrape.com

Playwright 优势：自动等待、异步支持、三大引擎、网络拦截，推荐首选。

反爬虫与应对策略

常见反爬机制

手段	说明	应对
User-Agent 检测	检查请求头	伪装常见浏览器 UA
IP 频率限制	同一 IP 大量请求被封	代理池 + 频率控制
Cookie 校验	需维持会话状态	Session 管理
JS 渲染	数据动态加载	浏览器自动化
验证码	图形/滑块验证	OCR / 打码平台
请求签名	参数加密校验	逆向分析加密逻辑

代码示例

python

# User-Agent 轮换
import random
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Chrome/120.0.0.0",
    "Mozilla/5.0 (Windows NT 10.0; rv:109.0) Gecko/20100101 Firefox/119.0",
]

# IP 代理池轮换
proxies = [
    {"http": "http://proxy1:port"},
    {"http": "http://proxy2:port"},
]
proxy = random.choice(proxies)
resp = requests.get(url, proxies=proxy, headers={"User-Agent": random.choice(USER_AGENTS)})

# 验证码 OCR
import pytesseract
from PIL import Image
code = pytesseract.image_to_string(Image.open("captcha.png"), config="--psm 8")

# Playwright 绕过 headless 检测
with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    context = browser.new_context(user_agent=random.choice(USER_AGENTS))
    page = context.new_page()
    page.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
        Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh'] });
    """)

# User-Agent 轮换
import random
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Chrome/120.0.0.0",
    "Mozilla/5.0 (Windows NT 10.0; rv:109.0) Gecko/20100101 Firefox/119.0",
]

# IP 代理池轮换
proxies = [
    {"http": "http://proxy1:port"},
    {"http": "http://proxy2:port"},
]
proxy = random.choice(proxies)
resp = requests.get(url, proxies=proxy, headers={"User-Agent": random.choice(USER_AGENTS)})

# 验证码 OCR
import pytesseract
from PIL import Image
code = pytesseract.image_to_string(Image.open("captcha.png"), config="--psm 8")

# Playwright 绕过 headless 检测
with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    context = browser.new_context(user_agent=random.choice(USER_AGENTS))
    page = context.new_page()
    page.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
        Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh'] });
    """)

数据存储

python

# CSV
import csv
with open("data.csv", "w", newline="", encoding="utf-8-sig") as f:
    w = csv.DictWriter(f, fieldnames=["title", "url"])
    w.writeheader()
    w.writerows([{"title": "文章1", "url": "https://ex.com/1"}])

# JSON
import json
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# SQLite
import sqlite3
conn = sqlite3.connect("data.db")
conn.execute("CREATE TABLE IF NOT EXISTS articles (title TEXT, url TEXT UNIQUE)")
conn.execute("INSERT OR IGNORE INTO articles VALUES (?, ?)", ("文章1", "https://ex.com/1"))
conn.commit()
conn.close()

# 异步写入 aiofiles
import asyncio, aiofiles
async def save_async(data):
    async with aiofiles.open("data.jsonl", "a", encoding="utf-8") as f:
        await f.write(json.dumps(data, ensure_ascii=False) + "\n")

# CSV
import csv
with open("data.csv", "w", newline="", encoding="utf-8-sig") as f:
    w = csv.DictWriter(f, fieldnames=["title", "url"])
    w.writeheader()
    w.writerows([{"title": "文章1", "url": "https://ex.com/1"}])

# JSON
import json
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# SQLite
import sqlite3
conn = sqlite3.connect("data.db")
conn.execute("CREATE TABLE IF NOT EXISTS articles (title TEXT, url TEXT UNIQUE)")
conn.execute("INSERT OR IGNORE INTO articles VALUES (?, ?)", ("文章1", "https://ex.com/1"))
conn.commit()
conn.close()

# 异步写入 aiofiles
import asyncio, aiofiles
async def save_async(data):
    async with aiofiles.open("data.jsonl", "a", encoding="utf-8") as f:
        await f.write(json.dumps(data, ensure_ascii=False) + "\n")

实战示例

目标：爬取 Quotes to Scrape 的名言列表。

requests + BeautifulSoup

python

import requests
from bs4 import BeautifulSoup
import csv
import time
import random

def fetch_page(url, session):
    try:
        resp = session.get(url, timeout=10)
        resp.raise_for_status()
    except requests.RequestException:
        return [], None
    soup = BeautifulSoup(resp.text, "lxml")
    quotes = []
    for q in soup.select("div.quote"):
        quotes.append({
            "text": q.select_one("span.text").get_text(strip=True),
            "author": q.select_one("small.author").get_text(strip=True),
            "tags": ", ".join(t.get_text(strip=True) for t in q.select("a.tag")),
        })
    btn = soup.select_one("li.next a")
    next_url = "http://quotes.toscrape.com" + btn["href"] if btn else None
    return quotes, next_url

session = requests.Session()
session.headers.update({"User-Agent": "Mozilla/5.0 Chrome/120.0.0.0"})
url = "http://quotes.toscrape.com/page/1/"
all_quotes = []
for page in range(1, 6):
    if not url: break
    print(f"第 {page} 页...")
    quotes, url = fetch_page(url, session)
    all_quotes.extend(quotes)
    time.sleep(random.uniform(0.5, 1.5))

with open("quotes.csv", "w", newline="", encoding="utf-8-sig") as f:
    w = csv.DictWriter(f, fieldnames=["text", "author", "tags"])
    w.writeheader(); w.writerows(all_quotes)
print(f"完成！共 {len(all_quotes)} 条")

import requests
from bs4 import BeautifulSoup
import csv
import time
import random

def fetch_page(url, session):
    try:
        resp = session.get(url, timeout=10)
        resp.raise_for_status()
    except requests.RequestException:
        return [], None
    soup = BeautifulSoup(resp.text, "lxml")
    quotes = []
    for q in soup.select("div.quote"):
        quotes.append({
            "text": q.select_one("span.text").get_text(strip=True),
            "author": q.select_one("small.author").get_text(strip=True),
            "tags": ", ".join(t.get_text(strip=True) for t in q.select("a.tag")),
        })
    btn = soup.select_one("li.next a")
    next_url = "http://quotes.toscrape.com" + btn["href"] if btn else None
    return quotes, next_url

session = requests.Session()
session.headers.update({"User-Agent": "Mozilla/5.0 Chrome/120.0.0.0"})
url = "http://quotes.toscrape.com/page/1/"
all_quotes = []
for page in range(1, 6):
    if not url: break
    print(f"第 {page} 页...")
    quotes, url = fetch_page(url, session)
    all_quotes.extend(quotes)
    time.sleep(random.uniform(0.5, 1.5))

with open("quotes.csv", "w", newline="", encoding="utf-8-sig") as f:
    w = csv.DictWriter(f, fieldnames=["text", "author", "tags"])
    w.writeheader(); w.writerows(all_quotes)
print(f"完成！共 {len(all_quotes)} 条")

Playwright 版本（处理 JS 渲染）

python

import asyncio, csv
from playwright.async_api import async_playwright

async def main():
    all_quotes = []
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        url = "https://quotes.toscrape.com/js/"
        for i in range(1, 6):
            await page.goto(url, wait_until="networkidle")
            await page.wait_for_selector("div.quote")
            for q in await page.query_selector_all("div.quote"):
                all_quotes.append({
                    "text": await q.query_selector("span.text").inner_text(),
                    "author": await q.query_selector("small.author").inner_text(),
                    "tags": ", ".join([await t.inner_text() for t in await q.query_selector_all("a.tag")]),
                })
            btn = await page.query_selector("li.next a")
            url = "https://quotes.toscrape.com/js/" + await btn.get_attribute("href") if btn else None
            if not url: break
            await asyncio.sleep(1)
        await browser.close()
    with open("quotes_js.csv", "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=["text", "author", "tags"])
        w.writeheader(); w.writerows(all_quotes)
    print(f"完成！共 {len(all_quotes)} 条")
asyncio.run(main())

import asyncio, csv
from playwright.async_api import async_playwright

async def main():
    all_quotes = []
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        url = "https://quotes.toscrape.com/js/"
        for i in range(1, 6):
            await page.goto(url, wait_until="networkidle")
            await page.wait_for_selector("div.quote")
            for q in await page.query_selector_all("div.quote"):
                all_quotes.append({
                    "text": await q.query_selector("span.text").inner_text(),
                    "author": await q.query_selector("small.author").inner_text(),
                    "tags": ", ".join([await t.inner_text() for t in await q.query_selector_all("a.tag")]),
                })
            btn = await page.query_selector("li.next a")
            url = "https://quotes.toscrape.com/js/" + await btn.get_attribute("href") if btn else None
            if not url: break
            await asyncio.sleep(1)
        await browser.close()
    with open("quotes_js.csv", "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=["text", "author", "tags"])
        w.writeheader(); w.writerows(all_quotes)
    print(f"完成！共 {len(all_quotes)} 条")
asyncio.run(main())

反爬虫策略（最佳实践）

python

import time, random
from functools import wraps

# 频率控制器
class RateLimiter:
    def __init__(self, cps=2):
        self.min_interval = 1.0 / cps
        self.last_call = 0.0
    def wait(self):
        elapsed = time.time() - self.last_call
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_call = time.time()

# 重试装饰器
def retry(max_retries=3, delay=1, backoff=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            _d = delay
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == max_retries - 1:
                        raise
                    print(f"重试 {i+1}: {e}")
                    time.sleep(_d)
                    _d *= backoff
        return wrapper
    return decorator

# 爬虫配置模板
CRAWLER_CONFIG = {
    "headers": {
        "User-Agent": "Mozilla/5.0 Chrome/120.0.0.0",
        "Accept-Language": "zh-CN,zh;q=0.9",
    },
    "timeout": (5, 15),
    "max_retries": 3,
    "calls_per_second": 2,
    "respect_robots_txt": True,
}

import time, random
from functools import wraps

# 频率控制器
class RateLimiter:
    def __init__(self, cps=2):
        self.min_interval = 1.0 / cps
        self.last_call = 0.0
    def wait(self):
        elapsed = time.time() - self.last_call
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_call = time.time()

# 重试装饰器
def retry(max_retries=3, delay=1, backoff=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            _d = delay
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == max_retries - 1:
                        raise
                    print(f"重试 {i+1}: {e}")
                    time.sleep(_d)
                    _d *= backoff
        return wrapper
    return decorator

# 爬虫配置模板
CRAWLER_CONFIG = {
    "headers": {
        "User-Agent": "Mozilla/5.0 Chrome/120.0.0.0",
        "Accept-Language": "zh-CN,zh;q=0.9",
    },
    "timeout": (5, 15),
    "max_retries": 3,
    "calls_per_second": 2,
    "respect_robots_txt": True,
}

总结

技术选型

场景	推荐方案
静态页面	requests + BeautifulSoup
异步高并发	httpx / aiohttp
大型爬虫项目	Scrapy
JS 动态渲染	Playwright
分布式爬取	Scrapy + Scrapy-Redis

核心原则

合规优先：遵守 robots.txt 和相关法律法规
频率控制：合理控制请求速度，不造成服务器负担
异常处理：完善的错误处理和重试机制
数据质量：清洗和验证爬取的数据
效率平衡：在速度和稳定性之间找到平衡

Python 爬虫 ​

爬虫基础概念 ​

什么是爬虫 ​

爬虫的分类 ​

法律与道德 ​

网络请求 ​

requests 库（重点） ​

httpx（现代替代） ​

HTML 解析 ​

BeautifulSoup ​

四大对象 ​

find / find_all / CSS 选择器 ​

lxml 与 XPath ​

Scrapy 框架 ​

架构 ​

创建项目 ​

Spider 编写 ​

Item 与 Pipeline ​

CrawlSpider（规则爬取） ​

分布式爬虫（Scrapy-Redis） ​

浏览器自动化 ​

Selenium ​

Playwright（推荐） ​

上下文与网络拦截 ​

反爬虫与应对策略 ​

常见反爬机制 ​

代码示例 ​

数据存储 ​

实战示例 ​

requests + BeautifulSoup ​

Playwright 版本（处理 JS 渲染） ​

反爬虫策略（最佳实践） ​

总结 ​

技术选型 ​

核心原则 ​

Python 爬虫

爬虫基础概念

什么是爬虫

爬虫的分类

法律与道德

网络请求

requests 库（重点）

httpx（现代替代）

HTML 解析

BeautifulSoup

四大对象

find / find_all / CSS 选择器

lxml 与 XPath

Scrapy 框架

架构

创建项目

Spider 编写

Item 与 Pipeline

CrawlSpider（规则爬取）

分布式爬虫（Scrapy-Redis）

浏览器自动化

Selenium

Playwright（推荐）

上下文与网络拦截

反爬虫与应对策略

常见反爬机制

代码示例

数据存储

实战示例

requests + BeautifulSoup

Playwright 版本（处理 JS 渲染）

反爬虫策略（最佳实践）

总结

技术选型

核心原则