Python 爬虫——爬虫基础问题解决

发表于 | 分类于 ，

https://rainmonth.github.io/posts/6be31a91.html

摘要

本文主要介绍爬虫学习所要掌握的基础知识，主要包括Web开发相关基础知识（Html、Css等）、Python相关基础知识、几个简单的Python库的使用等。还需要一些知识储备，比如：

  • re 模块，python 正则基础；

  • requests，用于获取网页内容，供后续处理；

  • lxml，用于处理 Html文档中的 元素；

  • 常用的爬虫策略与反爬策略；

re 模块

re 模块是python内置的模块，代码都在re.py文件中，是Python用来进行正则处理的。如果没有正则基础，在文件的头部有正则表达式最基本的说明，这个更通过网络搜索的结果是一样的，而且每个说明都有例子说明。

常用的正则使用示例

  • 手机号判断

  • 身份证判断

  • 是否包含数字

requests

requests首先需要能拿到网页的内容，这需要用到 requests 库，安装requests库后，会自动安装urllib3)库

1
pip install requests

基础使用方法如下，其他方法看源码无非就是request的调用

1
2
3
import requests

response = requests.request()

以下是 Python requests 库的深度解析，作为基于 urllib3 的上层封装，它提供了更人性化的 HTTP 客户端接口：

核心特性

  1. 极简 API 设计

    • 一行代码完成 GET/POST 等操作，无需手动处理 URL 编码、参数拼接

      1
      2
      import requests
      r = requests.get('https://api.example.com/data', params={'key': 'value'})

  2. 自动内容解析

    • 自动解码响应内容（r.text 根据编码推断字符串，r.json() 直接解析 JSON）

      1
      print(r.json()['result'])  # 直接获取结构化数据

  3. 会话持久化

    • 使用 Session 对象保持 Cookies、头信息、连接池复用

      1
      2
      3
      4
      with requests.Session() as s:
        s.headers.update({'User-Agent': 'MyApp'})
        s.get('https://example.com/login', auth=('user', 'pass'))
        r = s.post('https://example.com/api', data={'action': 'query'})

  4. SSL 验证与证书控制

    • 默认启用证书验证，支持自定义 CA 包或禁用验证（生产环境慎用）

      1
      requests.get('https://example.com', verify='/path/to/cert.pem')  # 自定义证书

基础用法示例

GET 请求

1
2
3
4
5
6
7
8
9
# 带查询参数 & 自定义头
headers = {'Accept': 'application/json'}
params = {'page': 1, 'limit': 20}
r = requests.get('https://api.example.com/items', 
                params=params, 
                headers=headers)

print(f"状态码: {r.status_code}")
print(f"响应头: {r.headers['Content-Type']}")

POST 请求

1
2
3
4
5
6
7
8
9
10
11
# 表单提交
data = {'username': 'admin', 'password': 'secret'}
r = requests.post('https://example.com/login', data=data)

# JSON 提交（自动设置 Content-Type）
payload = {'key': 'value'}
r = requests.post('https://example.com/api', json=payload)

# 文件上传（多部分编码）
files = {'file': open('report.pdf', 'rb')}
r = requests.post('https://upload.example.com', files=files)

高级功能

  1. 超时与重试

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    from requests.adapters import HTTPAdapter
    from requests.packages.urllib3.util.retry import Retry

    session = requests.Session()
    retry = Retry(total=3, backoff_factor=0.3, status_forcelist=[500, 502])
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)

    response = session.get('https://example.com', timeout=(3.05, 27))  # 连接/读取超时

  2. 流式处理大文件

    1
    2
    3
    4
    r = requests.get('https://example.com/large-file', stream=True)
    with open('download.zip', 'wb') as f:
        for chunk in r.iter_content(chunk_size=8192): 
            f.write(chunk)

  3. 代理配置

    1
    2
    3
    4
    5
    proxies = {
        'http': 'http://10.10.1.10:3128',
        'https': 'http://user:pass@10.10.1.10:3128',
    }
    requests.get('http://example.com', proxies=proxies)

响应对象关键属性

属性/方法 说明 示例
status_code HTTP 状态码 200
headers 响应头字典 r.headers['Content-Type']
text 解码后的字符串响应体 html = r.text
content 原始字节数据 img_data = r.content
json() 解析 JSON 响应体 data = r.json()
raise_for_status() 非 2xx 状态码时抛出异常 r.raise_for_status()

