Park Lam's 每日分享

Python爬虫：httplib2库详解

作者：

Park Lam

在

在当今数字化时代，数据采集已成为众多企业和个人的刚性需求。Python凭借其简洁的语法和丰富的生态系统，成为了网络爬虫开发的首选语言。在众多Python网络请求库中，httplib2以其独特的缓存机制和对HTTP协议的全面支持，为我们提供了一个可靠的数据采集解决方案。今天，让我们深入了解这个强大的HTTP客户端库。

httplib2简介与特性

httplib2是一个全功能的HTTP客户端库，专注于HTTP/1.1协议的完整实现。它不仅提供了基础的HTTP请求功能，还包含了许多高级特性，使其在网络爬虫领域占据重要地位。

核心优势：

  1. 智能缓存机制：httplib2实现了完整的HTTP缓存语义，能够自动处理ETag和Last-Modified等缓存控制头，显著减少重复请求，提高爬虫效率。
  2. 自动处理重定向：支持多级重定向跟踪，并保持请求方法的语义一致性。
  3. 压缩支持：自动处理gzip和deflate压缩，优化带宽使用。
  4. 认证支持：内置多种认证方式，包括Basic认证和Digest认证。

使用限制：

  1. 不支持异步操作
  2. 相比requests库，社区活跃度较低
  3. 文档相对简略

安装与基础使用

安装方法

1
pip install httplib2

基础使用示例

让我们从一个简单的GET请求开始：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import httplib2
import json
 
# 创建HTTP对象
h = httplib2.Http('.cache')  # .cache参数指定缓存目录
 
# 发起GET请求
response, content = h.request('https://api.github.com/users/github', 'GET')
 
# 查看响应头
print("响应状态:", response.status)
print("响应头:", json.dumps(dict(response), indent=2))
 
# 解码响应内容
content_str = content.decode('utf-8')
print("响应内容:", json.dumps(json.loads(content_str), indent=2))

这段代码展示了httplib2的基本用法。我们创建了一个Http对象，并指定了缓存目录。request方法返回一个元组，包含响应头和响应体。

进阶特性演示

1. 使用缓存机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import httplib2
import time
 
def make_request(url):
    h = httplib2.Http('.cache')
    start_time = time.time()
    response, content = h.request(url)
    end_time = time.time()
    return response, content, end_time - start_time
 
# 第一次请求
print("首次请求：")
response, content, duration = make_request('https://api.github.com/zen')
print(f"耗时: {duration:.2f}秒")
print(f"是否使用缓存: {response.get('status') == '304'}\n")
 
# 第二次请求
print("第二次请求（使用缓存）：")
response, content, duration = make_request('https://api.github.com/zen')
print(f"耗时: {duration:.2f}秒")
print(f"是否使用缓存: {response.get('status') == '304'}")

2. 处理认证请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import httplib2
import base64
 
def make_authenticated_request(url, username, password):
    h = httplib2.Http('.cache')
     
    # 构造Basic认证头
    auth_string = base64.b64encode(f"{username}:{password}".encode()).decode()
    headers = {'Authorization': f'Basic {auth_string}'}
     
    try:
        response, content = h.request(url, headers=headers)
        return response, content
    except httplib2.HttpLib2Error as e:
        print(f"请求失败: {e}")
        return None, None
 
# 示例使用
url = 'https://api.example.com/protected'
response, content = make_authenticated_request(url, 'username', 'password')
 
if response:
    print(f"状态码: {response.status}")
    print(f"响应内容: {content.decode('utf-8')}")

3. 自定义请求头和POST请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import httplib2
import json
 
