Python脚本爬虫Session如何保持

wen 实用脚本 1

本文目录导读:

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
方法 优点 缺点 适用场景
requests.Session 简单、自动管理cookies 内存中保存 简单爬虫
手动管理Cookies 灵活控制 代码繁琐 需要精细控制
持久化Cookies 可复用、避免重复登录 需要文件读写 大规模爬虫
上下文管理器 自动清理资源 一次性任务

最佳实践建议

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

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

抱歉,评论功能暂时关闭!