异常处理

1
2
3
4
5
6
7
8
9
10
11
try:
    r = requests.get('https://example.com', timeout=5)
    r.raise_for_status()  # 自动检查 4xx/5xx 错误
except requests.exceptions.HTTPError as errh:
    print("HTTP 错误:", errh)
except requests.exceptions.ConnectionError as errc:
    print("连接错误:", errc)
except requests.exceptions.Timeout as errt:
    print("超时:", errt)
except requests.exceptions.RequestException as err:
    print("未知请求异常:", err)

最佳实践

  1. 始终使用会话对象 - 提升性能（连接池复用）
  2. 明确超时设置 - 避免阻塞（推荐 timeout=(3.05, 27))
  3. 优先使用 json 参数 - 替代手动序列化
  4. 及时关闭响应 - 使用 with 语句或手动调用 r.close()

与 urllib3 对比

特性 requests urllib3
API 易用性 ⭐⭐⭐⭐⭐ ⭐⭐
连接池自动化 自动管理 需手动配置 PoolManager
流式下载 支持 (stream=True) 原生支持
适用场景 快速开发、REST API 调用 底层协议控制、高性能场景

💡 选择建议：日常开发首选 requests，需要极致性能或底层控制时使用 urllib3。二者可混合使用（requests 底层依赖 urllib3）。

urllib3

urllib3 是 Python 中一个功能强大且用户友好的 HTTP 客户端库，专注于提供线程安全的连接池、文件上传、重试机制等高级功能。以下是关键特性详解：

核心特性

  1. 线程安全 & 连接池

    • 自动管理连接复用，显著减少 TCP 三次握手开销

    • 支持 HTTPConnectionPool 和 HTTPSConnectionPool 管理不同主机的连接

      1
      2
      import urllib3
      http = urllib3.PoolManager(num_pools=5)  # 控制连接池数量

  2. 重试机制

    1
    2
    3
    from urllib3.util import Retry
    retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502])
    http = urllib3.PoolManager(retries=retries)  # 自动重试失败请求

  3. SSL/TLS 安全

    • 默认启用证书验证，支持自定义 CA 证书

      1
      http = urllib3.PoolManager(ca_certs='/path/to/certs.pem')

典型使用场景

  • GET 请求

    1
    2
    resp = http.request('GET', 'https://api.example.com/data')
    print(resp.status, resp.data.decode('utf-8'))

  • POST 表单 & JSON

    1
    2
    3
    4
    5
    6
    7
    resp = http.request('POST', 'https://api.example.com/submit',
                        fields={'name': 'Alice', 'age': 30})  # 表单提交
    # 或发送 JSON
    import json
    resp = http.request('POST', 'https://api.example.com/submit',
                        body=json.dumps({'data': 'test'}).encode('utf-8'),
                        headers={'Content-Type': 'application/json'})

  • 文件上传

    1
    2
    3
    with open('file.txt', 'rb') as f:
        resp = http.request('POST', 'https://upload.example.com',
                            fields={'file': ('filename.txt', f.read())})

常见对象介绍

以下是 urllib3 库中核心类的详细说明，掌握这些类能更好地控制 HTTP 请求行为：

1. PoolManager

  • 功能：全局入口类，管理所有 HTTP/HTTPS 连接池

  • 关键作用：

    • 自动创建和维护 HTTPConnectionPool/HTTPSConnectionPool
    • 控制最大连接池数量 (num_pools)
    • 处理跨主机的连接复用

  • 典型用法：

    1
    2
    3
    4
    5
    http = urllib3.PoolManager(
        num_pools=10,  # 最大连接池数量
        maxsize=5,     # 每个池最大连接数
        block=True     # 连接不足时是否阻塞等待
    )

2. HTTPConnectionPool

  • 功能：管理特定主机的 HTTP 连接池

  • 核心特性：

    • 同一主机复用 TCP 连接（减少握手开销）
    • 支持连接存活时间 (maxsize + block 参数)

  • 手动使用示例：

    1
    2
    pool = urllib3.HTTPConnectionPool('example.com', maxsize=5)
    resp = pool.request('GET', '/api/data')