def post_data(url, data):
    h = httplib2.Http('.cache')
     
    headers = {
        'Content-Type': 'application/json',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
     
    # 将数据转换为JSON字符串
    body = json.dumps(data)
     
    try:
        response, content = h.request(
            url,
            'POST',
            body=body,
            headers=headers
        )
        return response, content
    except httplib2.HttpLib2Error as e:
        print(f"请求失败: {e}")
        return None, None
 
# 示例数据
data = {
    'name': 'test_user',
    'email': 'test@example.com'
}
 
# 发送POST请求
response, content = post_data('https://api.example.com/users', data)
 
if response:
    print(f"状态码: {response.status}")
    print(f"响应内容: {content.decode('utf-8')}")

实战案例：构建网站监控工具

下面我们将使用httplib2构建一个简单的网站监控工具，它能够定期检查网站状态并记录响应时间：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import httplib2
import time
import json
from datetime import datetime
import os
 
class WebsiteMonitor:
    def __init__(self, cache_dir='.cache'):
        self.http = httplib2.Http(cache_dir)
        self.monitoring_data = {}
     
    def check_website(self, url):
        try:
            start_time = time.time()
            response, content = self.http.request(url)
            duration = time.time() - start_time
             
            return {
                'status': response.status,
                'response_time': round(duration, 3),
                'timestamp': datetime.now().isoformat(),
                'cache_hit': response.get('status') == '304'
            }
        except Exception as e:
            return {
                'status': 'error',
                'error_message': str(e),
                'timestamp': datetime.now().isoformat()
            }
     
    def monitor(self, urls, interval=300):
        """
        监控指定的URL列表，间隔时间为interval秒
        """
        print(f"开始监控 {len(urls)} 个网站...")
         
        try:
            while True:
                for url in urls:
                    result = self.check_website(url)
                     
                    if url not in self.monitoring_data:
                        self.monitoring_data[url] = []
                     
                    self.monitoring_data[url].append(result)
                    self._save_report(url, result)
                     
                    print(f"\n检查 {url}:")
                    print(f"状态: {result['status']}")
                    print(f"响应时间: {result.get('response_time', 'N/A')}秒")
                    print(f"时间: {result['timestamp']}")
                 
                time.sleep(interval)
                 
        except KeyboardInterrupt:
            print("\n监控已停止")
            self._generate_final_report()
     
    def _save_report(self, url, result):
        """保存单次检查结果"""
        filename = f"monitoring_{datetime.now().strftime('%Y%m%d')}.json"
         
        try:
            if os.path.exists(filename):
                with open(filename, 'r', encoding='utf-8') as f:
                    data = json.load(f)
            else:
                data = {}
             
            if url not in data:
                data[url] = []
             
            data[url].append(result)
             
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(data, f, indent=2)
                 
        except Exception as e:
            print(f"保存报告失败: {e}")
     
    def _generate_final_report(self):
        """生成最终监控报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'summary': {}
        }
         
        for url, data in self.monitoring_data.items():
            success_requests = [d for d in data if d['status'] == 200]
            cache_hits = [d for d in data if d.get('cache_hit', False)]
             
            report['summary'][url] = {
                'total_checks': len(data),
                'successful_checks': len(success_requests),
                'cache_hits': len(cache_hits),
                'average_response_time': sum(d.get('response_time', 0) for d in success_requests) / len(success_requests) if success_requests else 0
            }
         
        with open('monitoring_summary.json', 'w', encoding='utf-8') as f:
            json.dump(report, f, indent=2)
         
        print("\n监控报告已生成: monitoring_summary.json")
 
# 使用示例
if __name__ == '__main__':
    monitor = WebsiteMonitor()
    urls = [
        'https://www.python.org',
        'https://github.com',
        'https://pypi.org'
    ]
     
    monitor.monitor(urls, interval=60)  # 每60秒检查一次

这个实战案例展示了如何使用httplib2构建一个实用的网站监控工具。它具有以下特性：

  1. 支持多站点同时监控
  2. 自动记录响应时间和状态码
  3. 利用httplib2的缓存机制减少服务器负载
  4. 保存详细的监控记录
  5. 生成总结报告

使用注意事项

在使用httplib2进行网络爬虫时，请务必注意以下几点：

  1. 遵守网站的robots.txt规范
  2. 合理设置请求间隔，避免对目标服务器造成压力
  3. 注意处理各种异常情况
  4. 合理利用缓存机制，避免无谓的请求
  5. 遵守网站的使用条款和服务协议

相关资源

结语

httplib2虽然不如requests库知名，但其独特的缓存机制和对HTTP协议的完整支持，使其在特定场景下具有不可替代的优势。特别是在需要严格控制缓存行为的爬虫项目中，httplib2是一个值得考虑的选择。

关注我，每天推荐一款实用的Python爬虫工具！

更多文章