章节 2：数据解析技术栈

学习目标

掌握正则表达式元字符、量词、分组及 re 模块核心函数
熟练使用 BeautifulSoup4 解析 HTML/XML 文档树与 CSS 选择器
理解 lxml 与 XPath 语法，包括轴定位、函数筛选与多条件组合
使用 JSONPath 解析嵌套 JSON 数据
实现多线程并发爬虫与请求限流

2.1 正则表达式

2.1.1 元字符与量词

定义：正则表达式（Regular Expression）是一种用模式字符串匹配文本的工具，Python 通过 re 模块提供支持。

核心元字符：

元字符	含义	示例	匹配结果
`.`	匹配任意字符（除换行）	`a.b`	`a1b`, `a@b`, `axb`
`\d`	匹配数字 `[0-9]`	`\d{3}`	`123`, `456`
`\w`	匹配字母、数字、下划线	`\w+@\w+\.com`	`abc@example.com`
`\s`	匹配空白字符	`name\s+age`	`name age`
`^`	匹配字符串开头	`^Hello`	`Hello world`
`$`	匹配字符串结尾	`world$`	`Hello world`
`[]`	字符集，匹配其中之一	`[aeiou]`	`apple` 中的 `a`
`[^]`	否定字符集	`[^0-9]`	匹配非数字字符
`\b`	单词边界	`\bcat\b`	`cat`, 但非 `catch`

量词：

量词	含义	示例	匹配
`*`	0 次或多次（贪婪）	`ab*c`	`ac`, `abc`, `abbc`
`+`	1 次或多次（贪婪）	`ab+c`	`abc`, `abbc`（不匹配 `ac`）
`?`	0 次或 1 次	`ab?c`	`ac`, `abc`
`{n}`	恰好 n 次	`\d{4}`	`2024`
`{n,}`	至少 n 次	`\d{3,}`	`123`, `12345`
`{n,m}`	n 到 m 次	`\d{2,4}`	`12`, `123`, `1234`

贪婪与非贪婪：

python

import re

text = "<div>内容1</div><div>内容2</div>"

# 贪婪模式（默认）——尽可能多地匹配
greedy = re.findall(r"<div>(.*)</div>", text)
print(greedy)  # ['内容1</div><div>内容2']

# 非贪婪模式——在量词后加 ?
non_greedy = re.findall(r"<div>(.*?)</div>", text)
print(non_greedy)  # ['内容1', '内容2']

2.1.2 分组与捕获

定义：用 () 将正则表达式分组，可以提取子匹配或对分组整体应用量词。

python

import re

text = "联系方式：张三 138-0000-1111，李四 139-1111-2222"

# 命名分组（推荐）
pattern = r"(?P<name>\w+)\s+(?P<phone>\d{3}-\d{4}-\d{4})"
matches = re.finditer(pattern, text)
for m in matches:
    print(f"姓名: {m.group('name')}, 电话: {m.group('phone')}")

# 输出：
# 姓名: 张三, 电话: 138-0000-1111
# 姓名: 李四, 电话: 139-1111-2222

# 非捕获分组 (?:...)——只用于分组不捕获
pattern2 = r"(?:\d{3}-)?\d{4}-\d{4}"
print(re.findall(pattern2, "电话: 010-1234-5678, 手机: 138-0000-1111"))
# ['010-1234-5678', '138-0000-1111']

2.1.3 re 模块核心函数

函数	说明	返回类型
`re.search()`	扫描整个字符串，返回第一个匹配	Match \| None
`re.match()`	从字符串开头匹配	Match \| None
`re.findall()`	返回所有匹配的字符串列表	list[str]
`re.finditer()`	返回所有匹配的迭代器	Iterator[Match]
`re.sub()`	替换匹配的文本	str
`re.split()`	按匹配分割字符串	list[str]
`re.compile()`	编译正则表达式为 Pattern 对象	Pattern

python

import re

text = "我的邮箱是 zhangsan@example.com，备用邮箱是 li@test.org"

