爬虫

先导内容

Python

0. 爬虫概述

定义

网络爬虫是自动请求网页或接口、提取目标数据并保存的程序。核心不是“绕过限制”，而是稳定、合法、可维护地获取公开或授权数据。

0.1 基本流程

确定目标数据
  -> 分析页面或接口
  -> 构造请求
  -> 下载响应
  -> 解析数据
  -> 清洗数据
  -> 存储数据
  -> 增量更新和异常重试

0.2 合规原则

合规注意

• 遵守网站服务条款、版权要求和 robots.txt。
• 不采集隐私数据、账号数据、付费内容或需要越权访问的数据。
• 控制请求频率，避免影响对方服务。
• 需要登录或 API Token 时，优先使用官方 API。

---

1. HTTP 基础

1.1 请求和响应

概念	说明
URL	资源地址
Method	请求方法，如 GET、POST
Headers	请求头，描述客户端、认证、内容类型等
Cookies	会话状态
Query Params	URL 查询参数
Body	请求体，常见于 POST、PUT
Status Code	响应状态码

1.2 常见状态码

状态码	含义	处理建议
200	成功	正常解析
301/302	重定向	检查最终 URL
400	请求参数错误	检查参数和请求体
401/403	未认证或无权限	检查登录态、权限和合规性
404	资源不存在	记录并跳过
429	请求过快	降低频率、指数退避
500/502/503	服务端错误	重试或稍后再抓

1.3 请求头

headers = {
    "User-Agent": "Mozilla/5.0",
    "Accept": "text/html,application/xhtml+xml",
    "Accept-Language": "zh-CN,zh;q=0.9",
}

请求头建议

User-Agent 只能说明客户端类型，不应伪装成浏览器后进行高频访问。稳定爬取更依赖低频率、缓存、重试和官方接口。

---

2. 使用 requests

2.1 GET 请求

import requests
 
url = "https://example.com"
 
response = requests.get(
    url,
    headers={"User-Agent": "Mozilla/5.0"},
    timeout=10,                  # 必须设置超时，避免程序永久卡住
)
 
response.raise_for_status()      # 非 2xx/3xx 状态码直接抛异常
response.encoding = response.apparent_encoding
 
print(response.text)

2.2 查询参数

params = {
    "q": "python",
    "page": 1,
}
 
response = requests.get("https://example.com/search", params=params, timeout=10)
print(response.url)              # requests 会自动拼接和编码参数

2.3 POST JSON

payload = {"username": "alice", "password": "secret"}
 
response = requests.post(
    "https://example.com/api/login",
    json=payload,                # 自动序列化为 JSON，并设置 Content-Type
    timeout=10,
)
data = response.json()

2.4 Session

import requests
 
session = requests.Session()
session.headers.update({"User-Agent": "Mozilla/5.0"})
 
login_resp = session.post(
    "https://example.com/login",
    data={"username": "alice", "password": "secret"},
    timeout=10,
)
login_resp.raise_for_status()
 
profile_resp = session.get("https://example.com/profile", timeout=10)
print(profile_resp.text)

Session 的作用

Session 会复用 TCP 连接并保存 Cookie，适合需要登录态或连续请求的场景。

2.5 重试和退避

from time import sleep
import requests
 
 
def get_with_retry(url: str, retries: int = 3) -> requests.Response:
    last_error = None
    for attempt in range(retries):
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 429:
                # 429 表示请求过快，等待时间逐步增加
                sleep(2 ** attempt)
                continue
            response.raise_for_status()
            return response
        except requests.RequestException as exc:
            last_error = exc
            sleep(2 ** attempt)
    raise RuntimeError(f"请求失败：{url}") from last_error

---

3. 解析 HTML

3.1 BeautifulSoup

from bs4 import BeautifulSoup
 
html = """
<html>
  <body>
    <a class="item" href="/p/1">文章 1</a>
    <a class="item" href="/p/2">文章 2</a>
  </body>
</html>
"""
 
soup = BeautifulSoup(html, "html.parser")
 
for link in soup.select("a.item"):
    title = link.get_text(strip=True)
    href = link.get("href")
    print(title, href)

常用选择器

