如何结合Scrapy CrawlerProcess与Prefect 2？解决主线程信号错误

问题背景 #

我编写了一款抓取新闻的Scrapy爬虫（BankNews类），原本通过CrawlerProcess运行，现在需要封装为Prefect 2的Flow，但运行时出现以下错误：

File "D:\Development\spyingtool\venv\Lib\site-packages\twisted\internet\base.py", line 1282, in _handleSignals
    signal.signal(signal.SIGTERM, reactorBaseSelf.sigTerm)
File "C:\Users\seab.navin\AppData\Local\Programs\Python\Python311\Lib\signal.py", line 56, in signal
    handler = _signal.signal(_enum_to_int(signalnum), _enum_to_int(handler))
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: signal only works in main thread of the main interpreter

相关实现代码 #

1. 原Flow函数

from prefect import flow
from SpyingTools.spiders.bankWebsiteNews import BankNews
from scrapy.crawler import CrawlerProcess
@flow
def bank_website_news():
    settings = get_project_settings()
    process = CrawlerProcess(settings)

    process.crawl(BankNews)
    process.start()

2. BankNews爬虫类

import scrapy
from datetime import date, datetime, timedelta
from SpyingTools.items import NewsItem
from SpyingTools.helpers.spiderHelper import SpiderHelper

class BankNews(scrapy.Spider):
    sph= SpiderHelper()
    name           ='BANKWEBSITE'
    latest_date    = None
    bank           = "ACLEDA"
    index_news     = 2
    image_base_url = 'https://www.acledabank.com.kh/kh'

    custom_settings = {
        'ITEM_PIPELINES': {
            'SpyingTools.pipelines.DataPipeline': 400,
        }
    }

    def generateDaterange(self,start_date=None):
        if start_date==None:
            start_date = date(2021, 1, 1) 
        end_date = datetime.today()
        print(end_date)

        delta = end_date - start_date
        dates = [0]*(delta.days+1 ) 
        for i in range(delta.days + 1):
            day = start_date + timedelta(days=i)
            dates[i] = str(day).replace("-","")
        return dates

    def start_requests(self):
        base_url = "https://www.acledabank.com.kh/kh/eng/md_ln"
        self.latest_date = self.sph.get_latest_date_news(self.bank,self.bank).values[0][0]
        self.latest_date = self.latest_date + timedelta(days=1)
        dates= self.generateDaterange(self.latest_date)
        print(dates)
        for date_ in dates:    
            yield scrapy.Request(base_url+date_, self.parse)

    def parse(self,response):
        news      = NewsItem()
        url_r     = str(response.request.url).split("_ln")[-1]  
        if "_" in url_r:
            self.index_news+=1
        else:
            self.index_news= 2

        container = response.css("div.main")[0]
        div_khmer = container.css('div.font-khm').get()
        h1_khmer  =  container.css('h1.font-khm').get()

        title = container.css('h1::text').get()
        date_ = container.css('p.date::text').get()

        img      = container.css("div.imgbox  img::attr(src)").get()
        img_link = self.image_base_url+ img.split('..')[-1]

        if date_:
            content = "\n".join(container.xpath('p//text()')[7:].getall())
        else:
            content = "\n".join(container.xpath('p//text()')[6:].getall())
        
        yield scrapy.Request(response.request.url+"_"+str(self.index_news), self.parse)
        if div_khmer==None and h1_khmer==None:
            news['time']    =date_
            news['title']   = title
            news['content'] = content
            news['name']    = self.bank.upper()
            news['link']    = response.request.url
            news['image']   = img_link
            news["source"]  = self.bank
            yield news  

3. DataPipeline处理类

import pandas as pd
from datetime import date, datetime
from SpyingTools.pipelines.pipelineBase import PipelineBase
import sqlalchemy

class DataPipeline(PipelineBase):
    def __init__(self) -> None:
        super().__init__()

    def process_item(self, item, spider):
        if isinstance(item, NewsItem):
            df = pd.DataFrame([
                [
                    item['time'],item['title'],
                    item['content'],item["name"],item["link"],item["image"]
                ]
            ],
            columns=["DATE","TITLE","CONTENT","NAME","URL","IMAGE"])
            try:
                df["DATE"] = pd.to_datetime(df["DATE"])
            except:
                df["DATE"] = datetime.now()
        
            if item.get("source"):
                df["WEBSITE"]= item.get("source")
            else:
                df["WEBSITE"] = 'khmertimeskh'

            today = date.today()
            df["DOWNLOAD_DATE"] = today

            df.to_sql("NEWS", self.engine,index=False,if_exists='append',
            dtype={
                "TITLE": sqlalchemy.types.NVARCHAR(),
                "CONTENT":sqlalchemy.types.NVARCHAR(),
                'DATE': sqlalchemy.types.DATETIME(),
                'DOWNLOAD_DATE':sqlalchemy.types.DATE()
            })
            return df
        else:
            return item

解决方法 #

错误根源是：Scrapy的CrawlerProcess默认会让Twisted注册信号处理函数，但Prefect 2的Flow执行线程可能不是主线程，而Python的signal模块仅允许在主线程操作信号，导致冲突。

方案1：禁用Scrapy的信号处理 #

修改Flow函数，在初始化CrawlerProcess时添加install_signal_handlers=False参数，避免Twisted注册信号处理：

from prefect import flow
from SpyingTools.spiders.bankWebsiteNews import BankNews
from scrapy.crawler import CrawlerProcess
from scrapy.utils.project import get_project_settings
from twisted.internet import reactor

@flow
def bank_website_news():
    settings = get_project_settings()
    # 禁用信号处理，避免主线程冲突
    process = CrawlerProcess(settings, install_signal_handlers=False)
    process.crawl(BankNews)
    # 手动启动Twisted反应器
    reactor.run()

方案2：使用CrawlerRunner替代CrawlerProcess #

CrawlerRunner更适合嵌入到现有事件循环中，不会自动安装信号处理，需手动管理反应器生命周期：

from prefect import flow
from SpyingTools.spiders.bankWebsiteNews import BankNews
from scrapy.crawler import CrawlerRunner
from scrapy.utils.project import get_project_settings
from twisted.internet import reactor

@flow
def bank_website_news():
    settings = get_project_settings()
    runner = CrawlerRunner(settings)
    # 启动爬虫
    crawl_deferred = runner.crawl(BankNews)
    # 爬虫完成后停止反应器
    crawl_deferred.addBoth(lambda _: reactor.stop())
    # 启动反应器
    reactor.run()

注意事项 #

• 两种方案都确保Twisted不会在非主线程尝试注册信号，解决signal only works in main thread错误。
• 若Flow需要多次运行，由于每次Flow执行是独立进程，无需担心反应器重复启动问题。
• 确保Prefect 2环境与Scrapy、Twisted版本兼容。

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