1. 数据采集基础：HTTP协议详解、Requests库实战、API调用与认证机制

数据采集，说白了就是跟服务器打交道。你得知道怎么说话、说什么话、怎么证明你是你。这一章，我们就从最底层的HTTP协议开始，一路讲到用Python的Requests库写代码，最后搞定各种API认证。嗯，这些都是我每天吃饭的家伙。

1.1 HTTP协议：你每天都在用，但你真的懂吗？

HTTP协议，全称超文本传输协议。听着高大上，其实就是客户端和服务器之间约定好的一套「对话规则」。你打开浏览器访问网页，你的浏览器就在用HTTP跟服务器说话。

我个人习惯把HTTP拆成三块来理解：请求行、请求头、请求体。响应也是类似的：状态行、响应头、响应体。就这么简单。

请求方法：你该用GET还是POST？

我在项目中遇到过不少新手，不管取数据还是发数据，一律用GET。结果呢？要么URL太长被截断，要么敏感数据暴露在日志里。记住一个原则：读用GET，写用POST。当然，还有PUT、DELETE、PATCH这些，但日常采集用得最多就是GET和POST。

方法	用途	是否幂等	请求体
GET	获取资源	是	无
POST	创建资源	否	有
PUT	更新资源（全量）	是	有
DELETE	删除资源	是	通常无

状态码：服务器在跟你说什么？

状态码是服务器给你的「回执」。你想想看，你发了个请求过去，服务器总得告诉你结果吧？状态码就是干这个的。

• 2xx（成功）：200 OK 最常见，201 Created 表示创建成功
• 3xx（重定向）：301/302 表示资源挪地方了，Requests库默认会帮你自动跳转
• 4xx（客户端错误）：400 参数不对，401 没认证，403 没权限，404 找不到资源
• 5xx（服务器错误）：500 服务器挂了，502/503 网关或服务不可用

我曾经调试一个接口，一直返回403。查了半天，原来是请求头里少了个 User-Agent。服务器觉得我不是浏览器，直接拒了。嗯，这种坑踩一次就记住了。

1.2 Requests库实战：从Hello World到生产级采集

Python的Requests库，是我用过最顺手的HTTP客户端。没有之一。它的设计哲学就是「让HTTP服务人类」。说白了，就是让你用最少的代码，干最多的事。

安装与第一个请求

pip install requests

装完就能用了。来个最简单的：

import requests

response = requests.get('https://api.github.com')
print(response.status_code)  # 200
print(response.json())       # 自动解析JSON

看到没？三行代码，搞定一个API调用。这就是Requests库的魅力。

带参数的请求

实际采集时，很少有不带参数的请求。比如分页、筛选、排序，都得传参数。

# 方式一：直接拼在URL里
response = requests.get('https://api.example.com/users?page=1&limit=20')

# 方式二：用params参数（推荐）
params = {'page': 1, 'limit': 20}
response = requests.get('https://api.example.com/users', params=params)

我个人习惯用第二种。为什么？代码更干净，参数多了也好维护。你想想看，要是拼20个参数在URL里，那字符串得多长？

请求头：伪装成浏览器

很多服务器会检查 User-Agent，发现你不是浏览器就直接拒绝。解决办法很简单：加个请求头。

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Accept': 'application/json',
    'Accept-Language': 'zh-CN,zh;q=0.9'
}
response = requests.get('https://api.example.com/data', headers=headers)

Session：保持登录状态

有些接口需要登录后才能访问。每次请求都带一遍Cookie？太麻烦了。用Session对象，它会自动帮你维护Cookie。

session = requests.Session()

# 先登录
login_data = {'username': 'admin', 'password': '123456'}
session.post('https://api.example.com/login', data=login_data)

# 后续请求自动携带Cookie
response = session.get('https://api.example.com/profile')
print(response.json())

Session对象还有个好处：可以统一设置headers、超时时间、重试策略。我一般会在项目里封装一个全局Session，所有请求都走它。

异常处理：别让程序莫名其妙挂了

网络请求嘛，什么情况都可能发生。超时、断网、服务器宕机……你得做好兜底。

