如何在Scrapy中自动续期60分钟有效期的Access Token？

针对你的需求，这里提供两种实用的实现方式，适配不同场景：

方案一：单个爬虫内实现续期（适合单爬虫场景） #

直接在爬虫类中封装Token管理逻辑，每个请求发起前检查Token有效性，快过期时自动续期：

import time
import json
import scrapy

class MyAPISpider(scrapy.Spider):
    name = "my_api_spider"
    # 初始化Token相关属性
    access_token = None
    token_expires_at = 0  # 存储过期时间戳，初始为0表示未获取Token

    def start_requests(self):
        self.initialize_gcs_store()
        # 先获取有效Token，再启动业务请求
        yield from self.get_access_token(callback=self.start_business_requests)

    def get_access_token(self, callback):
        """获取或续期Token，完成后执行传入的业务回调"""
        current_time = time.time()
        # 提前5分钟续期，避免网络延迟导致请求时Token刚好过期
        if self.access_token and (self.token_expires_at - current_time) > 300:
            # Token仍有效，直接执行业务回调
            yield from callback()
            return
        
        # 需要获取新Token，修复原代码中headers被覆盖的问题
        url = "https://api.example.com/authenticate"
        headers = {
            'Content-Type': 'application/json',
            'authority': 'apply.workable.com'
        }
        payload = {
            "username": "username",
            "password": "password",
        }
        yield scrapy.Request(
            url=url,
            method='POST',
            headers=headers,
            body=json.dumps(payload),
            errback=self.errback,
            callback=self.parse_access_token,
            meta={'callback': callback}  # 传递业务回调到解析函数
        )

    def parse_access_token(self, response):
        token_data = json.loads(response.text)
        self.access_token = token_data["jwt"]
        # 计算过期时间：当前时间 + 60分钟（转秒）
        self.token_expires_at = time.time() + 60 * 60
        # 执行之前传入的业务回调
        callback = response.meta['callback']
        yield from callback()

    def start_business_requests(self):
        """发起实际业务API请求"""
        api_url = "https://api.example.com/data"
        headers = {
            'Authorization': f'Bearer {self.access_token}',
            'Content-Type': 'application/json'
        }
        yield scrapy.Request(
            url=api_url,
            headers=headers,
            callback=self.parse_business_data,
            errback=self.errback
        )

    def parse_business_data(self, response):
        # 处理业务数据逻辑
        data = json.loads(response.text)
        yield data

        # 若存在下一页，先检查Token有效性再发起请求
        next_page = data.get('next_page')
        if next_page:
            yield from self.get_access_token(callback=lambda: [scrapy.Request(
                url=next_page,
                headers={'Authorization': f'Bearer {self.access_token}'},
                callback=self.parse_business_data,
                errback=self.errback
            )])

关键细节 #

• 提前5分钟续期：预留缓冲时间，避免请求发送瞬间Token过期
• 修复原代码的headers覆盖问题：合并两个headers字典，避免Content-Type丢失
• 通过yield from传递回调生成的请求，保证请求被Scrapy正确调度

方案二：下载中间件全局处理（适合多爬虫共享续期逻辑） #

用Scrapy下载中间件统一管理Token续期，所有需要认证的请求自动添加Token，定时续期无需每个爬虫重复写代码：

1. 编写下载中间件 #

import time
import json
from scrapy import signals
from scrapy.http import Request

class TokenRenewMiddleware:
    def __init__(self, crawler):
        self.crawler = crawler
        # 从配置文件读取认证信息
        self.auth_url = crawler.settings.get('AUTH_URL')
        self.username = crawler.settings.get('AUTH_USERNAME')
        self.password = crawler.settings.get('AUTH_PASSWORD')
        self.access_token = None
        self.token_expires_at = 0

    @classmethod
    def from_crawler(cls, crawler):
        middleware = cls(crawler)
        # 爬虫启动时触发Token初始化
        crawler.signals.connect(middleware.spider_opened, signal=signals.spider_opened)
        return middleware

    def spider_opened(self, spider):
        # 爬虫启动后立即获取首个Token
        self.crawler.engine.crawl(self._get_token_request(), spider)

    def _get_token_request(self):
        """生成认证请求"""
        headers = {
            'Content-Type': 'application/json',
            'authority': 'apply.workable.com'
        }
        payload = {
            "username": self.username,
            "password": self.password,
        }
        return Request(
            url=self.auth_url,
            method='POST',
            headers=headers,
            body=json.dumps(payload),
            callback=self._parse_token,
            errback=self._handle_token_error
        )

    def _parse_token(self, response):
        """解析Token并设置续期任务"""
        token_data = json.loads(response.text)
        self.access_token = token_data["jwt"]
        self.token_expires_at = time.time() + 60 * 60
        # 提前5分钟安排续期任务
        self.crawler.engine.crawl(
            Request(
                url='about:blank',  # 用空请求触发续期检查
                callback=self._renew_token,
                meta={'dont_filter': True},
                priority=100  # 优先处理续期请求
            ),
            self.crawler.spider
        )

    def _renew_token(self, response):
        """检查Token有效性，决定是否续期"""
        current_time = time.time()
        if (self.token_expires_at - current_time) > 300:
            # Token仍有效，重新安排下一次续期检查
            self.crawler.engine.crawl(
                Request(
                    url='about:blank',
                    callback=self._renew_token,
                    meta={'dont_filter': True},
                    priority=100
                ),
                self.crawler.spider
            )
            return
        # Token即将过期，发起新的认证请求
        self.crawler.engine.crawl(self._get_token_request(), self.crawler.spider)

    def _handle_token_error(self, failure):
        """Token获取失败时重试"""
        self.crawler.engine.crawl(self._get_token_request(), self.crawler.spider)

    def process_request(self, request, spider):
        """给非认证请求添加Token头"""
        if self.access_token and request.url != self.auth_url:
            request.headers['Authorization'] = f'Bearer {self.access_token}'

2. 配置中间件和认证参数 #

在settings.py中添加以下配置：

# 启用自定义中间件，调整优先级确保在默认中间件前执行
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.TokenRenewMiddleware': 543,
}

# 认证相关配置
AUTH_URL = "https://api.example.com/authenticate"
AUTH_USERNAME = "username"
AUTH_PASSWORD = "password"

方案优势 #

• 全局统一管理：所有爬虫共享Token续期逻辑，无需重复编码
• 自动定时续期：无需在每个请求前手动检查，中间件自动处理
• 错误重试：Token获取失败时自动重试，提升爬虫稳定性

内容的提问来源于stack exchange，提问作者DJJ