专题：Python 自动化办公系统学习
关键词：Python, 自动化办公, requests库, HTTP请求, API交互, RESTful, Session, JSON, Python网络
一、requests概述
requests是Python生态中最流行的HTTP客户端库，由Kenneth Reitz于2011年创建，口号是"为人类设计的HTTP库"。它提供了极其简洁的API，让开发者可以用最少的代码完成HTTP请求的所有操作。requests基于urllib3实现，但封装了底层复杂的连接管理、连接池、SSL握手等细节，让开发者只需关注业务逻辑。
在自动化办公场景中，requests是不可或缺的核心组件。无论是调用企业微信/钉钉的Webhook发送通知，对接RESTful API同步数据，还是定时抓取外部系统的报表数据，requests都能胜任。它的跨平台特性和对Python 3.6+的全面支持，使其成为Python网络编程的事实标准。
安装requests非常简单，通过pip即可完成：
pip install requests
安装完成后，验证版本：
import requests
print(requests.__version__)
HTTP协议基础回顾
HTTP（超文本传输协议）是客户端与服务器之间通信的协议。一个完整的HTTP请求包含：请求方法（Method）、请求URL（Uniform Resource Locator）、请求头（Headers）、请求体（Body）四个组成部分。响应则包含状态码（Status Code）、响应头（Response Headers）和响应体（Response Body）三部分。
常见的HTTP请求方法有八种，其中最常用的是：GET用于获取资源，POST用于创建资源，PUT用于替换资源，PATCH用于部分更新资源，DELETE用于删除资源。每种方法在RESTful API设计中都有明确的语义约定。
URL的结构由协议（scheme）、主机（host）、端口（port）、路径（path）、查询参数（query string）和锚点（fragment）组成。例如 https://api.example.com:443/v1/users?page=1&limit=20#section，其中查询参数是键值对形式，用于向服务器传递过滤、分页等参数。
requests库的核心理念
requests的设计哲学是"让一切变得简单"。它自动处理了字符编码检测、Cookie持久化、SSL证书验证、重定向跟随等常见痛点。requests的API采用链式调用风格，语义清晰，即使是初学者也能快速上手。它内置了JSON解码器，对现代Web API提供了天然的支持。
import requests
# 最简单的GET请求 —— 一行代码完成
response = requests.get('https://httpbin.org/get')
print(response.status_code) # 200
print(response.json()) # 自动解析JSON响应
# 检查requests支持的请求方法
import requests
methods = ['get', 'post', 'put', 'delete', 'patch', 'head', 'options']
for method in methods:
func = getattr(requests, method)
print(f"requests.{method}: {func.__doc__[:50]}...")
核心理解：requests的强大之处不在于实现了多少功能，而在于把复杂网络操作的细节封装成直观的Python函数调用。它是Python"简洁即美"哲学的完美体现。
二、基础请求
requests为每一种HTTP请求方法提供了对应的顶级函数，包括requests.get()、requests.post()、requests.put()、requests.delete()、requests.patch()等。这些函数接受相似的参数，降低了学习成本。核心参数包括：url（请求地址）、params（查询参数）、data（表单数据）、json（JSON数据）、headers（请求头）、cookies（Cookie）等。
GET请求
GET请求用于从服务器获取资源，是使用最频繁的请求类型。GET请求的参数通过URL的查询字符串传递，适合传递非敏感数据。requests的params参数自动将字典转换为URL查询字符串格式，并对特殊字符进行URL编码。
import requests
# 带查询参数的GET请求
params = {
'q': 'requests library',
'page': 1,
'per_page': 20,
'sort': 'stars'
}
response = requests.get(
'https://api.github.com/search/repositories',
params=params
)
print(f"请求URL: {response.url}")
print(f"状态码: {response.status_code}")
data = response.json()
print(f"找到 {data['total_count']} 个仓库")
POST请求
POST请求用于向服务器提交数据，创建新的资源。requests支持多种数据格式的提交：使用data参数发送表单编码数据（application/x-www-form-urlencoded），使用json参数发送JSON数据（application/json），使用files参数发送multipart/form-data文件数据。现代RESTful API普遍使用JSON格式通信。
import requests
# POST请求 - 发送JSON数据
url = 'https://jsonplaceholder.typicode.com/posts'
payload = {
'title': 'Python requests学习笔记',
'body': '这是一篇关于requests库的详细学习笔记...',
'userId': 1
}
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
response = requests.post(url, json=payload, headers=headers)
print(f"创建成功，ID: {response.json()['id']}")
# POST请求 - 发送表单数据
form_data = {'username': 'admin', 'password': 'secret'}
response = requests.post('https://httpbin.org/post', data=form_data)
print(response.json()['form'])
PUT / DELETE / PATCH请求
PUT请求用于完全替换服务器上的资源，DELETE请求用于删除资源，PATCH请求用于部分更新资源。这三者在RESTful API设计中非常重要，requests对它们都提供了直接的方法支持。
import requests
base_url = 'https://jsonplaceholder.typicode.com/posts/1'
# PUT - 完全替换资源
put_response = requests.put(base_url, json={
'id': 1,
'title': '新标题',
'body': '新内容...',
'userId': 1
})
print(f"PUT状态码: {put_response.status_code}")
# PATCH - 部分更新资源
patch_response = requests.patch(base_url, json={'title': '仅更新标题'})
print(f"PATCH状态码: {patch_response.status_code}")
# DELETE - 删除资源
delete_response = requests.delete(base_url)
print(f"DELETE状态码: {delete_response.status_code}")
实践建议：在调用API时，务必根据语义选择正确的HTTP方法。错误的请求方法不仅可能导致服务器拒绝响应，还会使代码的意图变得模糊不清。GET请求不应修改服务器数据，POST/PUT/PATCH请求注意数据格式与Content-Type匹配。
三、响应处理
requests的响应对象（Response）封装了服务器返回的所有信息。通过这个对象，开发者可以获取状态码、响应头、响应体（文本/字节/JSON格式）、响应时间、编码方式、重定向历史等丰富信息。正确解析和处理响应是构建健壮网络应用的关键。
响应对象的核心属性
response.status_code返回整数状态码，配合requests.codes可以判断请求是否成功。response.text返回解码后的字符串内容（requests自动检测编码），response.content返回原始字节内容（适合二进制数据如图片文件），response.json()直接解析JSON响应为Python字典/列表（如解析失败抛出异常）。response.headers是一个大小写不敏感的字典，用于获取服务器返回的头信息。
import requests
response = requests.get('https://httpbin.org/response-headers?freeform=hello')
# 状态码判断
if response.status_code == 200:
print("请求成功！")
elif response.status_code == 404:
print("资源不存在")
elif response.status_code == 500:
print("服务器内部错误")
# 使用内置的ok属性判断
if response.ok: # 状态码小于400
data = response.json()
print("响应解析成功")
# 响应头信息
print(f"响应时间: {response.elapsed.total_seconds():.2f}秒")
print(f"内容类型: {response.headers.get('Content-Type')}")
编码处理与响应时间
requests在收到响应时会根据Content-Type头部的charset字段自动解码文本内容。如果自动检测的编码不正确，可以手动指定response.encoding属性。响应时间（response.elapsed）返回一个timedelta对象，记录从发送请求到收到响应头的时间差，可用于性能监控和超时调优。
import requests
# 处理不同编码的响应
response = requests.get('https://httpbin.org/get')
# 查看自动检测的编码
print(f"自动检测编码: {response.apparent_encoding}")
print(f"实际使用编码: {response.encoding}")
# 手动指定编码（当自动检测不正确时）
response.encoding = 'utf-8'
text = response.text
# 响应时间性能监控
urls = ['https://httpbin.org/delay/1', 'https://httpbin.org/delay/2']
for url in urls:
resp = requests.get(url, timeout=5)
print(f"请求 {url} 耗时: {resp.elapsed.total_seconds():.2f}s")
requests的响应对象设计得非常人性化。自动JSON解析、自动编码检测、大小写不敏感的headers字典，这些细节让开发者在处理真实API时可以少写大量样板代码。
四、Session与Cookie
requests.Session对象提供了一种在多个请求之间保持状态的机制。与每次独立使用requests.get()/post()不同，Session会持久化Cookie信息、使用同一个TCP连接池（减少连接建立的开销）、复用HTTP连接参数（如headers、proxies、auth等）。在需要登录态保持的自动化任务中，Session是不可或缺的工具。
Session的核心能力
Session对象最主要的三个能力是：Cookie持久化（自动保存服务器返回的Set-Cookie，后续请求自动携带）、连接复用（使用urllib3的连接池机制，减少三次握手开销）、参数继承（在Session级别设置的headers、params、auth等参数会自动应用到所有请求上）。这使得Session在爬虫、API客户端、自动化测试等场景中表现优异。
import requests
# 创建Session
session = requests.Session()
# 设置Session级默认参数
session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
'Accept-Language': 'zh-CN,zh;q=0.9'
})
# 模拟登录过程
login_data = {'username': 'test_user', 'password': 'test_pass'}
# 第一次请求 —— 登录（Cookie会被Session自动保存）
login_resp = session.post('https://httpbin.org/post', data=login_data)
print(f"登录状态码: {login_resp.status_code}")
# 后续请求无需再携带认证信息，Session自动管理Cookie
profile_resp = session.get('https://httpbin.org/cookies')
print(f"Cookie已自动携带: {profile_resp.json()}")
手动管理Cookie
除了Session自动管理Cookie外，requests还支持手动设置和读取Cookie。可以直接传递Cookie字典给请求，也可以使用http.cookiejar进行更精细的控制。在需要从外部持久化存储（如文件、数据库）中恢复登录状态时，手动管理Cookie非常实用。
import requests
import json
# 手动设置Cookie
custom_cookies = {
'session_id': 'abc123def456',
'user_token': 'xyz789',
'preferences': 'lang=zh'
}
response = requests.get(
'https://httpbin.org/cookies',
cookies=custom_cookies
)
print("发送的Cookie:", response.json())
# 导出Session中的Cookie为字典
session = requests.Session()
session.get('https://httpbin.org/cookies/set?name=value')
# 将Cookie字典保存到文件实现持久化
cookie_dict = requests.utils.dict_from_cookiejar(session.cookies)
with open('cookies.json', 'w') as f:
json.dump(cookie_dict, f)
# 从文件恢复Cookie
with open('cookies.json', 'r') as f:
restored_cookies = json.load(f)
new_session = requests.Session()
new_session.cookies.update(restored_cookies)
注意事项：Session对象不是线程安全的。在多线程环境下使用同一个Session实例会导致数据竞争。建议每个线程创建独立的Session实例，或者使用requests的thread-local session模式。另外，Session用完应调用.close()释放连接池资源，或使用上下文管理器（with语句）。
五、高级请求
在实际工作中，简单的GET/POST请求往往不足以满足需求。文件上传下载、超时控制、代理配置、SSL证书验证、流式请求等高级功能，在自动化办公场景中频繁出现。requests对所有这些场景都提供了完整的支持。
文件上传与下载
文件上传使用files参数，requests会自动构建multipart/form-data格式的请求体。文件下载建议使用流式模式（stream=True），配合iter_content或iter_lines迭代下载，避免大文件耗尽内存。requests还支持上传文件对象、字节数据、元组等多种格式。
import requests
# 文件上传 - 上传单个文件
with open('report.xlsx', 'rb') as f:
files = {'file': (f.name, f, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')}
response = requests.post('https://httpbin.org/post', files=files)
print(response.json()['files'])
# 文件下载 - 流式下载大文件（防止内存溢出）
url = 'https://httpbin.org/image/png'
response = requests.get(url, stream=True)
response.raise_for_status()
with open('downloaded_image.png', 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk: # 过滤掉keep-alive新块
f.write(chunk)
print("大文件下载完成")
超时、代理与SSL验证
超时控制是生产环境中必不可少的防护措施。requests支持连接超时和读取超时的分别设置。代理配置支持HTTP、HTTPS和SOCKS协议（需安装requests[socks]）。SSL验证可以控制证书检查行为，在开发环境可临时关闭，但生产环境应始终保持。
import requests
# 超时设置（连接超时5秒，读取超时15秒）
try:
response = requests.get(
'https://httpbin.org/delay/10',
timeout=(5, 15) # (connect_timeout, read_timeout)
)
except requests.Timeout:
print("请求超时！")
# 代理配置
proxies = {
'http': 'http://127.0.0.1:7890',
'https': 'http://127.0.0.1:7890'
}
response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=10)
print(f"通过代理访问，IP: {response.json()['origin']}")
# SSL证书验证控制
import certifi
# 使用自定义CA证书
response = requests.get('https://example.com', verify=certifi.where())
# 仅限开发环境：关闭SSL验证（生产环境禁止！）
# response = requests.get('https://self-signed.badssl.com', verify=False)
安全提醒：在生产环境中永远不要设置verify=False。这会禁用SSL证书验证，使你的请求容易受到中间人攻击。如果遇到自签名证书的问题，应该将证书添加到信任链中，或使用verify参数指向自定义CA bundle文件。
六、HTTP认证
现代API普遍要求请求携带认证信息。requests提供了从简单到复杂的多层认证支持：从基础的HTTP Basic Auth，到安全的Digest Auth，再到OAuth1/OAuth2、Bearer Token、API Key，甚至允许自定义认证实现。理解不同认证机制的原理和适用场景，是安全调用API的基础。
BasicAuth与Bearer Token
BasicAuth是最基础的HTTP认证方式，将"用户名:密码"进行Base64编码后放在Authorization头中。Bearer Token是OAuth2.0中最常用的认证方式，客户端持有token字符串，通过Authorization: Bearer 头发送给服务器。相比BasicAuth，Bearer Token可以设置有效期和权限范围，更加安全灵活。
import requests
from requests.auth import HTTPBasicAuth
# HTTP Basic Auth
response = requests.get(
'https://httpbin.org/basic-auth/user/passwd',
auth=HTTPBasicAuth('user', 'passwd')
)
print(f"BasicAuth结果: {response.json()}")
# auth参数支持简写元组形式
response = requests.get(
'https://httpbin.org/basic-auth/user/passwd',
auth=('user', 'passwd') # 自动使用BasicAuth
)
# Bearer Token认证
token = 'your_jwt_token_here'
headers = {'Authorization': f'Bearer {token}'}
response = requests.get(
'https://api.github.com/user',
headers=headers
)
print(f"Bearer Token认证: {response.status_code}")
OAuth与API Key认证
OAuth1需要签名请求（使用oauthlib库），OAuth2则通过授权码流程获取access_token。API Key是最简单的方式，将密钥放在请求头（X-API-Key）、查询参数或请求体中。不同的API服务商采用不同的Key传递方式，requests的灵活性可以轻松适配。
import requests
from requests_oauthlib import OAuth1
# OAuth1认证（如Twitter API）
# 需安装: pip install requests_oauthlib
"""
oauth = OAuth1(
client_key='your_app_key',
client_secret='your_app_secret',
resource_owner_key='user_token',
resource_owner_secret='user_token_secret'
)
response = requests.get('https://api.twitter.com/1.1/account/settings.json', auth=oauth)
"""
# API Key认证（最常见的三种方式）
# 1. Header方式
headers = {'X-API-Key': 'sk-xxxxxxxxxxxxxxxx'}
response = requests.get('https://api.example.com/v1/data', headers=headers)
# 2. Query参数方式
params = {'api_key': 'sk-xxxxxxxxxxxxxxxx'}
response = requests.get('https://api.example.com/v1/data', params=params)
# 3. 自定义认证器 —— 封装重复的认证逻辑
from requests.auth import AuthBase
class APIKeyAuth(AuthBase):
def __init__(self, api_key):
self.api_key = api_key
def __call__(self, request):
request.headers['X-API-Key'] = self.api_key
return request
response = requests.get(
'https://api.example.com/v1/data',
auth=APIKeyAuth('sk-xxxxxxxx')
)
安全实践：API Key和Token绝对不要硬编码在源代码中。应使用环境变量（os.environ）、配置文件（.env文件）、密钥管理服务（如AWS Secrets Manager、Vault）或Python的getpass模块来管理敏感信息。
七、异常与重试
网络请求天生具有不确定性。DNS解析失败、连接中断、服务器超载、HTTP 5xx错误都是常态。requests定义了一套完整的异常体系，配合urllib3的Retry机制，可以构建健壮的容错网络应用。合理处理异常和实现重试策略，是生产级网络应用的必备能力。
异常体系与捕获
requests的异常继承自requests.RequestException，主要子类包括：ConnectionError（网络连接失败，如DNS解析失败、拒绝连接）、Timeout（连接或读取超时）、HTTPError（HTTP状态码错误，配合response.raise_for_status()使用）、TooManyRedirects（超过最大重定向次数）、URLRequired（无效URL）等。正确捕获这些异常可以区分不同类型的错误。
import requests
from requests.exceptions import RequestException, ConnectionError, Timeout, HTTPError
# 全面的异常捕获
urls = [
'https://httpbin.org/status/500', # 服务器错误
'https://nonexistent-domain-12345.com', # DNS解析失败
'https://httpbin.org/delay/10', # 超时
'https://httpbin.org/get' # 正常请求
]
for url in urls:
try:
response = requests.get(url, timeout=3)
response.raise_for_status() # 状态码非2xx时抛出HTTPError
print(f"成功: {url}")
except Timeout:
print(f"超时: {url}")
except ConnectionError:
print(f"连接失败: {url}")
except HTTPError as e:
print(f"HTTP错误 {e.response.status_code}: {url}")
except RequestException as e:
print(f"其他请求错误: {url} -> {e}")
重试机制（Retry适配器）
requests本身不直接提供重试功能，但可以通过HTTPAdapter配合urllib3的Retry对象实现。Retry支持设置最大重试次数、重试的条件（如500/502/503/504状态码、连接错误、超时等）、重试间隔（支持指数退避策略）、以及特定HTTP方法的重试策略。指数退避是指每次重试的等待时间按指数增长，防止对服务器造成冲击。
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 配置重试策略
retry_strategy = Retry(
total=3, # 总重试次数（不包括首次）
backoff_factor=0.5, # 重试间隔: 0.5, 1, 2, 4, 8... 秒（指数退避）
status_forcelist=[500, 502, 503, 504], # 哪些状态码需要重试
allowed_methods=['GET', 'POST'], # 对哪些HTTP方法启用重试
raise_on_status=True # 重试耗尽后抛出异常
)
# 创建配置了重试的Session
session = requests.Session()
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('https://', adapter)
session.mount('http://', adapter)
# 使用带重试的Session发送请求
try:
response = session.get('https://httpbin.org/status/500', timeout=10)
print(f"最终状态码: {response.status_code}")
except requests.exceptions.RetryError as e:
print("重试耗尽，请求最终失败")
设计原则：重试不是万能的。对GET等幂等请求可以放心重试；对POST等非幂等请求，需要确认API是否支持幂等性（如通过请求中的idempotency_key），否则可能导致重复下单等严重问题。建议：GET重试3次，POST只在确认幂等时重试。
八、频率控制
在调用第三方API或进行网络爬虫时，频率控制（Rate Limiting）至关重要。过高的请求频率会导致IP被封禁、API Key被吊销、服务器负载过高等问题。专业的频率控制策略包括：固定间隔请求、令牌桶算法、滑动窗口计数、自适应限流等。Python生态中既有简单的sleep方案，也有专业的限流库。
基础频率控制方法
最基础的频率控制就是time.sleep()，简单直接。但固定间隔无法应对突发流量和服务器响应时间波动。更高级的方案是令牌桶算法：以固定速率向桶中添加令牌，每次请求消耗一个令牌，桶满则拒绝新请求。这种方案允许一定程度的突发请求，同时保证长期平均速率可控。
import requests
import time
# 方法1: 固定间隔（最简单）
urls = ['https://httpbin.org/get'] * 5
for url in urls:
response = requests.get(url)
print(f"请求完成: {response.status_code}")
time.sleep(1) # 每秒最多1个请求
# 方法2: 令牌桶算法实现
import threading
class TokenBucket:
def __init__(self, rate, capacity):
self.rate = rate # 令牌添加速率（个/秒）
self.capacity = capacity # 桶容量
self.tokens = capacity # 当前令牌数
self.last_time = time.time()
self.lock = threading.Lock()
def consume(self, tokens=1):
with self.lock:
now = time.time()
# 添加这段时间产生的新令牌
self.tokens = min(self.capacity,
self.tokens + (now - self.last_time) * self.rate)
self.last_time = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
# 使用令牌桶：每秒最多2个请求，桶容量3
bucket = TokenBucket(rate=2, capacity=3)
for i in range(10):
while not bucket.consume():
time.sleep(0.1) # 等待令牌
response = requests.get('https://httpbin.org/get')
print(f"请求 {i+1}: {response.status_code}")
并发请求限流与爬虫礼貌策略
在多线程或异步场景下，需要全局控制所有线程的总请求频率。可以使用信号量限制并发数，结合令牌桶控制总速率。爬虫礼貌策略还包括：遵守robots.txt规则、设置合理的User-Agent、监控响应状态码（4xx/5xx时降低频率）、随机化请求间隔避免被识别为机器人。
import requests
import time
import random
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
# 并发请求 + 全局限流
class RateLimiter:
def __init__(self, max_concurrent=3, min_interval=0.5):
self.semaphore = threading.Semaphore(max_concurrent)
self.min_interval = min_interval
self.last_request = 0.0
self.lock = threading.Lock()
def acquire(self):
with self.lock:
elapsed = time.time() - self.last_request
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request = time.time()
self.semaphore.acquire()
def release(self):
self.semaphore.release()
def crawl_url(url, limiter):
limiter.acquire()
try:
# 随机化User-Agent和请求间隔
headers = {'User-Agent': random.choice([
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
])}
response = requests.get(url, headers=headers, timeout=10)
return (url, response.status_code, len(response.text))
finally:
limiter.release()
# 模拟爬取多个URL
urls = ['https://httpbin.org/get'] * 10
limiter = RateLimiter(max_concurrent=3, min_interval=0.3)
with ThreadPoolExecutor(max_workers=5) as executor:
futures = [executor.submit(crawl_url, url, limiter) for url in urls]
for future in as_completed(futures):
url, status, size = future.result()
print(f"完成: {url} -> {status} ({size} bytes)")
礼仪准则：好的爬虫与坏爬虫的区别不在于技术能力，而在于是否尊重服务器。始终检查robots.txt、设置合理的请求间隔、控制并发数、携带清晰的User-Agent（标明用途和联系方式）。尊重API的Rate Limit响应头（X-RateLimit-*），收到429状态码时立即降速。
九、实战案例
理论知识的最终目的是解决实际问题。本节将通过三个完整的实战案例，展示requests在真实场景中的应用：RESTful API客户端封装（通用的CRUD操作基类）、天气数据自动获取（调用公开API采集数据并存储）、RSS订阅自动抓取（解析XML并提取结构化信息）。每个案例都遵循生产级代码规范，包含异常处理和日志记录。
案例1：RESTful API客户端封装
将API调用封装为客户端类，可以集中管理base_url、认证信息、请求头、错误处理等公共逻辑。通过继承和多态，可以快速适配不同的后端服务。下面实现一个通用的API客户端基类，支持完整的CRUD操作，并集成重试和日志功能。
import requests
import logging
from typing import Optional, Dict, Any
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
class APIClient:
"""通用的RESTful API客户端封装"""
def __init__(self, base_url: str, api_key: Optional[str] = None):
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.logger = logging.getLogger(self.__class__.__name__)
if api_key:
self.session.headers['Authorization'] = f'Bearer {api_key}'
self.session.headers['Content-Type'] = 'application/json'
self.session.headers['Accept'] = 'application/json'
def _request(self, method: str, endpoint: str,
**kwargs) -> Dict[str, Any]:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
response = self.session.request(method, url, timeout=10, **kwargs)
response.raise_for_status()
self.logger.info(f"{method} {endpoint} -> {response.status_code}")
return response.json()
except requests.exceptions.RequestException as e:
self.logger.error(f"{method} {endpoint} 失败: {e}")
raise
def get(self, endpoint: str, params: Optional[Dict] = None):
return self._request('GET', endpoint, params=params)
def post(self, endpoint: str, data: Optional[Dict] = None):
return self._request('POST', endpoint, json=data)
def put(self, endpoint: str, data: Optional[Dict] = None):
return self._request('PUT', endpoint, json=data)
def delete(self, endpoint: str):
return self._request('DELETE', endpoint)
def close(self):
self.session.close()
案例2：天气数据自动获取
调用和风天气（或OpenWeatherMap）的公开API，定时获取指定城市的天气数据，并保存为JSON或CSV格式。这个案例演示了：API Key的管理方式、参数传递、JSON解析、数据持久化、以及定时任务的配合。这是自动化办公中"数据采集"环节的典型场景。
import requests
import json
import os
import time
from datetime import datetime
class WeatherFetcher:
"""天气数据自动获取器"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = 'https://api.openweathermap.org/data/2.5'
def get_weather(self, city: str) -> dict:
params = {
'q': city,
'appid': self.api_key,
'units': 'metric', # 摄氏度
'lang': 'zh_cn' # 中文返回
}
response = requests.get(
f"{self.base_url}/weather",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
return {
'city': data['name'],
'temperature': data['main']['temp'],
'humidity': data['main']['humidity'],
'pressure': data['main']['pressure'],
'description': data['weather'][0]['description'],
'wind_speed': data['wind']['speed'],
'timestamp': datetime.now().isoformat()
}
def save_weather_report(self, cities: list, output_file: str = 'weather_report.json'):
"""获取多个城市的天气并保存到文件"""
report = {'generated_at': datetime.now().isoformat(), 'cities': {}}
for city in cities:
try:
weather = self.get_weather(city)
report['cities'][city] = weather
print(f"[OK] {city}: {weather['temperature']}°C, {weather['description']}")
except Exception as e:
print(f"[ERR] {city}: {e}")
time.sleep(1) # 礼貌间隔
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(f"天气报告已保存至: {output_file}")
案例3：RSS订阅自动抓取
RSS是传统的内容聚合格式，很多新闻网站和老牌博客仍然提供RSS订阅。通过requests获取XML数据，结合Python标准库xml.etree.ElementTree解析，可以自动采集最新的文章列表。这个案例展示了：文本内容请求（非JSON响应）、XML解析、结构化提取、以及增量更新的思路。
import requests
import xml.etree.ElementTree as ET
from datetime import datetime
class RSSReader:
"""RSS订阅自动抓取器"""
def __init__(self, user_agent: str = None):
self.session = requests.Session()
self.session.headers['User-Agent'] = user_agent or (
'Mozilla/5.0 (compatible; RSSReader/1.0; +https://example.com/bot)'
)
def fetch_feed(self, rss_url: str) -> list:
"""获取RSS源并解析为文章列表"""
response = self.session.get(rss_url, timeout=15)
response.raise_for_status()
# 解析XML
root = ET.fromstring(response.content)
# RSS 2.0格式: /rss/channel/item
# Atom格式: /feed/entry
articles = []
# 尝试解析RSS 2.0
items = root.findall('.//item')
for item in items:
article = {
'title': item.findtext('title', ''),
'link': item.findtext('link', ''),
'description': item.findtext('description', '')[:200],
'pub_date': item.findtext('pubDate', ''),
'source': rss_url
}
articles.append(article)
return articles
def fetch_multiple(self, rss_urls: list) -> dict:
"""同时抓取多个RSS源"""
results = {}
for url in rss_urls:
try:
articles = self.fetch_feed(url)
results[url] = articles
print(f"[OK] {url}: 获取 {len(articles)} 篇文章")
except Exception as e:
print(f"[ERR] {url}: {e}")
results[url] = []
return results
def close(self):
self.session.close()
案例总结：以上三个案例覆盖了requests在自动化办公中的核心使用模式：API客户端封装实现了代码复用和统一错误处理；天气数据采集展示了定时数据管道的构建方法；RSS抓取说明了非JSON内容类型的处理方式。将这些模式组合起来，可以构建复杂的自动化工作流。