Web数据爬取与API获取

数据分析专题 · 从网络获取数据

专题：Python数据分析系统学习

关键词：数据分析, 数据采集, requests, API, BeautifulSoup, Selenium, 网络爬虫, 数据缓存

一、引言：为什么需要网络数据获取

在现代数据分析工作中，数据来源是决定分析质量的首要因素。尽管许多企业拥有内部数据库，但公开的Web数据仍然占据着不可替代的地位——从金融市场的实时行情、电商平台的商品信息，到社交媒体的用户评论、政府公开的统计数据，网络是最大的公共数据湖。Python之所以成为数据科学的首选语言，很大程度上归功于其丰富的网络数据获取生态：requests处理HTTP通信、BeautifulSoup解析HTML文档、Selenium驾驭动态页面、Scrapy构建大规模爬虫。本章将从最基础的HTTP请求开始，逐步深入到完整的API数据流水线，帮助你建立系统化的Web数据获取能力。

学习前提：建议具备基本的Python语法知识（函数、类、异常处理），并已安装Python 3.8+环境。所有代码示例均可直接运行，但请遵守目标网站的robots.txt和服务条款。

网络数据获取的核心流程

无论使用哪种技术，Web数据获取都遵循一个通用流水线：首先向目标服务器发送HTTP请求，服务器返回响应内容（可能是HTML、JSON、XML或二进制文件），然后对响应内容进行解析和提取，最后将提取的数据清洗并存储为结构化格式。理解这个流程有助于我们在面对不同的数据源时选择合适的工具组合。

二、requests库：HTTP请求的核心工具

requests库是Python生态中最流行的HTTP客户端库，以其简洁的API和丰富的功能著称。它封装了底层的urllib3，提供了直观的请求方法、自动的响应解码、方便的回话管理等特性。作为网络数据获取的第一步，掌握requests是后续所有技术的基础。

2.1 基本请求与响应

最基础的GET请求只需一行代码。requests自动处理了URL编码、响应内容的解码（通过分析HTTP头中的charset字段）、连接池复用等底层细节，让开发者专注于业务逻辑。

import requests

# 最基本的GET请求
response = requests.get('https://api.github.com')
print(response.status_code)  # 200 表示成功
print(response.text)          # 原始响应文本
print(response.encoding)      # 自动检测的编码方式

# 带参数的GET请求
params = {'q': 'python', 'sort': 'stars', 'per_page': 10}
response = requests.get('https://api.github.com/search/repositories', params=params)
data = response.json()  # 直接解析JSON响应
print(data['total_count'])

2.2 请求头与身份伪装

许多网站会检查请求头中的User-Agent字段来识别客户端类型。如果使用默认的"python-requests/x.x.x"标识，很容易被服务器拒绝或触发反爬机制。因此，设置合理的请求头是爬虫开发的基本功。除了User-Agent之外，Referer、Accept-Language、Cookie等头部字段也参与了服务器的请求验证逻辑。

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                  'AppleWebKit/537.36 (KHTML, like Gecko) '
                  'Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Referer': 'https://www.google.com/'
}

response = requests.get('https://httpbin.org/headers', headers=headers)
print(response.json())

2.3 POST请求与表单提交

当需要向服务器提交数据时（如登录、搜索、提交表单），使用POST请求。requests支持两种POST数据格式：表单编码（application/x-www-form-urlencoded）和JSON编码（application/json），分别对应data和json参数。

# 表单方式提交（data参数）
form_data = {'username': 'test_user', 'password': 'test_pass'}
response = requests.post('https://httpbin.org/post', data=form_data)

# JSON方式提交（json参数，自动设置Content-Type）
json_data = {'name': 'John', 'age': 30, 'city': 'New York'}
response = requests.post('https://httpbin.org/post', json=json_data)
print(response.json()['json'])

2.4 Cookie管理与会话保持

有些网站需要维持登录状态或会话信息，这时就用到Cookies。requests提供了两种Cookie处理方式：一种是手动在headers中设置Cookie字符串，另一种是使用Session对象自动管理Cookie。Session对象尤其重要，它能跨请求保持Cookie（如登录后的session ID），同时还能复用TCP连接池，提升性能。

# 方式一：手动传递Cookie（适合一次性请求）
cookies = {'session_id': 'abc123', 'user_token': 'xyz789'}
response = requests.get('https://httpbin.org/cookies', cookies=cookies)

# 方式二：使用Session（推荐用于多步操作）
session = requests.Session()
session.headers.update({'User-Agent': 'MyApp/1.0'})

# 登录——Cookie由session自动保存
login_resp = session.post('https://httpbin.org/post',
                         data={'user': 'admin', 'pass': 'secret'})

# 后续请求自动携带登录后的Cookie
profile_resp = session.get('https://httpbin.org/cookies')
print(profile_resp.json())