# search——第一个匹配
m = re.search(r"[\w.]+@[\w.]+\.\w+", text)
print(m.group())  # zhangsan@example.com

# findall——所有匹配
emails = re.findall(r"[\w.]+@[\w.]+\.\w+", text)
print(emails)  # ['zhangsan@example.com', 'li@test.org']

# sub——替换
result = re.sub(r"[\w.]+@[\w.]+\.\w+", "[隐藏邮箱]", text)
print(result)  # 我的邮箱是 [隐藏邮箱]，备用邮箱是 [隐藏邮箱]

# split——分割
parts = re.split(r"[，,；;]\s*", "北京, 上海；广州，深圳")
print(parts)  # ['北京', '上海', '广州', '深圳']

# compile——预编译提高性能
pattern = re.compile(r"\d{3}-\d{4}-\d{4}")
phone = pattern.search("联系：138-0000-1111")
if phone:
    print(phone.group())  # 138-0000-1111

2.1.4 实战：爬虫数据提取

python

import re
import requests

# 从 HTML 中提取所有链接
html = """
<div class="links">
    <a href="https://example.com/page1">页面1</a>
    <a href="/page2">页面2</a>
    <a href="https://example.com/page3">页面3</a>
</div>
"""

# 提取所有 href 属性值
links = re.findall(r'href="([^"]+)"', html)
print(links)
# ['https://example.com/page1', '/page2', 'https://example.com/page3']

# 从 JavaScript 变量中提取 JSON 数据
script = 'var productInfo = {"id": 1001, "name": "商品A", "price": 99.9};'
match = re.search(r'var\s+productInfo\s*=\s*({.*?});', script)
if match:
    import json
    data = json.loads(match.group(1))
    print(data)  # {'id': 1001, 'name': '商品A', 'price': 99.9}

2.2 BeautifulSoup4

2.2.1 安装与初始化

bash

pip install beautifulsoup4 lxml

定义：BeautifulSoup4（bs4）是一个从 HTML/XML 文件中提取数据的 Python 库，能够将输入文档转换为 Unicode 编码，并生成解析树。

python

from bs4 import BeautifulSoup

html_doc = """
<html>
<head><title>商品列表</title></head>
<body>
    <div class="product" id="p1">
        <h2 class="name">iPhone 15</h2>
        <span class="price">¥6999</span>
        <p class="desc">最新款智能手机</p>
    </div>
    <div class="product" id="p2">
        <h2 class="name">MacBook Pro</h2>
        <span class="price">¥14999</span>
        <p class="desc">高性能笔记本</p>
    </div>
</body>
</html>
"""

# 使用 lxml 解析器
soup = BeautifulSoup(html_doc, "lxml")

# 打印美化后的 HTML
print(soup.prettify())

2.2.2 文档树导航

python

# 标签名定位（返回第一个匹配）
print(soup.title)          # <title>商品列表</title>
print(soup.title.string)   # 商品列表
print(soup.title.text)     # 商品列表

# 获取标签名
print(soup.h2.name)        # h2

# 获取属性
print(soup.div["id"])      # p1
print(soup.div.get("class"))  # ['product']
print(soup.div.attrs)      # {'class': ['product'], 'id': 'p1'}

# 父子节点
print(soup.div.parent.name)     # body
print([c.name for c in soup.body.children])  # 直接子节点
print([c.name for c in soup.body.descendants])  # 所有子孙节点

# 兄弟节点
print(soup.div.next_sibling)       # 下一个同辈节点
print(soup.div.previous_sibling)   # 上一个同辈节点

2.2.3 搜索方法

python

# find()——返回第一个匹配
product = soup.find("div", class_="product")
print(product.h2.text)  # iPhone 15

# find_all()——返回所有匹配
products = soup.find_all("div", class_="product")
for p in products:
    name = p.find("h2", class_="name").text
    price = p.find("span", class_="price").text
    print(f"{name}: {price}")
# iPhone 15: ¥6999
# MacBook Pro: ¥14999

