跳到主要内容
预计阅读 39 分钟

11｜HTTP请求与HTML解析：结构化Web数据采集

一个数据分析师的困境

某数据分析师每周需要从三个公开信息网站手动收集行业数据，整理成表格供团队使用。每次操作需要 4 个小时：打开网页、逐条复制、粘贴到表格、核对格式。如果数据源增加到十个怎么办？如果采集频率从每周变成每天呢？

这就是 Web 数据采集要解决的问题：用程序代替人类完成”访问网页 -> 定位数据 -> 提取内容 -> 保存结果”的全流程。本章将从底层的 HTTP 通信协议讲起，逐步构建一个完整的数据采集管线。

第一部分：HTTP 通信原理

在编写数据采集程序之前，有必要理解一件基础的事：浏览器是如何获取网页内容的？

请求-响应模型

浏览器与服务器之间的交互遵循严格的请求-响应协议：

 客户端 (浏览器/程序)               服务器
        |                            |
        |  --- HTTP Request -------> |  "请给我 /products 页面"
        |                            |  服务器查找并组装页面内容
        | <--- HTTP Response ------- |  "状态 200，页面内容如下..."
        |                            |
        |  客户端解析并呈现收到的内容   |

数据采集程序所做的事情就是模拟客户端的角色：发送 HTTP 请求，接收响应，然后从响应内容中提取结构化数据。

请求方法

HTTP 定义了多种请求方法，数据采集中最常用的两种：

  • GET：请求获取资源。在浏览器地址栏输入 URL 后触发的就是 GET 请求。
  • POST：向服务器提交数据。登录表单、搜索查询等场景使用 POST。

响应状态码

服务器通过状态码告知请求的处理结果：

状态码语义应对策略
200请求成功正常解析响应内容
301/302资源已迁移跟随重定向地址
403拒绝访问检查请求头配置或访问权限
404资源不存在检查 URL 拼写
429请求频率超限降低请求速率
500服务器内部故障稍后重试

第二部分：用 requests 发起 HTTP 请求

requests 库是 Python 生态中使用最广泛的 HTTP 客户端库，其 API 设计以人类可读性为首要目标。

在发送任何请求之前：请先访问目标站点的 robots.txt（如 https://example.com/robots.txt），确认你要采集的路径是否被允许。这是负责任的数据采集的第一步。详见本章第八部分。

安装

pip install requests

GET 请求基础

import requests

# 发送 GET 请求
resp = requests.get("https://httpbin.org/get")

# 检查状态码
print(f"Status: {resp.status_code}")  # 200

# 获取文本响应
print(resp.text[:200])

# 如果响应是 JSON 格式，直接解析为 Python 字典
payload = resp.json()
print(payload["origin"])  # 你的公网 IP

健壮的请求处理

生产级代码必须处理网络异常、超时和服务器错误：

import requests

target_url = "https://example.com/api/data"
try:
    resp = requests.get(target_url, timeout=15)
    resp.raise_for_status()  # 状态码为 4xx/5xx 时自动抛出异常
    print(f"Received {len(resp.text)} characters")
except requests.exceptions.Timeout:
    print("Request timed out")
except requests.exceptions.HTTPError as err:
    print(f"HTTP error: {err}")
except requests.exceptions.ConnectionError:
    print("Connection failed - check network")
except requests.exceptions.RequestException as err:
    print(f"Request failed: {err}")

自定义请求头与查询参数

许多服务器会检查请求头中的 User-Agent 字段。如果识别到非浏览器客户端，可能返回简化页面或拒绝响应。设置合理的请求头可以获得与浏览器一致的响应内容。

import requests

query_params = {"category": "electronics", "page": 3}

request_headers = {
    "User-Agent": "DataCollector/1.0 (educational project)",
    "Accept-Language": "en-US,en;q=0.9"
}

resp = requests.get("https://httpbin.org/get",
                    params=query_params,
                    headers=request_headers)
print(resp.url)  # 查看构造后的完整 URL

POST 请求

import requests

# 提交表单数据
form_payload = {"account": "test_user", "token": "abc123"}
resp = requests.post("https://httpbin.org/post", data=form_payload)
print(resp.json()["form"])

# 提交 JSON 数据
json_payload = {"query": "SELECT * FROM products", "limit": 100}
resp = requests.post("https://httpbin.org/post", json=json_payload)
print(resp.json()["json"])

文件下载

对于二进制文件（图片、PDF 等），使用流式传输避免大文件占满内存：

import requests

download_url = "https://example.com/report.pdf"
resp = requests.get(download_url, stream=True)
resp.raise_for_status()

with open("local_report.pdf", "wb") as out_file:
    for block in resp.iter_content(chunk_size=8192):
        out_file.write(block)
print("Download complete")

第三部分：用 BeautifulSoup 解析 HTML