3. Retry

  • 功能：定义请求失败时的重试策略

  • 可配置参数：

    1
    2
    3
    4
    5
    6
    7
    Retry(
        total=3,                 # 最大总重试次数
        connect=2,               # 连接错误重试次数
        read=2,                  # 读取超时重试次数
        status_forcelist=[502],  # 强制重试的 HTTP 状态码
        backoff_factor=0.5       # 重试间隔算法因子
    )

  • 高级用法：

    1
    2
    from urllib3.util import Retry
    retry = Retry(raise_on_status=False)  # 不抛出状态码异常

4. ProxyManager

  • 功能：通过代理服务器发送请求

  • 特性：

    • 支持 HTTP/HTTPS/SOCKS 代理
    • 自动处理代理认证

  • 示例：

    1
    2
    3
    4
    5
    proxy = urllib3.ProxyManager(
        'http://user:pass@proxy.example.com:8080/',
        num_pools=5
    )
    resp = proxy.request('GET', 'http://target-site.com')

5. Response

  • 功能：封装 HTTP 响应数据

  • 关键属性：

    1
    2
    3
    4
    resp.status        # 状态码 (200)
    resp.headers       # 响应头 (CaseInsensitiveDict)
    resp.data          # 原始字节数据
    resp.reason        # 状态说明 ('OK')

  • 最佳实践：

    1
    2
    3
    with http.request('GET', url, preload_content=False) as resp:
        for chunk in resp.stream(1024):  # 流式读取大响应
            process(chunk)

6. 异常类

类名 触发场景 处理建议
HTTPError HTTP 状态码异常 (4xx/5xx) 检查 resp.status
MaxRetryError 超过最大重试次数 检查网络或调整 Retry 配置
TimeoutError 连接/读取超时 优化 timeout 参数
SSLError SSL 证书验证失败 检查证书路径或禁用验证

类协作流程

1
2
3
4
5
6
graph TD
    A[PoolManager] -->|创建| B(HTTPConnectionPool)
    B -->|处理请求| C{是否成功?}
    C -->|失败| D[Retry 策略]
    D -->|重试| B
    C -->|成功| E[Response 对象]

掌握这些类的用法，可以更精细地控制连接管理、错误处理等底层细节，适合需要高性能 HTTP 客户端的场景。

高级配置

1
2
3
4
5
6
7
8
9
# 超时控制（连接/读取）
http.request(..., timeout=urllib3.Timeout(connect=2.0, read=10.0))

# 代理设置
http = urllib3.ProxyManager('http://proxy.example.com:8080/')

# 自定义头信息
headers = {'User-Agent': 'MyApp/1.0'}
http.request(..., headers=headers)

异常处理

1
2
3
4
5
6
try:
    resp = http.request(...)
except urllib3.exceptions.HTTPError as e:
    print("HTTP 错误:", e)
except urllib3.exceptions.MaxRetryError:
    print("超过最大重试次数")

性能优化建议

  1. 连接复用 - 始终通过 PoolManager 实例发起请求
  2. 超时设置 - 避免因网络问题阻塞进程
  3. 限制连接池大小 - 防止占用过多系统资源
  4. 及时释放响应体 - 使用 resp.release_conn() 或上下文管理器

与标准库对比

特性 urllib3 标准库 urllib
连接池管理 ✅ 自动复用 ❌ 无
重试机制 ✅ 可定制策略 ❌ 需手动实现
SSL 验证 ✅ 默认开启 ❌ 需复杂配置

💡 适用场景：需精细控制 HTTP 行为时选择 urllib3，快速开发可考虑 requests（基于 urllib3 的上层封装）

lxml

lxml是一个功能丰富并易于使用的用来处理xml或HTML文件的库。作为高效处理 XML/HTML 文档的利器，其底层基于 C 语言库 libxml2/libxslt，兼具高性能与丰富功能：

核心特性

  • 闪电解析速度：比纯 Python 解析库快 5-20 倍
  • XPath 1.0/2.0 支持：精准定位文档节点
  • HTML 容错处理：自动修复残缺标签（类似浏览器行为）
  • 内存友好：支持增量解析大文件

安装与基础用法

1
pip install lxml  # 安装

解析 XML

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from lxml import etree

xml = """
<bookstore>
  <book category="web">
    <title lang="en">XQuery Kick Start</title>
    <author>James McGovern</author>
    <year>2003</year>
    <price>40.00</price>
  </book>
</bookstore>
"""

root = etree.fromstring(xml)  # 解析字符串
print(root.tag)  # 输出根节点标签名: bookstore

# 或解析文件
tree = etree.parse('data.xml')
root = tree.getroot()

核心模块详解