# 多种过滤方式
soup.find_all("div")                          # 标签名
soup.find_all(class_="product")               # CSS 类
soup.find_all(id="p1")                        # ID
soup.find_all("div", class_="product")        # 组合
soup.find_all("h2", string="iPhone 15")       # 文本内容
soup.find_all("div", limit=1)                 # 限制数量

# 正则表达式过滤
import re
price_spans = soup.find_all("span", string=re.compile(r"¥"))
print([s.text for s in price_spans])  # ['¥6999', '¥14999']

# 自定义函数过滤
def has_price_class(tag):
    return tag.name == "span" and "price" in tag.get("class", [])
print([t.text for t in soup.find_all(has_price_class)])

2.2.4 CSS 选择器

python

# select_one()——返回第一个匹配
first_product = soup.select_one(".product")
print(first_product.h2.text)  # iPhone 15

# select()——返回所有匹配
products = soup.select(".product")
print(len(products))  # 2

# 常用 CSS 选择器示例
soup.select("div")                    # 所有 div
soup.select(".product")               # class="product"
soup.select("#p1")                    # id="p1"
soup.select("div.product")            # div 且 class="product"
soup.select("div > h2")               # div 的直接子元素 h2
soup.select("div + div")              # div 后面的同级 div
soup.select("[class~=name]")          # class 包含 "name"
soup.select("h2.name")                # h2 且 class="name"
soup.select("div:first-child")        # 第一个 div 子元素
soup.select("div:nth-of-type(2)")     # 第二个 div

2.2.5 修改文档树

python

# 修改标签属性
soup.div["data-id"] = "1001"
del soup.div["class"]

# 修改文本内容
soup.title.string = "新标题"

# 删除标签
soup.find("p", class_="desc").decompose()

# 插入新标签
from bs4 import Tag, NavigableString
new_tag = soup.new_tag("a", href="https://apple.com")
new_tag.string = "查看详情"
soup.body.append(new_tag)

2.3 lxml 与 XPath

2.3.1 lxml 基础

定义：lxml 是一个高性能的 XML/HTML 解析库，底层使用 C 语言 libxml2/libxslt，支持 XPath 1.0 和 XSLT。

python

from lxml import etree

html = """
<html>
<body>
    <ul>
        <li class="item"><a href="/a">链接A</a></li>
        <li class="item active"><a href="/b">链接B</a></li>
        <li class="item"><a href="/c">链接C</a></li>
    </ul>
</body>
</html>
"""

# 解析 HTML
tree = etree.HTML(html)

# 补全后的 HTML
result = etree.tostring(tree, pretty_print=True, encoding="unicode")
print(result)

2.3.2 XPath 基础语法

定义：XPath（XML Path Language）是一种在 XML/HTML 文档中定位节点的查询语言。

路径表达式：

表达式	含义
`/`	从根节点选取
`//`	从任意位置选取（跨层级）
`.`	当前节点
`..`	父节点
`@`	选取属性
`*`	通配符，匹配任何元素
`text()`	选取文本节点

python

# 基本路径
tree.xpath("/html/body/ul/li")           # 绝对路径——所有 li
tree.xpath("//li")                        # 所有 li 元素
tree.xpath("//li/a")                      # 所有 li 下的 a
tree.xpath("//a/@href")                   # 所有 a 的 href 属性值
tree.xpath("//a/text()")                  # 所有 a 的文本

# 通配符
tree.xpath("//ul/*")                      # ul 的所有直接子元素
tree.xpath("//*[@class]")                 # 所有带 class 属性的元素

# 父子/祖先
tree.xpath("//a/..")                      # 所有 a 的父元素
tree.xpath("//a/ancestor::ul")            # 所有 a 的祖先 ul

2.3.3 XPath 谓词（条件筛选）

python

# 索引（1-based！）
tree.xpath("//li[1]")                     # 第一个 li
tree.xpath("//li[last()]")                # 最后一个 li
tree.xpath("//li[position()<3]")          # 前两个 li

# 属性过滤
tree.xpath("//li[@class='item']")         # class="item" 的 li
tree.xpath("//li[@class]")                # 有 class 属性的 li
tree.xpath("//li[not(@class)]")           # 没有 class 属性的 li