选择器	含义
div	选择所有 div
.item	选择 class="item"
#main	选择 id="main"
ul li	选择 ul 下所有 li
a[href]	选择带 href 属性的 a

3.2 lxml XPath

from lxml import etree
 
tree = etree.HTML(html)
 
titles = tree.xpath("//a[@class='item']/text()")
hrefs = tree.xpath("//a[@class='item']/@href")
 
for title, href in zip(titles, hrefs):
    print(title.strip(), href)

解析器选择

页面结构简单时用 BeautifulSoup 更直观；需要精确定位或性能更高时可以使用 lxml 和 XPath。

3.3 正则只做辅助

import re
 
text = "价格：￥199.00"
price = re.search(r"￥(\d+(?:\.\d+)?)", text)
if price:
    print(float(price.group(1)))

不要用正则解析完整 HTML

HTML 结构嵌套复杂，正则容易漏匹配或错匹配。正则适合从已提取文本中再抽取手机号、价格、日期等小片段。

---

4. 解析 JSON 接口

import requests
 
response = requests.get("https://example.com/api/articles", timeout=10)
response.raise_for_status()
 
payload = response.json()
 
for item in payload["data"]:
    print(item["id"], item["title"])

4.1 从浏览器分析接口

分析步骤

1. 打开浏览器开发者工具。
2. 切到 Network 面板。
3. 刷新页面或触发加载更多。
4. 过滤 Fetch/XHR。
5. 查看请求 URL、参数、Headers、响应 JSON。

4.2 分页接口

import requests
 
all_items = []
 
for page in range(1, 6):
    response = requests.get(
        "https://example.com/api/articles",
        params={"page": page, "page_size": 20},
        timeout=10,
    )
    response.raise_for_status()
    data = response.json()
    items = data.get("data", [])
    if not items:
        break
    all_items.extend(items)
 
print(len(all_items))

---

5. 数据保存

5.1 CSV

import csv
 
rows = [
    {"title": "文章 1", "url": "https://example.com/1"},
    {"title": "文章 2", "url": "https://example.com/2"},
]
 
with open("articles.csv", "w", newline="", encoding="utf-8-sig") as file:
    writer = csv.DictWriter(file, fieldnames=["title", "url"])
    writer.writeheader()
    writer.writerows(rows)

5.2 JSON Lines

import json
 
with open("articles.jsonl", "w", encoding="utf-8") as file:
    for row in rows:
        # 一行一个 JSON，适合追加写入和大文件流式处理
        file.write(json.dumps(row, ensure_ascii=False) + "\n")

5.3 SQLite

import sqlite3
 
conn = sqlite3.connect("crawler.db")
cur = conn.cursor()
 
cur.execute(
    """
    create table if not exists article (
        id integer primary key autoincrement,
        title text not null,
        url text not null unique
    )
    """
)
 
cur.executemany(
    "insert or ignore into article(title, url) values (?, ?)",
    [(row["title"], row["url"]) for row in rows],
)
 
conn.commit()
conn.close()

---

6. 反爬和稳定性

6.1 限速

from random import uniform
from time import sleep
 
for url in urls:
    response = requests.get(url, timeout=10)
    # 随机间隔能降低请求尖峰，但不能替代合规授权
    sleep(uniform(1.0, 3.0))

6.2 去重

seen: set[str] = set()
 
for url in urls:
    if url in seen:
        continue
    seen.add(url)
    # 请求和解析逻辑

6.3 增量更新

latest_id = load_latest_id_from_db()
 
for item in fetch_items():
    if item["id"] <= latest_id:
        break
    save_item(item)

6.4 常见问题

问题	原因	处理
乱码	编码判断错误	设置 response.encoding
超时	网络慢或服务器不响应	设置 timeout 和重试
数据为空	选择器失效或页面 JS 渲染	检查 HTML 和 Network
403	权限、登录态或访问策略	使用授权 API 或停止抓取
429	请求频率过高	限速、退避、缓存

---

7. 异步爬虫

使用场景

异步适合大量 IO 等待的接口抓取。它不会让单个请求变快，但能在等待响应时处理其他请求。

import asyncio
import httpx
 
 
async def fetch(client: httpx.AsyncClient, url: str) -> str:
    response = await client.get(url)
    response.raise_for_status()
    return response.text
 
 