try:
    response = requests.get('https://api.example.com/data', timeout=5)
    response.raise_for_status()  # 状态码不是2xx就抛异常
except requests.exceptions.Timeout:
    print('请求超时了，重试一下？')
except requests.exceptions.ConnectionError:
    print('网络不通，检查一下网络')
except requests.exceptions.HTTPError as e:
    print(f'HTTP错误：{e.response.status_code}')
except requests.exceptions.RequestException as e:
    print(f'其他错误：{e}')

我曾经在生产环境遇到过一个问题：某个接口偶尔会超时，但没加异常处理，结果整个采集任务卡死了。从那以后，我写请求必加 timeout 和 try-except。

1.3 API调用与认证机制：证明你是你

大多数API不会让你随便调。你得证明你有权限。常见的认证方式就那么几种，我们一个一个说。

Basic Auth：最原始的认证

把用户名和密码用冒号拼起来，再Base64编码，塞进请求头里。简单粗暴，但明文传输，不安全。

from requests.auth import HTTPBasicAuth

response = requests.get(
    'https://api.example.com/secure',
    auth=HTTPBasicAuth('admin', '123456')
)
# 或者简写
response = requests.get(
    'https://api.example.com/secure',
    auth=('admin', '123456')
)

Token/Bearer Token：目前最主流的方式

先登录获取一个token，后续请求都带上这个token。服务器验证token有效，就放行。

# 第一步：获取token
login_resp = requests.post('https://api.example.com/login', json={
    'username': 'admin',
    'password': '123456'
})
token = login_resp.json()['access_token']

# 第二步：用token访问资源
headers = {'Authorization': f'Bearer {token}'}
response = requests.get('https://api.example.com/profile', headers=headers)

Token一般有过期时间。我习惯在代码里判断一下，如果返回401，就重新登录获取新token。这样程序可以一直跑下去。

API Key：简单但容易被滥用

直接把key拼在URL里或者请求头里。简单是简单，但key泄露了就完蛋。

# URL参数方式
response = requests.get('https://api.example.com/data?api_key=your_key_here')

# 请求头方式（更推荐）
headers = {'X-API-Key': 'your_key_here'}
response = requests.get('https://api.example.com/data', headers=headers)

OAuth 2.0：企业级认证标准

OAuth 2.0比较复杂，但很多大厂都在用。核心流程是：获取授权码 → 换取access_token → 用token访问资源。我们做数据采集时，最常见的是客户端凭证模式（Client Credentials）。

# 客户端凭证模式
data = {
    'grant_type': 'client_credentials',
    'client_id': 'your_client_id',
    'client_secret': 'your_client_secret'
}
response = requests.post('https://api.example.com/oauth/token', data=data)
token = response.json()['access_token']

# 后续请求
headers = {'Authorization': f'Bearer {token}'}
response = requests.get('https://api.example.com/protected/data', headers=headers)

1.4 实战：组装一个完整的采集函数

把上面学的东西串起来，写一个能用的采集函数。嗯，这就是我平时干活的样子。

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

def create_session():
    """创建一个带重试机制的Session"""
    session = requests.Session()
    
    # 重试策略
    retry = Retry(
        total=3,
        backoff_factor=0.5,
        status_forcelist=[500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    
    # 默认请求头
    session.headers.update({
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Accept': 'application/json'
    })
    
    return session

def fetch_data(url, params=None, auth_type='bearer'):
    """通用数据采集函数"""
    session = create_session()
    
    # 认证处理
    if auth_type == 'bearer':
        token = os.getenv('API_TOKEN')
        session.headers['Authorization'] = f'Bearer {token}'
    elif auth_type == 'basic':
        username = os.getenv('API_USERNAME')
        password = os.getenv('API_PASSWORD')
        session.auth = (username, password)
    
    try:
        response = session.get(url, params=params, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f'采集失败：{e}')
        return None

# 使用示例
data = fetch_data(
    'https://api.example.com/users',
    params={'page': 1, 'limit': 50},
    auth_type='bearer'
)

这个函数我用了好几年，改来改去，核心逻辑没变过。你拿去改改就能用。