# 多条件
tree.xpath("//li[@class='item' and @id]") # 同时满足
tree.xpath("//li[@class='item' or @class='active']")  # 任一满足

# 文本匹配
tree.xpath("//a[contains(text(), '链接')]")   # 文本包含
tree.xpath("//a[starts-with(@href, '/a')]")   # 属性开头匹配
tree.xpath("//a[string-length(@href)>2]")     # 属性长度大于 2

2.3.4 XPath 轴（Axes）

定义：轴定义了相对于当前节点的节点集合方向。

轴	含义	示例
`child`	子节点（默认）	`//li/child::a`
`parent`	父节点	`//a/parent::li`
`ancestor`	所有祖先	`//a/ancestor::div`
`descendant`	所有子孙	`//ul/descendant::a`
`following`	之后的所有节点	`//li[1]/following::li`
`preceding`	之前的所有节点	`//li[2]/preceding::li`
`following-sibling`	之后的所有兄弟节点	`//li[1]/following-sibling::li`
`self`	当前节点	`//li/self::*`

python

# 轴示例
tree = etree.HTML("""
<div id="main">
    <h2>标题</h2>
    <ul>
        <li id="a">A</li>
        <li id="b">B</li>
        <li id="c">C</li>
    </ul>
</div>
""")

# 获取 id="b" 的 li 之后的所有兄弟 li
result = tree.xpath("//li[@id='b']/following-sibling::li")
print([r.text for r in result])  # ['C']

# 获取 li 的祖先 div
result = tree.xpath("//li[@id='a']/ancestor::div/@id")
print(result)  # ['main']

2.3.5 XPath 函数

python

# 字符串函数
tree.xpath("//a[contains(@href, 'example')]")     # 包含
tree.xpath("//a[starts-with(@href, '/')]")         # 开头
tree.xpath("//a[ends-with(@href, '.pdf')]")        # 结尾（XPath 2.0+）
tree.xpath("//a[normalize-space(text())='链接A']")  # 去除首尾空白

# 数字函数
tree.xpath("//li[count(a)>1]")                    # 有多个 a 的 li
tree.xpath("//li[position() mod 2 = 0]")          # 偶数位置的 li

# 其他
tree.xpath("//li[string-length(text())>0]")       # 非空文本
tree.xpath("//p[not(*) and not(text())]")         # 空 p 标签
tree.xpath("//*[@class and contains(@class,'active')]")  # class 包含 active

2.3.6 lxml 与 XPath 完整实战

python

import requests
from lxml import etree

# 爬取豆瓣电影 Top250 标题
url = "https://movie.douban.com/top250"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
}
resp = requests.get(url, headers=headers)
tree = etree.HTML(resp.text)

# XPath 提取电影标题
titles = tree.xpath('//span[@class="title"][1]/text()')
print(titles[:5])

# 提取评分
ratings = tree.xpath('//span[@class="rating_num"]/text()')
print(ratings[:5])

# 提取评价人数
votes = tree.xpath('//div[@class="star"]/span[last()]/text()')
print(votes[:5])

2.4 JSONPath 解析

定义：JSONPath 是一种从 JSON 文档中提取数据的查询语言，类似于 XPath 之于 XML。

2.4.1 安装与基础语法

bash

pip install jsonpath-ng

JSONPath 语法对照表：

JSONPath	含义	等价 XPath
`$`	根对象	`/`
`.key`	子属性	`/key`
`['key']`	子属性（括号形式）	`/key`
`[n]`	数组第 n 个元素	`[n+1]`
`[*]`	数组所有元素	`/*`
`..`	递归下降（深查找）	`//`
`@`	当前节点	`.`
`?()`	过滤表达式	`[predicate]`
`*`	通配符	`*`

2.4.2 实战示例

python

import json
from jsonpath_ng import parse

