Python脚本爬虫Session如何保持

本文目录导读：

1. 使用 requests.Session() （最常用）
2. 手动管理Cookies
3. 持久化保存Cookies
4. 进阶：使用requests.Session的上下文管理器
5. 处理Session超时和重连
6. 实战示例：完整爬虫Session管理
7. 注意事项

在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

最佳实践建议：

1. 优先使用 requests.Session
2. 需要断点续爬时，使用pickle持久化cookies
3. 多线程场景使用Session池
4. 总是设置合理的User-Agent和请求头
5. 添加错误重试机制

选择哪种方式取决于你的具体需求，但绝大多数情况下 requests.Session() 已经足够使用。