HTTP 请求返回的是原始 HTML 源码——一堆嵌套的标签和属性。BeautifulSoup 库能将这些源码解析为可查询的树形结构，让你可以像查询数据库一样定位页面元素。

安装

pip install beautifulsoup4

基本解析

from bs4 import BeautifulSoup

page_source = """
<html>
<head><title>Tech Gadgets Store</title></head>
<body>
    <h1 id="banner">New Arrivals</h1>
    <div class="product-grid">
        <div class="product-card">
            <h3 class="product-name">Wireless Earbuds Pro</h3>
            <span class="product-price">$129.99</span>
            <a href="/item/101">View Details</a>
        </div>
        <div class="product-card">
            <h3 class="product-name">Smart Fitness Band</h3>
            <span class="product-price">$79.50</span>
            <a href="/item/102">View Details</a>
        </div>
    </div>
</body>
</html>
"""

doc = BeautifulSoup(page_source, "html.parser")

# 获取页面标题
print(doc.title.text)  # Tech Gadgets Store

# 获取 h1 标签的文本
print(doc.h1.text)  # New Arrivals

第四部分：CSS 选择器定位元素

CSS 选择器是定位 HTML 元素最灵活的方式。BeautifulSoup 的 select() 返回所有匹配元素的列表，select_one() 返回第一个匹配元素。

选择器语法速查

选择器形式语义示例
tag按标签名doc.select("p")
#id_value按 ID 属性doc.select("#banner")
.class_name按 class 属性doc.select(".product-card")
ancestor descendant后代元素doc.select(".product-grid .product-name")
parent > child直接子元素doc.select("div > h3")
[attr]包含某属性doc.select("a[href]")

数据提取实战

from bs4 import BeautifulSoup

# 使用前面定义的 doc 对象
cards = doc.select(".product-card")
print(f"Found {len(cards)} products")

for card in cards:
    name = card.select_one(".product-name").text
    price = card.select_one(".product-price").text
    detail_link = card.select_one("a").get("href")
    print(f"Product: {name} | Price: {price} | Link: {detail_link}")

# 输出：
# Product: Wireless Earbuds Pro | Price: $129.99 | Link: /item/101
# Product: Smart Fitness Band | Price: $79.50 | Link: /item/102

数据提取的两种方式：

  • .text 或 .get_text()：获取标签内部的纯文本
  • .get("属性名")：获取标签的属性值（如 href、src、class）
  • .attrs：以字典形式返回标签的所有属性

第五部分：多页数据采集

真实网站的数据通常分布在多个页面上。自动翻页的核心思路是：观察页码参数在 URL 中的变化规律，然后用循环构造每一页的 URL。

import requests
from bs4 import BeautifulSoup
import time

def collect_paginated_data(base_endpoint, total_pages=5):
    """依次采集多页数据"""
    collected = []
    http_headers = {
        "User-Agent": "DataCollector/1.0 (educational project)"
    }

    for pg in range(1, total_pages + 1):
        page_url = f"{base_endpoint}?page={pg}"
        print(f"Fetching page {pg}: {page_url}")

        try:
            resp = requests.get(page_url, headers=http_headers, timeout=12)
            resp.raise_for_status()
        except requests.exceptions.RequestException as err:
            print(f"  Page {pg} failed: {err}")
            continue

        doc = BeautifulSoup(resp.text, "html.parser")

        entries = doc.select(".entry")
        if not entries:
            print(f"  No entries on page {pg}, stopping.")
            break

        for entry in entries:
            heading = entry.select_one(".heading")
            if heading:
                collected.append(heading.text.strip())

        print(f"  Collected {len(entries)} entries")

        # 控制请求频率
        time.sleep(1.5)

    return collected

关于翻页 URL 的多样性： 不同网站的分页实现差异很大。有的用 ?page=2，有的用路径 /page/2，有的用偏移量 ?offset=20。需要在浏览器中手动翻页，观察 URL 的变化模式。

第六部分：数据持久化

保存为 CSV

import csv

def persist_as_csv(records, output_path):
    """将字典列表写入 CSV 文件"""
    if not records:
        print("No data to save")
        return

    with open(output_path, "w", newline="", encoding="utf-8-sig") as fh:
        writer = csv.DictWriter(fh, fieldnames=records[0].keys())
        writer.writeheader()
        writer.writerows(records)
    print(f"Saved {len(records)} records to {output_path}")

保存为 JSON

import json

def persist_as_json(records, output_path):
    """将数据写入 JSON 文件"""
    with open(output_path, "w", encoding="utf-8") as fh:
        json.dump(records, fh, ensure_ascii=False, indent=2)
    print(f"Saved to {output_path}")

第七部分：综合实战——公开书店数据采集