1. ElementTree API

  • 节点操作

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    # 遍历子节点
    for child in root:
        print(child.tag, child.attrib)

    # 获取第一个 book 节点
    first_book = root.find('book')  
    # 获取所有 price 节点
    prices = root.findall('.//price')  

    # 修改节点内容
    first_book.find('price').text = "49.99"

  • 构建 XML

    1
    2
    3
    4
    5
    root = etree.Element("root")
    child = etree.SubElement(root, "child")
    child.set("id", "123")
    child.text = "文本内容"
    print(etree.tostring(root, pretty_print=True).decode())

2. XPath 数据提取

1
2
3
4
5
6
7
8
9
# 选择所有价格超过 30 的书籍标题
titles = root.xpath('//book[price>30]/title/text()')

# 带命名空间的 XPath（常见于 XML 文档）
ns = {'ns': 'http://example.com/ns'}
elements = root.xpath('//ns:element', namespaces=ns)

# 获取属性值
langs = root.xpath('//title/@lang')

3. HTML 解析模块

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from lxml.html import fromstring

broken_html = "<div><p>未闭合标签"
doc = fromstring(broken_html)  # 自动修复结构

# CSS 选择器
links = doc.cssselect('a.external')  
for link in links:
    print(link.get('href'), link.text_content())

# 表单操作（自动化测试常用）
form = doc.forms[0]
form.fields = {'user': 'admin', 'pass': 'secret'}
submit_url = form.action  # 获取表单提交地址

性能优化技巧

  1. 增量解析大文件

    1
    2
    3
    4
    context = etree.iterparse('large.xml', events=('end',), tag='item')
    for event, elem in context:
        process(elem)
        elem.clear()  # 及时清理已处理节点

  2. 预编译 XPath

    1
    2
    find_title = etree.XPath('//title/text()')
    titles = find_title(root)  # 复用编译后的表达式

  3. 禁用无用功能

    1
    2
    parser = etree.XMLParser(remove_blank_text=True, recover=True)  # 忽略空白/错误恢复
    tree = etree.parse('data.xml', parser)

常见问题处理

编码问题

1
2
3
4
# 强制指定输入输出编码
parser = etree.XMLParser(encoding='gbk')
tree = etree.parse('gbk_data.xml', parser=parser)
output = etree.tostring(tree, encoding='utf-8')

处理 CDATA

1
2
3
# 创建 CDATA 节点
cdata = etree.CDATA('<不可转义的内容>')
elem.append(cdata)

错误捕获

1
2
3
4
5
6
from lxml.etree import XMLSyntaxError

try:
    etree.parse('invalid.xml')
except XMLSyntaxError as e:
    print(f"解析错误: {e.message} (行 {e.position[0]})")

与 BeautifulSoup 对比

特性 lxml BeautifulSoup
解析速度 ⭐⭐⭐⭐⭐ (C 扩展) ⭐⭐ (纯 Python)
HTML 容错 自动修复 依赖解析器 (如 html5lib)
XPath 支持 原生支持 需第三方库 (lxml 驱动)
内存占用 低 (SAX 模式) 较高 (DOM 树)
学习曲线 中等 简单

典型应用场景

  1. 爬虫数据提取：XPath + HTML 解析快速抽取结构化数据
  2. XML 配置文件处理：验证/修改符合 Schema 的配置文件
  3. Web 模板引擎：结合 XSLT 转换生成动态页面
  4. 大数据处理：流式解析 GB 级 XML 文件

扩展功能

1
2
3
4
5
6
7
8
9
# XSLT 转换（需 xslt 文件）
transform = etree.XSLT(etree.parse('style.xsl'))
result = transform(tree)
print(str(result))

# Schema 验证
schema = etree.XMLSchema(file='schema.xsd')
if schema.validate(tree):
    print("文档有效")

💡 最佳实践：优先使用 XPath 而非逐层遍历，结合 lxml.html 处理网页抓取，对性能敏感场景启用解析器优化参数。

常用的爬虫策略和反爬策略

以下是爬虫策略与反爬策略的深度解析及攻防实战指南，结合技术实现与合规建议：

一、常用爬虫策略（攻击方）

1. 基础请求优化

  • 请求头伪装

    1
    2
    3
    4
    5
    6
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Referer': 'https://www.example.com/',
        'Accept-Language': 'zh-CN,zh;q=0.9'
    }
    requests.get(url, headers=headers)

  • Cookie 管理
    使用 requests.Session() 自动保持会话，或手动更新 cookies