async def main() -> None:
    urls = ["https://example.com"] * 10
    limits = httpx.Limits(max_connections=5)
 
    async with httpx.AsyncClient(timeout=10, limits=limits) as client:
        tasks = [fetch(client, url) for url in urls]
        pages = await asyncio.gather(*tasks)
        print(len(pages))
 
 
asyncio.run(main())

并发控制

并发数不是越大越好。过高并发会导致本机连接耗尽、对方服务压力升高，也更容易触发限制。

---

8. Scrapy

适用场景

Scrapy 适合中大型爬虫项目：内置调度、去重、并发、管道、重试、中间件和日志。

8.1 创建项目

python -m pip install scrapy
scrapy startproject demo_spider
cd demo_spider
scrapy genspider example example.com

8.2 Spider 示例

import scrapy
 
 
class ExampleSpider(scrapy.Spider):
    name = "example"
    allowed_domains = ["example.com"]
    start_urls = ["https://example.com"]
 
    def parse(self, response):
        for article in response.css("a.item"):
            yield {
                "title": article.css("::text").get(default="").strip(),
                "url": response.urljoin(article.css("::attr(href)").get()),
            }
 
        next_page = response.css("a.next::attr(href)").get()
        if next_page:
            # response.follow 会自动处理相对路径
            yield response.follow(next_page, callback=self.parse)

8.3 常用配置

# settings.py
ROBOTSTXT_OBEY = True
DOWNLOAD_DELAY = 1.0
CONCURRENT_REQUESTS = 8
RETRY_TIMES = 3
FEED_EXPORT_ENCODING = "utf-8"

---

9. 浏览器自动化

使用场景

如果数据必须通过浏览器执行 JavaScript 后才出现，可以使用 Playwright 或 Selenium。但浏览器自动化成本高、速度慢，应优先寻找公开接口或服务端渲染数据。

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://example.com", wait_until="networkidle")
 
    titles = page.locator("a.item").all_text_contents()
    print(titles)
 
    browser.close()

---

10. 项目结构和最佳实践

10.1 目录结构

crawler-project/
  crawler/
    __init__.py
    fetcher.py     # 请求、重试、限速
    parser.py      # HTML/JSON 解析
    storage.py     # CSV/DB 保存
    models.py      # 数据结构
  tests/
  data/
  pyproject.toml
  README.md

10.2 可维护性建议

• 请求、解析、存储分层，不要都写在一个函数里。
• 所有请求设置 timeout，失败要记录日志。
• 解析逻辑写单元测试，可保存一份 HTML 样本作为 fixture。
• 数据字段统一命名，保存前做类型转换和空值处理。
• 使用配置文件管理 URL、并发数、延迟、输出路径。
• 记录抓取时间、来源 URL 和版本，便于追溯。

10.3 最小完整示例

from dataclasses import dataclass
from pathlib import Path
import csv
import requests
from bs4 import BeautifulSoup
 
 
@dataclass
class Article:
    title: str
    url: str
 
 
def fetch(url: str) -> str:
    response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)
    response.raise_for_status()
    response.encoding = response.apparent_encoding
    return response.text
 
 
def parse(html: str, base_url: str) -> list[Article]:
    soup = BeautifulSoup(html, "html.parser")
    articles: list[Article] = []
 
    for link in soup.select("a.item"):
        title = link.get_text(strip=True)
        href = link.get("href")
        if not title or not href:
            continue
        # requests.compat.urljoin 用于把相对路径转为绝对路径
        url = requests.compat.urljoin(base_url, href)
        articles.append(Article(title=title, url=url))
 
    return articles
 
 
def save_csv(path: Path, articles: list[Article]) -> None:
    with path.open("w", newline="", encoding="utf-8-sig") as file:
        writer = csv.DictWriter(file, fieldnames=["title", "url"])
        writer.writeheader()
        for article in articles:
            writer.writerow({"title": article.title, "url": article.url})
 
 
def main() -> None:
    url = "https://example.com"
    html = fetch(url)
    articles = parse(html, url)
    save_csv(Path("articles.csv"), articles)
 
 
if __name__ == "__main__":
    main()