下面以 books.toscrape.com（一个专为数据采集练习搭建的模拟书店网站）为目标，构建一个完整的采集与分析管线。该网站允许自由采集，共有 1000 本书分布在 50 个分页中。

"""
公开书店数据采集管线
====================
目标站点：books.toscrape.com（允许采集的练习站点）
采集内容：书名、定价、星级评分、库存状态
输出：CSV 数据文件 + JSON 数据文件 + 控制台统计
"""
import requests
from bs4 import BeautifulSoup
import csv
import json
import time


def fetch_catalog(page_limit=3):
    """按页采集书籍目录"""
    endpoint_tpl = "https://books.toscrape.com/catalogue/page-{}.html"
    catalog = []

    star_mapping = {
        "One": 1, "Two": 2, "Three": 3, "Four": 4, "Five": 5
    }

    http_config = {
        "headers": {"User-Agent": "DataCollector/1.0 (educational project)"},
        "timeout": 12,
    }

    for pg_num in range(1, page_limit + 1):
        pg_url = endpoint_tpl.format(pg_num)
        print(f"Fetching page {pg_num}...")

        try:
            resp = requests.get(pg_url, **http_config)
            resp.raise_for_status()
        except requests.exceptions.RequestException as err:
            print(f"  Failed: {err}")
            continue

        doc = BeautifulSoup(resp.text, "html.parser")
        product_nodes = doc.select("article.product_pod")

        for node in product_nodes:
            # 书名：在 h3 > a 的 title 属性中
            book_title = node.select_one("h3 a")["title"]

            # 定价
            book_price = node.select_one(".price_color").text

            # 星级：通过 CSS class 判断
            star_element = node.select_one(".star-rating")
            star_classes = star_element.get("class")  # ['star-rating', 'Four']
            star_count = star_mapping.get(star_classes[1], 0)

            # 库存状态
            stock_status = node.select_one(".availability").text.strip()

            catalog.append({
                "title": book_title,
                "price": book_price,
                "rating": star_count,
                "stock": stock_status,
            })

        print(f"  {len(product_nodes)} books collected")
        time.sleep(1)

    return catalog


def analyze_and_export(catalog):
    """分析采集数据并导出"""
    if not catalog:
        print("Empty catalog, nothing to export")
        return

    # 控制台预览
    print(f"\nTotal: {len(catalog)} books")
    print("-" * 65)
    for book in catalog[:5]:
        print(f"  [{book['rating']} stars] {book['title'][:40]}")
        print(f"           {book['price']}  |  {book['stock']}")
    remaining = len(catalog) - 5
    if remaining > 0:
        print(f"  ... and {remaining} more")
    print("-" * 65)

    # 导出 CSV
    csv_output = "catalog_export.csv"
    with open(csv_output, "w", newline="", encoding="utf-8-sig") as fh:
        writer = csv.DictWriter(fh, fieldnames=catalog[0].keys())
        writer.writeheader()
        writer.writerows(catalog)
    print(f"CSV exported: {csv_output}")

    # 导出 JSON
    json_output = "catalog_export.json"
    with open(json_output, "w", encoding="utf-8") as fh:
        json.dump(catalog, fh, ensure_ascii=False, indent=2)
    print(f"JSON exported: {json_output}")

    # 统计分析
    ratings = [b["rating"] for b in catalog]
    mean_rating = sum(ratings) / len(ratings)
    top_rated = sum(1 for r in ratings if r == 5)
    above_avg = sum(1 for r in ratings if r >= 4)

    print(f"\nAnalysis:")
    print(f"  Mean rating: {mean_rating:.1f} stars")
    print(f"  5-star books: {top_rated} ({top_rated/len(catalog)*100:.0f}%)")
    print(f"  4+ star books: {above_avg} ({above_avg/len(catalog)*100:.0f}%)")


def main():
    print("=" * 65)
    print("  Book Catalog Data Collection Pipeline")
    print("=" * 65)

    catalog = fetch_catalog(page_limit=3)
    analyze_and_export(catalog)


if __name__ == "__main__":
    main()

预期输出

=================================================================
  Book Catalog Data Collection Pipeline
=================================================================
Fetching page 1...
  20 books collected
Fetching page 2...
  20 books collected
Fetching page 3...
  20 books collected

Total: 60 books
-----------------------------------------------------------------
  [3 stars] A Light in the Attic
           £51.77  |  In stock
  [1 stars] Tipping the Velvet
           £53.74  |  In stock
  [1 stars] Soumission
           £50.10  |  In stock
  [4 stars] Sharp Objects
           £47.82  |  In stock
  [5 stars] Sapiens: A Brief History of Humankind
           £54.23  |  In stock
  ... and 55 more
-----------------------------------------------------------------
CSV exported: catalog_export.csv
JSON exported: catalog_export.json