2. 动态内容处理

  • Headless 浏览器
    使用 Selenium/Puppeteer 渲染 JavaScript：

    1
    2
    3
    4
    5
    6
    from selenium.webdriver import ChromeOptions
    options = ChromeOptions()
    options.add_argument('--headless')  # 无头模式
    driver = webdriver.Chrome(options=options)
    driver.get(url)
    print(driver.page_source)

  • API 逆向工程
    通过浏览器开发者工具分析 XHR/Fetch 请求，直接调用数据接口

3. 分布式架构

  • IP 代理池
    集成付费/免费代理服务，实现 IP 轮换：

    1
    2
    3
    4
    5
    proxies = {
        'http': 'http://10.10.1.10:3128',
        'https': 'http://user:pass@10.10.1.10:3128'
    }
    requests.get(url, proxies=proxies)

  • 消息队列调度
    使用 Redis/RabbitMQ 分配任务，实现多节点协同抓取

4. 反反爬绕过技术

  • 验证码破解
    商用方案：接入打码平台（如超级鹰）
    自研方案：CNN 训练识别简单验证码
  • TLS 指纹伪装
    使用 curl_cffi 等库模拟浏览器 TLS 指纹

5. 数据清洗与存储

  • 去重策略
    Bloom Filter 过滤已抓取 URL
    数据指纹比对（MD5 等）
  • 异构数据存储
    结构化数据存 MySQL/PostgreSQL
    非结构化数据存 MongoDB/MinIO

二、主流反爬策略（防御方）

1. 基础特征检测

  • 流量特征阻断

    • 短时间高频访问（如 >100 次/分钟）
    • 非常规 User-Agent 格式
    • 缺失 Referer/Accept-Language 头

  • 防御代码示例（Node.js）：

    1
    2
    3
    if(req.headers['user-agent'].includes('Python-urllib')) {
        return res.status(403).send('Bot detected!');
    }

2. 高级行为分析

  • 鼠标轨迹检测
    通过 JavaScript 监控用户操作模式
  • 页面停留时间
    正常用户访问间隔 >3 秒，爬虫往往连续请求
  • Honeypot 陷阱
    隐藏不可见链接（CSS: display:none），捕获点击者

3. 动态防护体系

  • 加密参数
    使用前端生成 _token、sign 等加密校验参数

    1
    2
    # 爬虫需逆向生成算法
    token = hashlib.md5(f"{timestamp}secret_key".encode()).hexdigest()

  • API 限速
    令牌桶算法控制接口调用频次：

    1
    2
    3
    4
    from flask_limiter import Limiter
    limiter = Limiter(key_func=get_remote_address)
    @app.route("/api", methods=["GET"])
    @limiter.limit("10/minute")  # 限速

4. 深度学习对抗

  • 用户行为建模
    收集点击流数据，训练 LSTM 模型识别机器行为
  • 动态挑战响应
    随机触发验证码（滑块、点选等）

三、攻防成本对比表

策略 爬虫方成本 防御方成本 突破难度
User-Agent 检测 低 低 ⭐
IP 代理池 中 中 ⭐⭐
验证码识别 高 高 ⭐⭐⭐⭐
TLS 指纹验证 高 高 ⭐⭐⭐⭐
行为模式分析 极高 极高 ⭐⭐⭐⭐⭐

四、合规建议

  1. 遵守 robots.txt 

    1
    2
    3
    4
    5
    6
    from urllib.robotparser import RobotFileParser
    rp = RobotFileParser()
    rp.set_url('https://example.com/robots.txt')
    rp.read()
    if rp.can_fetch('MyBot', url):
        # 执行抓取

  2. 控制请求频率
    添加随机延迟（2-10 秒）

    1
    2
    import random, time
    time.sleep(random.uniform(1, 3))

  3. 数据使用授权
    优先使用官方 API（如 Twitter API、Facebook Graph API）

五、未来趋势

  • AI 驱动攻防：GAN 生成对抗网络制造拟人流量
  • 区块链溯源：爬虫行为上链存证
  • 边缘计算防护：Cloudflare Workers 等边缘节点实时拦截

掌握这些策略需要持续关注最新反爬技术演进，建议定期研究：

  • 浏览器 DevTools 协议更新
  • WebAssembly 反混淆技术
  • Web 标准（如 Privacy Sandbox）的影响