本文目录导读:

- 使用
requests.Session()(最常用) - 手动管理Cookies
- 持久化保存Cookies
- 进阶:使用
requests.Session的上下文管理器 - 处理Session超时和重连
- 实战示例:完整爬虫Session管理
- 注意事项
在Python爬虫中保持Session(会话)主要有以下几种方式,我按推荐程度和使用场景详细介绍:
使用 requests.Session() (最常用)
import requests
# 创建Session对象
session = requests.Session()
# 设置请求头(可持久化)
session.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept-Language': 'zh-CN,zh;q=0.9',
}
# 第一次登录(获取cookies)
login_data = {
'username': 'your_username',
'password': 'your_password'
}
response = session.post('https://example.com/login', data=login_data)
# 后续请求自动携带cookies
response2 = session.get('https://example.com/profile')
response3 = session.get('https://example.com/dashboard')
# Session对象会自动处理cookies
手动管理Cookies
从响应中提取Cookies
import requests
session = requests.Session()
# 获取登录后的cookies
response = session.post('https://example.com/login')
cookies = response.cookies
# 后续请求手动携带cookies
response2 = requests.get('https://example.com/profile', cookies=cookies)
使用CookieJar
import requests
from http.cookiejar import CookieJar
# 创建CookieJar
cookie_jar = CookieJar()
session = requests.Session()
session.cookies = cookie_jar
# 或者直接设置
session.cookies.set('name', 'value', domain='example.com')
持久化保存Cookies
保存到文件
import requests
import pickle
session = requests.Session()
# 登录获取cookies
session.post('https://example.com/login')
# 保存cookies到文件
with open('cookies.pkl', 'wb') as f:
pickle.dump(session.cookies, f)
# 下次运行时加载cookies
with open('cookies.pkl', 'rb') as f:
session.cookies.update(pickle.load(f))
# 使用保存的cookies
response = session.get('https://example.com/protected')
使用字典形式保存
import json
import requests
session = requests.Session()
session.post('https://example.com/login')
# 转为字典保存
cookies_dict = session.cookies.get_dict()
with open('cookies.json', 'w') as f:
json.dump(cookies_dict, f)
# 加载cookies
with open('cookies.json', 'r') as f:
cookies_dict = json.load(f)
session.cookies.update(cookies_dict)
进阶:使用requests.Session的上下文管理器
import requests
with requests.Session() as session:
session.headers.update({'User-Agent': 'Mozilla/5.0'})
# 执行一系列请求
session.post('https://example.com/login')
response1 = session.get('https://example.com/page1')
response2 = session.get('https://example.com/page2')
# 退出上下文后,session自动关闭
处理Session超时和重连
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
# 配置重试策略
retry_strategy = Retry(
total=3, # 总重试次数
backoff_factor=1, # 重试间隔
status_forcelist=[500, 502, 503, 504], # 需要重试的状态码
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('http://', adapter)
session.mount('https://', adapter)
# 设置超时
session.timeout = 10 # 全局超时
try:
response = session.get('https://example.com/page')
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
实战示例:完整爬虫Session管理
import requests
import pickle
import os
from typing import Dict, Optional
class SessionManager:
def __init__(self, cookie_file='cookies.pkl'):
self.session = requests.Session()
self.cookie_file = cookie_file
# 设置默认请求头
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/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',
})
# 加载保存的cookies
self._load_cookies()
def _load_cookies(self):
"""从文件加载cookies"""
if os.path.exists(self.cookie_file):
with open(self.cookie_file, 'rb') as f:
self.session.cookies.update(pickle.load(f))
def _save_cookies(self):
"""保存cookies到文件"""
with open(self.cookie_file, 'wb') as f:
pickle.dump(self.session.cookies, f)
def login(self, url: str, data: Dict) -> bool:
"""登录并保存cookies"""
try:
response = self.session.post(url, data=data)
if response.status_code == 200:
self._save_cookies()
return True
return False
except Exception as e:
print(f"登录失败: {e}")
return False
def get(self, url: str, params: Optional[Dict] = None) -> Optional[requests.Response]:
"""带session的GET请求"""
try:
response = self.session.get(url, params=params)
# 如果401,尝试重新登录
if response.status_code == 401:
print("Session过期,需要重新登录")
return None
return response
except Exception as e:
print(f"请求失败: {e}")
return None
def post(self, url: str, data: Dict) -> Optional[requests.Response]:
"""带session的POST请求"""
try:
response = self.session.post(url, data=data)
return response
except Exception as e:
print(f"请求失败: {e}")
return None
def close(self):
"""关闭session"""
self._save_cookies()
self.session.close()
# 使用示例
if __name__ == "__main__":
manager = SessionManager()
# 登录
if manager.login('https://example.com/login', {
'username': 'test',
'password': '123456'
}):
# 访问需要登录的页面
response = manager.get('https://example.com/profile')
if response:
print(response.text)
# 关闭session
manager.close()
注意事项
User-Agent伪装
# 随机User-Agent from fake_useragent import UserAgent ua = UserAgent() session.headers['User-Agent'] = ua.random
处理重定向
# 自动跟随重定向
response = session.get('https://example.com', allow_redirects=True)
# 不自动跟随
response = session.get('https://example.com', allow_redirects=False)
Session池管理(多线程)
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
class SessionPool:
def __init__(self, size=5):
self.pool = Queue(maxsize=size)
for _ in range(size):
self.pool.put(requests.Session())
def get_session(self):
return self.pool.get()
def return_session(self, session):
self.pool.put(session)
处理验证码
# 一些网站需要验证码,可以尝试使用OCR或打码平台
def solve_captcha(image_path):
# 使用第三方打码服务
pass
| 方法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
requests.Session |
简单、自动管理cookies | 内存中保存 | 简单爬虫 |
| 手动管理Cookies | 灵活控制 | 代码繁琐 | 需要精细控制 |
| 持久化Cookies | 可复用、避免重复登录 | 需要文件读写 | 大规模爬虫 |
| 上下文管理器 | 自动清理资源 | 一次性任务 |
最佳实践建议:
- 优先使用
requests.Session - 需要断点续爬时,使用pickle持久化cookies
- 多线程场景使用Session池
- 总是设置合理的User-Agent和请求头
- 添加错误重试机制
选择哪种方式取决于你的具体需求,但绝大多数情况下 requests.Session() 已经足够使用。