Analysis:
  Mean rating: 3.0 stars
  5-star books: 12 (20%)
  4+ star books: 24 (40%)

第八部分：采集规范与 robots.txt

技术能力不等于行为许可。负责任的数据采集需要遵守一套行业规范。

robots.txt 协议

绝大多数网站在根目录下放置 robots.txt 文件，声明哪些路径允许采集、哪些路径禁止访问。

import requests

robots_resp = requests.get("https://www.example.com/robots.txt")
print(robots_resp.text)

典型内容：

User-agent: *
Disallow: /internal/
Disallow: /api/private/
Allow: /api/public/
Crawl-delay: 3
Sitemap: https://www.example.com/sitemap.xml

robots.txt 不是技术强制措施——程序可以无视它。但遵守它是互联网社区的基本行为准则。

采集行为准则

  1. 遵守 robots.txt 声明：被禁止的路径不采集。
  2. 控制请求频率：每次请求间至少间隔 1-2 秒，避免对目标服务器造成负荷。
  3. 标识请求来源：在 User-Agent 中如实说明身份和用途。
  4. 不触碰隐私数据：个人身份信息（证件号、通信地址等）绝对不采集。
  5. 尊重知识产权：采集到的内容版权归原站点所有，个人研究可用，商业使用须获得授权。
  6. 监控影响：如果目标站点响应变慢，立即停止采集。

请求间隔与退避策略

import time
import random

# 固定间隔
for pg in range(1, 20):
    resp = requests.get(f"https://example.com/items?page={pg}")
    time.sleep(2)

# 随机间隔——模式更自然
for pg in range(1, 20):
    resp = requests.get(f"https://example.com/items?page={pg}")
    time.sleep(random.uniform(1.5, 4.0))

指数退避重试

网络请求不可能 100% 成功。指数退避（exponential backoff）是处理暂时性故障的标准策略——每次重试的等待时间翻倍，避免在服务器已经过载时继续施压。

import requests
import time

def resilient_fetch(target_url, max_retries=3):
    """带指数退避的 HTTP 请求"""
    req_headers = {
        "User-Agent": "ResearchBot/1.0 (academic data collection)"
    }

    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.get(target_url, headers=req_headers, timeout=12)
            resp.raise_for_status()
            return resp
        except requests.exceptions.RequestException as err:
            print(f"Attempt {attempt} failed: {err}")
            if attempt < max_retries:
                delay = 2 ** attempt  # 2s, 4s, 8s
                print(f"Waiting {delay}s before retry...")
                time.sleep(delay)
            else:
                print(f"All {max_retries} attempts exhausted for {target_url}")
                return None

本章回顾

本章构建了一条完整的 Web 数据采集管线，涵盖以下关键步骤：

阶段工具核心知识点
发送请求requestsget()/post()、状态码检查、超时与异常处理
解析 HTMLBeautifulSoupselect() / select_one() + CSS 选择器
提取数据.text / .get()文本内容 vs. 属性值
多页采集循环 + URL 拼接观察分页 URL 规律
数据存储csv / jsonDictWriter + json.dump
合规采集robots.txt / 频率控制指数退避、请求间隔、身份标识

掌握了本章的内容后，你已经具备了从公开网站系统化地收集数据的能力。后续可以进一步了解 JavaScript 动态渲染页面的处理（Selenium/Playwright）、反采集策略的应对、以及大规模分布式采集框架（Scrapy）等进阶主题。但对于绝大多数日常数据收集需求，本章的技术栈已经足够。

延伸：数据采集项目的组织方式

当采集任务变得复杂——多个数据源、不同的页面结构、定时执行——你需要一套更有组织的项目结构。以下是几条实用建议：

将采集逻辑与数据处理逻辑分离。 采集器只负责获取原始数据并保存到中间格式（如 JSON Lines），数据处理器从中间格式读取数据进行清洗、转换和入库。这样当某个环节出错时，不需要重新执行整个管线。

为每次采集任务生成唯一标识。 使用时间戳或 UUID 标记每次运行的输出文件。这样即使多次执行也不会覆盖历史数据，便于回溯和对比。

from datetime import datetime

run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
output_file = f"catalog_{run_id}.json"

构建健壮的错误恢复机制。 网络采集不可避免地会遇到失败。记录每一页的采集状态，支持从断点恢复，避免因中间某一页失败而从头重来。

监控和报警。 对于定期运行的采集任务，监控采集到的数据量。如果数据量突然为零或大幅减少，很可能是目标网站结构发生了变化，需要及时修改选择器。

这些工程实践不会让你的代码更短，但会让你的采集系统更可靠、更可维护。下一章，我们将学习如何用 Flask 框架将 Python 代码变成可通过浏览器访问的 Web 应用。

购买课程解锁全部内容

零基础到独立开发：Python 自动化与 Web 实战

¥29.90