data = {
    "store": {
        "book": [
            {"title": "爬虫实战", "price": 59.9, "category": "technology"},
            {"title": "三国演义", "price": 39.9, "category": "literature"},
            {"title": "数据科学", "price": 79.9, "category": "technology"},
        ],
        "bicycle": {"color": "red", "price": 199.9},
    }
}

# 提取所有书名
expr = parse("$.store.book[*].title")
titles = [match.value for match in expr.find(data)]
print(titles)  # ['爬虫实战', '三国演义', '数据科学']

# 递归查找所有 price
expr = parse("$..price")
prices = [match.value for match in expr.find(data)]
print(prices)  # [59.9, 39.9, 79.9, 199.9]

# 过滤：提取价格大于 50 的书
expr = parse("$.store.book[?(@.price > 50)]")
books = [match.value for match in expr.find(data)]
for b in books:
    print(b["title"], b["price"])
# 爬虫实战 59.9
# 数据科学 79.9

2.5 多线程并发爬虫

2.5.1 ThreadPoolExecutor 线程池

定义：concurrent.futures.ThreadPoolExecutor 提供了一个高层次的线程池接口，可以自动管理线程创建、任务分配和结果收集。

python

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/delay/3",
]


# 方式一：submit()——返回 Future 对象
def fetch(url):
    resp = requests.get(url, timeout=10)
    return url, resp.status_code

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}

    for future in as_completed(futures):
        url, status = future.result()
        print(f"{url} -> {status}")


# 方式二：map()——批量提交
def fetch_simple(url):
    resp = requests.get(url, timeout=10)
    return resp.status_code

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch_simple, urls))
    print(results)  # [200, 200, 200]

2.5.2 Queue 限流

定义：queue.Queue 结合线程池可以实现请求频率控制和任务队列管理，防止对目标服务器造成过大压力。

python

import threading
import time
import requests
from queue import Queue
from concurrent.futures import ThreadPoolExecutor


class RateLimiter:
    """简单的请求限流器"""

    def __init__(self, max_qps=2):
        self.max_qps = max_qps
        self.last_request_time = 0
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            elapsed = time.time() - self.last_request_time
            wait_time = 1.0 / self.max_qps - elapsed
            if wait_time > 0:
                time.sleep(wait_time)
            self.last_request_time = time.time()


# 带限流的爬虫
url_queue = Queue()
for i in range(20):
    url_queue.put(f"https://httpbin.org/delay/0.1?page={i}")

limiter = RateLimiter(max_qps=5)  # 每秒最多 5 次请求
results = []


def worker():
    while not url_queue.empty():
        try:
            url = url_queue.get_nowait()
        except:
            break
        limiter.acquire()
        try:
            resp = requests.get(url, timeout=10)
            results.append((url, resp.status_code))
        except Exception as e:
            print(f"错误: {url} -> {e}")
        finally:
            url_queue.task_done()


with ThreadPoolExecutor(max_workers=10) as executor:
    for _ in range(10):
        executor.submit(worker)

url_queue.join()
print(f"完成 {len(results)} 个请求")

5.5.3 生产者-消费者模式

python

import threading
import time
import requests
from queue import Queue


class ProducerConsumerCrawler:
    def __init__(self, max_workers=5, queue_size=100):
        self.url_queue = Queue(maxsize=queue_size)
        self.data_queue = Queue()
        self.max_workers = max_workers
        self.stop_event = threading.Event()

    def producer(self, urls):
        """生产者：生成 URL"""
        for url in urls:
            if self.stop_event.is_set():
                break
            self.url_queue.put(url)
            print(f"[生产者] 添加: {url}")
        # 发送结束信号
        for _ in range(self.max_workers):
            self.url_queue.put(None)

    def consumer(self):
        """消费者：下载页面"""
        while True:
            url = self.url_queue.get()
            if url is None:  # 结束信号
                break
            try:
                resp = requests.get(url, timeout=10)
                self.data_queue.put((url, resp.text[:100]))
                print(f"[消费者] 下载完成: {url}")
            except Exception as e:
                print(f"[消费者] 失败: {url} -> {e}")

    def run(self, urls):
        threads = []
        # 启动生产者线程
        p = threading.Thread(target=self.producer, args=(urls,))
        threads.append(p)

        # 启动消费者线程池
        for _ in range(self.max_workers):
            c = threading.Thread(target=self.consumer)
            threads.append(c)

        for t in threads:
            t.start()
        for t in threads:
            t.join()

        print(f"共采集 {self.data_queue.qsize()} 个页面")