2.5 超时设置与重试机制

网络请求不可靠，服务器可能响应缓慢或完全不响应。如果不设置超时，程序可能永远挂起。requests支持两种超时：connect timeout（连接建立超时）和read timeout（读取数据超时）。此外，结合urllib3的Retry机制可以实现自动重试，应对临时性的网络故障。

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# 配置重试策略：最多重试3次，对5xx错误和连接错误进行重试
retry_strategy = Retry(
    total=3,
    backoff_factor=1,  # 重试间隔：1s, 2s, 4s（指数退避）
    status_forcelist=[500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('http://', adapter)
session.mount('https://', adapter)

try:
    response = session.get(
        'https://httpbin.org/delay/3',
        timeout=(3.0, 5.0)  # (connect timeout, read timeout)
    )
    print('请求成功', response.status_code)
except requests.exceptions.Timeout:
    print('请求超时')
except requests.exceptions.ConnectionError:
    print('连接失败')

2.6 SSL验证与证书处理

在访问HTTPS网站时，requests默认会验证SSL证书。这在大多数情况下是必要的安全措施，但有时会遇到自签名证书或证书过期的情况。可以设置verify=False跳过验证（但不推荐用于生产环境），或指定自定义的CA证书包。

# 跳过SSL验证（仅用于开发和测试）
response = requests.get('https://self-signed.badssl.com', verify=False)

# 指定自定义CA证书
response = requests.get('https://example.com', verify='/path/to/certfile.pem')

# 使用certifi默认证书（推荐的生产环境做法）
# 安装：pip install certifi
import certifi
response = requests.get('https://example.com', verify=certifi.where())

安全警告：在生产环境中永远不要使用verify=False。这会使得你的HTTPS连接容易遭受中间人攻击。如果遇到SSL错误，应该排查证书问题而非简单跳过验证。

三、HTTP API调用与RESTful实践

现代Web服务大量使用RESTful API作为数据交换接口。与网页爬取相比，API调用更规范、更高效——数据直接以JSON格式返回，无需解析HTML文档。掌握API调用是数据分析师连接外部数据源的关键技能。

3.1 RESTful API基础

RESTful API遵循资源导向的设计哲学，通过HTTP方法表达操作语义：GET获取资源，POST创建资源，PUT/PATCH更新资源，DELETE删除资源。API的响应通常包含状态码（200成功、201创建成功、400参数错误、401未认证、404资源不存在、429请求过频、500服务器错误）和JSON格式的数据体。理解这些约定有助于快速调试和对接不同的API服务。

import requests
import json

api_base = 'https://jsonplaceholder.typicode.com'

# GET：获取资源列表
resp = requests.get(f'{api_base}/posts', params={'userId': 1})
posts = resp.json()
print(f'获取到 {len(posts)} 篇文章')

# POST：创建新资源
new_post = {
    'title': 'Python数据采集入门',
    'body': '本文介绍如何使用Python进行网络数据采集...',
    'userId': 1
}
resp = requests.post(f'{api_base}/posts', json=new_post)
print('创建成功，ID:', resp.json()['id'])

# PUT：完整更新资源
resp = requests.put(f'{api_base}/posts/1',
                    json={'title': '更新后的标题', 'body': '更新后的内容', 'userId': 1})

# DELETE：删除资源
resp = requests.delete(f'{api_base}/posts/1')
print('删除状态:', resp.status_code)  # 应为 200 或 204

3.2 API认证与Token管理

大多数公共API通过API Key或Bearer Token进行身份认证。常见的认证方式包括：在URL参数中传递api_key、在请求头中设置Authorization: Bearer 、使用OAuth 2.0流程获取访问令牌。Token通常有过期时间，生产环境中需要实现自动刷新机制。

# Bearer Token认证（最常见的方式）
headers = {
    'Authorization': 'Bearer your_api_token_here',
    'Accept': 'application/json'
}
resp = requests.get('https://api.github.com/user', headers=headers)

# API Key放在请求头中
headers = {'X-API-Key': 'your_api_key_here'}
resp = requests.get('https://api.example.com/data', headers=headers)

# OAuth 2.0密码模式（获取Token）
token_resp = requests.post('https://api.example.com/oauth/token', json={
    'grant_type': 'password',
    'username': 'user@example.com',
    'password': 'secure_pass',
    'client_id': 'your_client_id',
    'client_secret': 'your_client_secret'
})
access_token = token_resp.json()['access_token']
print('Token:', access_token)

3.3 分页处理与限速控制

当API返回的数据量很大时，通常会采用分页机制。常见的分页方式有：基于页码的page/page_size、基于游标的cursor、基于偏移量的offset/limit。处理分页需要遍历所有页面直到没有更多数据。同时，调用API时必须遵守限速（Rate Limiting）规则，通常以每分钟/每小时允许的请求次数表示。过快的请求会导致429状态码或被封禁IP。

import time

def fetch_all_pages(base_url, headers, per_page=100, max_pages=None):
    """
    通用分页获取函数（基于页码的分页方式）
    自动处理分页遍历，并在请求之间添加延迟以遵守限速
    """
    all_items = []
    page = 1
    rate_limit_per_minute = 60  # 每分钟最多60次请求
    delay = 60.0 / rate_limit_per_minute  # 每次请求之间的间隔（秒）

    while True:
        try:
            resp = requests.get(
                base_url,
                headers=headers,
                params={'page': page, 'per_page': per_page},
                timeout=10
            )
            resp.raise_for_status()
            data = resp.json()

            if not data:
                break  # 没有更多数据

            all_items.extend(data)
            print(f'已获取第 {page} 页，共 {len(data)} 条')

            # 检查是否还有下一页（根据响应头的Link字段）
            if 'next' not in resp.links:
                break

            page += 1
            if max_pages and page > max_pages:
                break

            # 限速：等待后再发下一次请求
            time.sleep(delay)

        except requests.exceptions.HTTPError as e:
            if resp.status_code == 429:
                retry_after = int(resp.headers.get('Retry-After', 60))
                print(f'请求过频，等待 {retry_after} 秒...')
                time.sleep(retry_after)
                continue
            else:
                raise e

    return all_items

3.4 API错误处理最佳实践

健壮的API客户端必须处理各种异常情况：网络中断、服务器错误（5xx）、客户端错误（4xx）、数据解析错误等。好的做法是使用自定义异常类和重试装饰器来构建可复用的API调用层。

from functools import wraps
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class APIError(Exception):
    """API调用异常基类"""
    def __init__(self, message, status_code=None, response=None):
        super().__init__(message)
        self.status_code = status_code
        self.response = response

def api_retry(max_retries=3, backoff=2):
    """API调用重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (requests.exceptions.ConnectionError,
                              requests.exceptions.Timeout) as e:
                    last_exception = e
                    wait_time = backoff ** attempt
                    logger.warning(f'连接失败(第{attempt+1}次)，{wait_time}秒后重试：{e}')
                    time.sleep(wait_time)
                except requests.exceptions.HTTPError as e:
                    if e.response.status_code in [500, 502, 503]:
                        last_exception = e
                        time.sleep(backoff ** attempt)
                    else:
                        raise  # 非服务端错误，不重试
            raise last_exception
        return wrapper
    return decorator

@api_retry(max_retries=3)
def call_api(url, **kwargs):
    resp = requests.get(url, timeout=10, **kwargs)
    resp.raise_for_status()
    return resp.json()

四、BeautifulSoup：HTML解析利器

并非所有网站都提供API接口。当需要从传统HTML页面中提取数据时，BeautifulSoup是最受欢迎的解析库。它能够将复杂的HTML文档转化为可遍历的树形结构，并提供简洁的API进行元素查找、属性提取和文本获取。配合requests使用，几乎可以应对所有的静态页面采集需求。

4.1 解析器的选择与安装

BeautifulSoup本身只是一个封装层，底层需要依赖具体的解析器。最常用的两个解析器是Python内置的html.parser（无需额外安装，速度一般）和lxml（需要安装，速度更快，容错性更强）。对于格式不规范的网页，lxml通常表现更好。

# 安装BeautifulSoup和lxml解析器
# pip install beautifulsoup4 lxml

from bs4 import BeautifulSoup

html_doc = """
<html>
    <head><title>测试页面</title></head>
    <body>
        <div class="content" id="main">
            <h1>文章标题</h1>
            <p class="summary">这是一段摘要</p>
            <div class="tags">
                <span class="tag">Python</span>
                <span class="tag">爬虫</span>
                <span class="tag">数据</span>
            </div>
            <ul class="list">
                <li>项目一</li>
                <li>项目二</li>
            </ul>
        </div>
    </body>
</html>
"""

# 使用lxml解析器（速度更快，推荐）
soup = BeautifulSoup(html_doc, 'lxml')

# 使用html.parser（无需安装额外依赖）
soup = BeautifulSoup(html_doc, 'html.parser')

4.2 find与find_all：精准定位元素

find_all方法返回所有匹配的元素列表，find方法只返回第一个匹配元素。它们支持按标签名、属性、CSS类、文本内容等多种条件进行筛选，还可以组合使用以实现更精确的定位。

soup = BeautifulSoup(html_doc, 'lxml')

# 按标签名查找
title = soup.find('h1')  # 返回第一个h1标签
all_span = soup.find_all('span')  # 返回所有span标签列表

# 按CSS类名查找（使用class_参数，避免与Python关键字冲突）
summary = soup.find(class_='summary')
tags = soup.find_all(class_='tag')

# 按属性查找（使用attrs字典或直接传入）
main_div = soup.find('div', id='main')
content_div = soup.find('div', attrs={'class': 'content'})

# 组合条件查找
list_items = soup.find_all('li', class_='item')

# 按文本内容查找（使用text/string参数）
# soup.find_all('span', text='Python')

# 限制返回数量
first_two = soup.find_all('span', limit=2)

4.3 CSS选择器：更灵活的定位方式

如果你熟悉CSS选择器语法，select方法提供了更简洁的元素定位方式。它支持标签选择器、类选择器、ID选择器、属性选择器、层级选择器（后代/子元素/相邻兄弟）以及伪类选择器。对于复杂的嵌套结构，CSS选择器通常比链式调用find更易读。

soup = BeautifulSoup(html_doc, 'lxml')

# 基本选择器
soup.select('div')              # 所有div标签
soup.select('.content')          # class="content"的元素
soup.select('#main')             # id="main"的元素

# 层级选择器
soup.select('div p')             # div内的所有p元素（后代）
soup.select('div > p')           # div的直接子p元素

# 属性选择器
soup.select('span[class="tag"]')  # 属性精确匹配
soup.select('a[href^="https"]')  # href以https开头
soup.select('img[src$=".jpg"]')  # src以.jpg结尾

# 多条件组合
soup.select('div.content > ul.list li')

# 伪类选择器
soup.select('li:first-child')     # 第一个li
soup.select('li:nth-of-type(2)')  # 第二个li
soup.select('li:last-child')      # 最后一个li

# 实际使用：提取所有标签文本
tags = [tag.get_text() for tag in soup.select('.tag')]
print(tags)  # ['Python', '爬虫', '数据']

4.4 标签属性提取与文本获取

找到元素之后，下一步是提取所需的数据。BeautifulSoup提供了多种数据提取方法：get_text()获取纯文本内容，get('attribute_name')获取属性值，或者直接访问tag的属性字典。对于链接（a标签）和图片（img标签），提取href和src属性是最常见的操作。

# 获取纯文本内容
h1_text = soup.h1.get_text()           # '文章标题'
h1_text = soup.h1.get_text(strip=True)  # 去除首尾空白

# 获取属性值
# 假设有一个a标签：<a href="https://example.com" class="link">链接</a>
link = soup.find('a')
url = link.get('href')    # 或 link['href']
link_text = link.get_text()
css_class = link.get('class')  # 返回列表：['link']

# 获取所有链接
for a in soup.find_all('a'):
    href = a.get('href')
    if href and href.startswith('http'):
        print(a.get_text(strip=True), href)

# 获取所有图片URL
images = []
for img in soup.select('img[src]'):
    images.append({
        'src': img['src'],
        'alt': img.get('alt', ''),
        'width': img.get('width')
    })

4.5 遍历文档树

BeautifulSoup将HTML文档表示为一个树形结构，每个Tag对象代表一个节点。通过父子关系（parent/children/contents）、兄弟关系（next_sibling/previous_sibling）和上下级关系（ descendants）可以灵活遍历整个文档树。这在处理不规则或不具备唯一标识符的页面时非常有用。

# 父子关系
div = soup.find('div', class_='content')
parent = div.parent                # 父节点
children = list(div.children)       # 子节点列表（包含Text节点）
contents = div.contents             # 与children类似，但返回列表而非生成器

# 遍历所有后代
for descendant in div.descendants:
    if descendant.name == 'span':
        print(descendant.get_text())

# 兄弟关系
h1 = soup.find('h1')
next_sib = h1.next_sibling         # 下一个兄弟节点
prev_sib = h1.previous_sibling     # 上一个兄弟节点

# 查找下一个特定标签
next_p = h1.find_next('p')          # 之后第一个p标签
all_next_p = h1.find_all_next('p')  # 之后所有p标签

# 实际场景：提取表格数据
table = soup.find('table')
rows = []
for tr in table.find_all('tr'):
    cells = [td.get_text(strip=True) for td in tr.find_all(['td', 'th'])]
    rows.append(cells)

性能提示：在解析大型HTML文档时，lxml解析器比html.parser快5-10倍。对于需要频繁查找的操作，先用find找到容器元素，再在容器范围内搜索，可以显著减少搜索范围。

五、Selenium：动态页面自动化

现代Web应用大量使用JavaScript动态渲染内容，传统的requests+BeautifulSoup方案无法获取这些动态加载的数据。Selenium通过控制真实的浏览器内核（Chrome/Firefox/Edge），能够执行JavaScript、处理AJAX请求、模拟用户交互，是处理动态页面的终极解决方案。

5.1 WebDriver配置与基本使用

使用Selenium前需要下载对应浏览器的WebDriver。以Chrome为例，需要下载与浏览器版本匹配的chromedriver。推荐使用webdriver-manager库自动管理驱动版本，避免手动安装和版本冲突的麻烦。

# 安装Selenium和WebDriver管理器
# pip install selenium webdriver-manager

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

# 自动下载并管理WebDriver
service = Service(ChromeDriverManager().install())

# 创建浏览器配置
options = webdriver.ChromeOptions()
options.add_argument('--headless')  # 无头模式（不显示浏览器窗口）
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
options.add_argument('--disable-gpu')
options.add_argument('--window-size=1920,1080')
options.add_experimental_option('prefs', {
    'profile.default_content_setting_values.notifications': 2  # 禁用通知
})

driver = webdriver.Chrome(service=service, options=options)
driver.get('https://example.com')
print('页面标题:', driver.title)
print('当前URL:', driver.current_url)

# 使用完毕后关闭
driver.quit()

5.2 等待策略：处理动态加载

动态页面的最大挑战是"不知道数据何时加载完成"。Selenium提供了显式等待和隐式等待两种策略。显式等待（WebDriverWait）针对特定条件进行等待，是最可靠的方式；隐式等待（implicitly_wait）设置一个全局的等待时间，但效率较低。对于精确控制，永远优先使用显式等待。

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

# 显式等待——等待特定元素出现（推荐方式）
try:
    element = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.ID, 'data-table'))
    )
    print('表格已加载')
except:
    print('等待超时，元素未出现')

# 常用的等待条件
EC.presence_of_element_located((By.ID, 'my-id'))          # 元素存在于DOM中
EC.visibility_of_element_located((By.CLASS_NAME, 'data'))   # 元素可见
EC.element_to_be_clickable((By.XPATH, '//button'))         # 元素可点击
EC.text_to_be_present_in_element((By.TAG_NAME, 'h1'), '标题')  # 文本出现

# 等待AJAX请求完成（检查页面中某一元素的文本变化）
WebDriverWait(driver, 10).until(
    EC.text_to_be_present_in_element((By.ID, 'status'), '完成')
)

# 等待某个元素消失（加载指示器消失）
WebDriverWait(driver, 10).until(
    EC.invisibility_of_element_located((By.CLASS_NAME, 'loading'))
)

5.3 页面交互操作

Selenium不仅能读取页面内容，还能模拟用户的浏览器操作：点击按钮、填写表单、滚动页面、下拉选择等。这使得它能够处理需要登录、翻页或触发的数据加载场景。结合显式等待，可以构建出稳健的自动化操作流程。

from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.select import Select
import time

# 1. 文本输入与提交
search_box = driver.find_element(By.NAME, 'q')
search_box.clear()
search_box.send_keys('Python数据采集')
search_box.send_keys(Keys.RETURN)  # 按回车提交

# 2. 按钮点击
button = WebDriverWait(driver, 10).until(
    EC.element_to_be_clickable((By.XPATH, '//button[text()="加载更多"]'))
)
button.click()

# 3. 下拉选择框
select_element = Select(driver.find_element(By.ID, 'page-size'))
select_element.select_by_value('50')     # 按value选择
# select_element.select_by_visible_text('每页50条')  # 按显示文本选择

# 4. 滚动页面（加载延迟内容）
driver.execute_script('window.scrollTo(0, document.body.scrollHeight);')
time.sleep(2)  # 等待懒加载内容

# 5. 获取iframe中的内容
iframe = driver.find_element(By.TAG_NAME, 'iframe')
driver.switch_to.frame(iframe)
iframe_content = driver.find_element(By.TAG_NAME, 'body').text
driver.switch_to.default_content()  # 切回主文档

# 6. 执行自定义JavaScript
page_height = driver.execute_script('return document.body.scrollHeight')
element_text = driver.execute_script(
    'return arguments[0].innerText', element
)

5.4 页面截图与数据提取

Selenium提供了页面截图功能，可以用于调试、数据验证或生成报告。结合页面源码获取（page_source），可以将Selenium采集到的动态内容传递给BeautifulSoup进行解析，充分发挥两个库的各自优势。

from bs4 import BeautifulSoup
from datetime import datetime

# 截图保存（用于调试）
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
driver.save_screenshot(f'screenshot_{timestamp}.png')

# 获取完整页面源码
html = driver.page_source
soup = BeautifulSoup(html, 'lxml')

# 从动态页面中提取结构化数据
products = []
for item in soup.select('.product-item'):
    product = {
        'name': item.select_one('.product-name').get_text(strip=True),
        'price': item.select_one('.price').get_text(strip=True),
        'url': item.select_one('a').get('href'),
        'rating': item.select_one('.rating').get('data-score')
    }
    products.append(product)

print(f'共提取到 {len(products)} 个商品')

# 全页面截图（滚动截屏）
def fullpage_screenshot(driver, file_path):
    """获取整个页面的滚动截图"""
    total_height = driver.execute_script('return document.body.scrollHeight')
    viewport_height = driver.execute_script('return window.innerHeight')
    driver.set_window_size(1920, total_height)
    time.sleep(1)
    driver.save_screenshot(file_path)
    driver.set_window_size(1920, viewport_height)  # 恢复视口

使用注意：Selenium启动的是真实浏览器，内存占用较大（通常200-500MB）。在服务器环境中务必使用无头模式（headless）。如果服务器没有图形界面，还需要安装虚拟显示工具如Xvfb。

六、API数据规范化与清洗

从网络获取的原始数据通常是杂乱的——JSON键名不一致、数据类型不匹配、包含空值和无关字段。在进入分析流程之前，需要对数据进行规范化处理。pandas库提供了强大的JSON解析和数据清洗功能，能够将杂乱的API响应快速转换为整洁的DataFrame。

6.1 JSON响应解析与DataFrame构建

requests的response.json()方法可以将JSON响应直接转换为Python字典或列表。pandas的json_normalize函数（在较新版本中为pd.json_normalize）能够自动展开嵌套的JSON结构，将多层级的字典扁平化为表格格式。这是处理复杂API响应最强大的工具之一。

import pandas as pd
import requests
from pandas import json_normalize

# 获取GitHub API数据
resp = requests.get('https://api.github.com/search/repositories',
                    params={'q': 'language:python', 'sort': 'stars', 'per_page': 5})
data = resp.json()

# 提取items列表并展开嵌套字段
df = json_normalize(data['items'])

# 选择需要的列
columns = [
    'name', 'full_name', 'owner.login',
    'stargazers_count', 'forks_count',
    'open_issues_count', 'language',
    'html_url', 'description', 'created_at'
]
df_repos = df[columns]

# 重命名列（英文→中文）
df_repos.columns = ['仓库名', '全名', '所有者', '星标数',
                        '复刻数', '问题数', '语言', 'URL',
                        '描述', '创建时间']

print(df_repos.head())
print(f'DataFrame形状: {df_repos.shape}')

6.2 复杂嵌套JSON的扁平化处理

真实世界的API响应往往包含深层嵌套结构，例如评论列表中的用户信息、订单中的商品明细、分类体系中的父子层级。json_normalize的record_path和meta参数可以分别指定要展开的列表路径和需要保留的上级字段，实现精准的扁平化转换。

# 模拟复杂的嵌套JSON
complex_data = {
    'status': 'success',
    'total': 2,
    'orders': [
        {
            'order_id': 'ORD-001',
            'customer': {'name': '张三', 'level': 'VIP'},
            'items': [
                {'product': 'Python书', 'qty': 1, 'price': 79.0},
                {'product': '笔记本', 'qty': 2, 'price': 15.0}
            ]
        },
        {
            'order_id': 'ORD-002',
            'customer': {'name': '李四', 'level': '普通'},
            'items': [
                {'product': '显示器', 'qty': 1, 'price': 1999.0}
            ]
        }
    ]
}

# 展开items列表，同时保留order_id和customer信息
df_orders = json_normalize(
    data=complex_data['orders'],
    record_path='items',
    meta=[
        'order_id',
        ['customer', 'name'],
        ['customer', 'level']
    ]
)

print(df_orders)
# 输出：
#   product  qty  price order_id customer.name customer.level
# 0 Python书    1   79.0  ORD-001           张三            VIP
# 1    笔记本    2   15.0  ORD-001           张三            VIP
# 2    显示器    1 1999.0  ORD-002           李四           普通

# 计算每个订单的金额
df_total = df_orders.groupby('order_id').apply(
    lambda x: (x['qty'] * x['price']).sum()
).reset_index(name='total_amount')
print(df_total)

6.3 数据清洗常见操作

从API获取的数据通常需要经过以下清洗步骤：处理缺失值（fillna/dropna）、转换数据类型（astype/to_datetime/to_numeric）、去除重复行（drop_duplicates）、处理异常值（clip/quantile）、标准化文本（strip/lower/regex）。将这些操作封装成一个清洗函数可以提高代码的可复用性。

def clean_api_data(df):
    """API数据通用清洗函数"""
    df = df.copy()

    # 1. 处理缺失值
    if 'description' in df.columns:
        df['description'] = df['description'].fillna('')

    # 2. 转换日期字段
    date_cols = [col for col in df.columns if 'date' in col or 'time' in col or 'at' in col]
    for col in date_cols:
        try:
            df[col] = pd.to_datetime(df[col])
        except:
            pass

    # 3. 字符串清理
    str_cols = df.select_dtypes(include=['object']).columns
    for col in str_cols:
        if df[col].dtype == 'object':
            df[col] = df[col].astype('str').str.strip()

    # 4. 数字类型转换
    for col in ['stargazers_count', 'forks_count', 'open_issues_count']:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0).astype(int)

    # 5. 去除重复行
    df = df.drop_duplicates()

    return df

七、数据缓存策略

在数据采集和API调用的实际工作中，网络请求往往是整个流程中最耗时的环节。频繁请求同一资源不仅浪费时间，还给目标服务器带来不必要的负载。合理的缓存策略可以显著提升效率，同时避免触发反爬机制。requests-cache库为requests提供了透明的缓存支持，是数据采集工具箱中的重要组件。

7.1 requests-cache：透明缓存

requests-cache是requests的一个插件，它在不改变现有代码的前提下，自动缓存HTTP响应。缓存可以是内存、SQLite数据库或Redis服务器。当再次请求相同的URL时（相同的方法、参数和头信息），它会直接返回缓存结果，避免重复网络请求。

# 安装：pip install requests-cache
import requests_cache
import time

# 安装缓存后端（SQLite数据库，缓存有效期3600秒）
requests_cache.install_cache(
    'api_cache',                    # SQLite数据库文件名
    backend='sqlite',                # 后端类型：sqlite/memory/redis
    expire_after=3600,               # 缓存有效期（秒）
    allowable_methods=['GET', 'HEAD'],  # 仅缓存GET和HEAD请求
    ignored_parameters=['_t'],        # 忽略时间戳等动态参数
    match_headers=['Accept'],         # 根据请求头区分缓存
)

# 安装后，requests.get/post等函数自动获得缓存能力
# 无需修改任何现有代码！

# 第一次请求——实际发送HTTP请求
start = time.time()
resp1 = requests.get('https://api.github.com/repos/pandas-dev/pandas')
print(f'第一次请求耗时: {time.time() - start:.2f}秒')
print('来自缓存:', resp1.from_cache)  # False

# 第二次请求——直接从缓存读取
start = time.time()
resp2 = requests.get('https://api.github.com/repos/pandas-dev/pandas')
print(f'第二次请求耗时: {time.time() - start:.2f}秒')
print('来自缓存:', resp2.from_cache)  # True

# 查看缓存统计信息
print('缓存统计:', requests_cache.get_cache().stats())

# 清除缓存
requests_cache.clear()

# 移除缓存（恢复为普通的requests）
requests_cache.uninstall_cache()

7.2 自定义缓存策略

不同的API接口有不同的更新频率。静态数据（如历史价格、固定参照表）可以缓存数小时甚至数天，而实时数据（如股票行情）必须绕过缓存。requests-cache支持为不同的URL模式设置不同的缓存策略，通过CachedSession可以实现细粒度控制。

from requests_cache import CachedSession
from datetime import timedelta

session = CachedSession(
    'data_cache',
    backend='sqlite',
    expire_after=timedelta(hours=1)
)

# 不同的URL使用不同的缓存过期时间

# 静态数据——缓存24小时
session.get('https://api.exchange.com/symbols',
            expire_after=86400)

# 动态数据——不缓存
session.get('https://api.exchange.com/ticker',
            expire_after=-1)  # -1 表示立即过期（不缓存）

# 按URL模式配置缓存策略
session.settings.urls_to_expire_after = {
    '*/historical/*': 86400,  # 历史数据：24小时
    '*/search/*': 300,        # 搜索结果：5分钟
    '*/realtime/*': -1,      # 实时数据：不缓存
}

# 条件请求——使用ETag和Last-Modified
# requests-cache自动支持条件请求，节省带宽
resp = session.get('https://api.github.com/repos/pandas-dev/pandas')
print('ETag:', resp.headers.get('ETag'))
print('过期时间:', resp.expires)

最佳实践：在开发调试阶段禁用缓存，方便测试最新数据。进入生产运行后再启用缓存。可以通过设置环境变量来切换模式。

八、综合实战案例

将前面学到的技术整合起来，完成一个完整的数据采集与分析流程。以下案例从GitHub API获取Python热门仓库的数据，进行清洗和分析，并将结果保存为CSV文件。这个流程代表了Web数据获取的典型工作模式：API调用 → 数据解析 → 清洗转换 → 存储分析。

8.1 完整的数据采集流水线

import requests
import pandas as pd
from pandas import json_normalize
import time
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class GitHubDataCollector:
    """GitHub数据采集器：获取Python热门仓库并分析"""

    API_BASE = 'https://api.github.com'

    def __init__(self, token=None):
        self.session = requests.Session()
        if token:
            self.session.headers.update({'Authorization': f'Bearer {token}'})
        self.session.headers.update({
            'Accept': 'application/vnd.github.v3+json',
            'User-Agent': 'DataCollector/1.0'
        })

    def search_repos(self, query, max_pages=5, per_page=100):
        """搜索仓库，支持分页遍历"""
        all_items = []
        for page in range(1, max_pages + 1):
            try:
                resp = self.session.get(
                    f'{self.API_BASE}/search/repositories',
                    params={'q': query, 'page': page, 'per_page': per_page},
                    timeout=15
                )
                resp.raise_for_status()
                data = resp.json()
                if not data['items']:
                    break
                all_items.extend(data['items'])
                logger.info(f'第{page}页：获取到 {len(data["items"])} 个仓库')
                time.sleep(0.5)  # 限速
            except requests.exceptions.RequestException as e:
                logger.error(f'第{page}页请求失败: {e}')
                break
        return all_items

    def to_dataframe(self, items):
        """将API响应转换为清洗后的DataFrame"""
        df = json_normalize(items)
        columns = {
            'full_name': 'full_name',
            'owner.login': 'owner',
            'stargazers_count': 'stars',
            'forks_count': 'forks',
            'open_issues_count': 'open_issues',
            'language': 'language',
            'description': 'description',
            'html_url': 'url',
            'created_at': 'created_at',
            'updated_at': 'updated_at',
            'size': 'size_kb'
        }
        df = df[[k for k in columns if k in df.columns]]
        df.columns = [columns[c] for c in df.columns]
        df['description'] = df['description'].fillna('')
        df['created_at'] = pd.to_datetime(df['created_at'])
        df['updated_at'] = pd.to_datetime(df['updated_at'])
        return df

# 执行采集
collector = GitHubDataCollector()
raw_data = collector.search_repos('language:python stars:>1000', max_pages=3)
df = collector.to_dataframe(raw_data)
logger.info(f'共获取 {len(df)} 个仓库数据')

# 基本分析
print('=== 语言分布（用于多语言仓库） ===')
print(df['language'].value_counts().head(10))
print('\n=== 星标数统计 ===')
print(df['stars'].describe())

# 保存结果
output_file = f'github_python_repos_{datetime.now().strftime("%Y%m%d")}.csv'
df.to_csv(output_file, index=False, encoding='utf-8-sig')
logger.info(f'数据已保存至: {output_file}')

九、总结与最佳实践

Web数据获取的核心在于选择合适的工具组合。静态页面使用requests+BeautifulSoup，动态页面使用Selenium，结构化数据优先通过API获取。缓存策略、错误处理和限速控制是构建健壮数据采集系统的三大支柱。以下是最佳实践的总结。

工具选择指南

场景	推荐工具	理由
获取JSON/XML格式的API数据	requests + json_normalize	高效、稳定、响应直接可解析
爬取静态HTML页面	requests + BeautifulSoup	轻量、快速、资源占用少
爬取JavaScript渲染页面	Selenium + BeautifulSoup	真实浏览器环境，完整渲染
大规模爬虫项目	Scrapy框架	异步并发、中间件支持、内置去重
需要身份认证的API	requests.Session + OAuth	自动管理Cookie和Token

六大核心原则

尊重规则：每个网站都有robots.txt文件，遵守其中的爬取规则是基本的网络礼仪。同时遵守API的Rate Limit限制，避免对服务器造成压力。
优雅降级：网络请求不可靠，所有代码都必须包含异常处理。设置合理的超时（connect=3s, read=10s），实现指数退避的重试机制。
缓存优先：对于不经常变化的数据，使用requests-cache或本地文件缓存。这不仅提升性能，还降低了被封禁的风险。
数据验证：从网络获取的数据质量参差不齐，必须进行类型检查、空值处理、格式验证。永远不要相信外部数据的完整性。
渐进式采集：从少量数据开始验证流程，确认解析逻辑正确后再扩展采集规模。避免一次性发送大量请求导致IP被封。
版本化存储：原始数据应与处理后的数据分开存储。保留原始JSON响应的副本，方便后续重新处理和回溯分析。

"数据采集不是一次性的任务，而是持续的数据工程流程。良好的采集架构决定了数据分析的质量上限。"

进阶学习方向

掌握了基础工具之后，可以进一步学习以下进阶主题：使用Scrapy构建分布式爬虫框架，利用asyncio+aiohttp实现异步并发采集，通过代理IP池解决反爬限制，结合Docker部署采集服务，以及使用Apache Airflow编排定时采集任务。这些技术将帮助你构建企业级的数据采集基础设施。

法律与道德声明：网络数据采集必须在法律和道德的框架内进行。在采集任何网站数据之前，请务必确认该网站的使用条款（Terms of Service）和服务条款是否允许数据采集。特别注意不要采集涉及个人隐私、版权保护或商业机密的数据。不当的数据采集行为可能导致法律诉讼或IP永久封禁。