# 运行
crawler = ProducerConsumerCrawler(max_workers=3)
crawler.run([f"https://httpbin.org/get?page={i}" for i in range(10)])

小结

正则表达式：元字符 + 量词 + 分组是核心三板斧，re.findall() 和re.sub() 是爬虫中最常用的两个函数
BeautifulSoup4：find() / find_all() 适合精确搜索，select() 适合 CSS 选择器风格的批量提取
lxml + XPath：性能最优，// 路径 + [@属性] 谓词 + contains/text() 是日常三件套
JSONPath：$..key 递归查找 + [?(@.price>50)] 过滤表达式最适合 API 响应解析
多线程爬虫：ThreadPoolExecutor.submit() 搭配 as_completed() 是最佳实践，Queue 限流器保护目标服务器

练习

正则提取：给定 HTML 文本，使用正则表达式提取所有图片 URL（<img src="..."）和 alt 文本。
BeautifulSoup 实战：爬取任意新闻网站首页，使用 select() 提取所有新闻标题和链接，保存到 CSV 文件。
XPath 实战：用 lxml 解析豆瓣电影 Top250，提取每部电影的排名、标题、评分和评价人数，打印表格。
JSONPath 练习：给定的嵌套 JSON（模拟 API 响应），用 JSONPath 提取所有 userId=1 的 title 字段。
并发爬虫：编写一个多线程爬虫，爬取 20 个页面（使用 https://httpbin.org/delay/0.1?n=X），要求 QPS 限制为 3，记录总耗时。

章节 2：数据解析技术栈 ​

学习目标 ​

2.1 正则表达式 ​

2.1.1 元字符与量词 ​

2.1.2 分组与捕获 ​

2.1.3 re 模块核心函数 ​

2.1.4 实战：爬虫数据提取 ​

2.2 BeautifulSoup4 ​

2.2.1 安装与初始化 ​

2.2.2 文档树导航 ​

2.2.3 搜索方法 ​

2.2.4 CSS 选择器 ​

2.2.5 修改文档树 ​

2.3 lxml 与 XPath ​

2.3.1 lxml 基础 ​

2.3.2 XPath 基础语法 ​

2.3.3 XPath 谓词（条件筛选） ​

2.3.4 XPath 轴（Axes） ​

2.3.5 XPath 函数 ​

2.3.6 lxml 与 XPath 完整实战 ​

2.4 JSONPath 解析 ​

2.4.1 安装与基础语法 ​

2.4.2 实战示例 ​

2.5 多线程并发爬虫 ​

2.5.1 ThreadPoolExecutor 线程池 ​

2.5.2 Queue 限流 ​

5.5.3 生产者-消费者模式 ​

小结 ​

练习 ​

章节 2：数据解析技术栈

学习目标

2.1 正则表达式

2.1.1 元字符与量词

2.1.2 分组与捕获

2.1.3 re 模块核心函数

2.1.4 实战：爬虫数据提取

2.2 BeautifulSoup4

2.2.1 安装与初始化

2.2.2 文档树导航

2.2.3 搜索方法

2.2.4 CSS 选择器

2.2.5 修改文档树

2.3 lxml 与 XPath

2.3.1 lxml 基础

2.3.2 XPath 基础语法

2.3.3 XPath 谓词（条件筛选）

2.3.4 XPath 轴（Axes）

2.3.5 XPath 函数

2.3.6 lxml 与 XPath 完整实战

2.4 JSONPath 解析

2.4.1 安装与基础语法

2.4.2 实战示例

2.5 多线程并发爬虫

2.5.1 ThreadPoolExecutor 线程池

2.5.2 Queue 限流

5.5.3 生产者-消费者模式

小结

练习