Persistência de Dados no Scrapy

Arquivos Principais Envolvidos

items.py

Define a estrutura dos dados coletados. É aqui que se declaram os campos que serão extraídos durante o scraping, utilizando scrapy.Field().

import scrapy

class ArticleItem(scrapy.Item):
    titulo = scrapy.Field()
    autor = scrapy.Field()
    conteudo = scrapy.Field()
    data_publicacao = scrapy.Field()

pipelines.py

Responsável por processar os itens coletados. Após os dados serem extraídos pelo spider e encapsulados em um objeto Item, eles são evniados ao pipeline para operações como limpeza, validação ou armazenamento.

Fluxo de Persistência

1. O spider extrai os dados e os empacota em um objeto Item.
2. O item é enviado ao pipeline usando yield item.
3. O método process_item no pipeline recebe o item e executa ações como salvar em arquivos ou bancos de dados.
4. O arquivo settings.py deve ativar o pipeline na configuração ITEM_PIPELINES.

Métodos de Armazenamento

Exportação para Arquivos (CSV, JSON, etc)

Scrapy permite exportar diretamente os resultados via linha de comando:

scrapy crawl meu_spider -o dados.csv
scrapy crawl meu_spider -o dados.json

Para garantir codificação correta (como UTF-8), adicione ao settings.py:

FEED_EXPORT_ENCODING = 'utf-8'

Observação: Em versões anteriores à 1.6, aqruivos CSV podem conter linhas vazias. Isso pode ser corrigido modificando o parâmetro newline='' no módulo interno exporters.py do Scrapy.

Armazenamento em Banco de Dados

Redis

Armazena itens em uma lista do Redis. Ideal para filas ou cache temporário.

import redis

class RedisPipeline:
    def open_spider(self, spider):
        self.cliente = redis.StrictRedis(host='localhost', port=6379, db=0)

    def process_item(self, item, spider):
        dados = {
            'titulo': item.get('titulo'),
            'autor': item.get('autor')
        }
        self.cliente.lpush('artigos', str(dados))
        return item

MongoDB

Utiliza pymongo para inserir documentos no MongoDB.

from pymongo import MongoClient

class MongoPipeline:
    def __init__(self, host, porta, banco, colecao):
        self.host = host
        self.porta = porta
        self.banco = banco
        self.colecao = colecao

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            host=crawler.settings.get('MONGO_HOST'),
            porta=crawler.settings.get('MONGO_PORT'),
            banco=crawler.settings.get('MONGO_DB'),
            colecao=crawler.settings.get('MONGO_COLLECTION')
        )

    def open_spider(self, spider):
        self.conexao = MongoClient(self.host, self.porta)
        self.db = self.conexao[self.banco]

    def close_spider(self, spider):
        self.conexao.close()

    def process_item(self, item, spider):
        if all(item.values()):
            self.db[self.colecao].insert_one(dict(item))
        return item

MySQL (Síncrono e Assíncrono)

Para escrita síncrona com pymysql:

import pymysql

class MysqlSyncPipeline:
    def __init__(self):
        self.conn = pymysql.connect(host='localhost', user='root', password='senha', db='scrapy_db', charset='utf8mb4')
        self.cursor = self.conn.cursor()

    def process_item(self, item, spider):
        sql = "INSERT INTO artigos (titulo, autor) VALUES (%s, %s)"
        self.cursor.execute(sql, (item['titulo'], item['autor']))
        self.conn.commit()
        return item

    def close_spider(self, spider):
        self.cursor.close()
        self.conn.close()

Para melhor desempenho, use escrita assíncrona com Twisted e adbapi:

from twisted.enterprise import adbapi
import pymysql

class MysqlAsyncPipeline:
    def __init__(self, dbpool):
        self.dbpool = dbpool

    @classmethod
    def from_settings(cls, settings):
        db_config = {
            'host': settings['MYSQL_HOST'],
            'db': settings['MYSQL_DB'],
            'user': settings['MYSQL_USER'],
            'passwd': settings['MYSQL_PASS'],
            'charset': 'utf8mb4',
            'cursorclass': pymysql.cursors.DictCursor,
            'use_unicode': True,
        }
        pool = adbapi.ConnectionPool('pymysql', **db_config)
        return cls(pool)

    def process_item(self, item, spider):
        d = self.dbpool.runInteraction(self._do_insert, item)
        d.addErrback(self._handle_erro, item, spider)
        return item

    def _do_insert(self, cursor, item):
        sql = "INSERT INTO artigos (titulo, autor) VALUES (%(titulo)s, %(autor)s)"
        cursor.execute(sql, dict(item))

    def _handle_erro(self, falha, item, spider):
        print(f"Erro ao inserir: {falha}")

Download de Imagens

O Scrapy oferece um pipeline embutido para baixar imagens automaticamente.

from scrapy.pipelines.images import ImagesPipeline
import scrapy

class CustomImagePipeline(ImagesPipeline):
    def get_media_requests(self, item, info):
        yield scrapy.Request(url=item['url_imagem'])

    def item_completed(self, results, item, info):
        for sucesso, dados in results:
            if sucesso:
                item['caminho_local'] = dados['path']
        return item

No settings.py, configure:

ITEM_PIPELINES = {
    'meuprojeto.pipelines.CustomImagePipeline': 1
}

IMAGES_STORE = '/caminho/local/imagens'
IMAGES_MIN_HEIGHT = 200
IMAGES_MIN_WIDTH = 200

Configuração de Prioridades no Pipeline

Vários ppielines podem ser definidos com níveis de prioridade. O número menor indica maior prioridade:

ITEM_PIPELINES = {
    'meuprojeto.pipelines.MongoPipeline': 100,
    'meuprojeto.pipelines.RedisPipeline': 200,
}

• Se um pipeline retorna item, o próximo é executado.
• Se lançar DropItem(), interrompe o fluxo.
• Pode-se filtrar por nome do spider usando if spider.name == 'nome_spider'.

Exemplo Completo: Salvando em Arquivo de Texto

Spider:

import scrapy
from meuprojeto.items import ArticleItem

class BlogSpider(scrapy.Spider):
    name = 'blog_spider'
    start_urls = ['https://exemplo.com/blog']

    def parse(self, response):
        for post in response.css('.post'):
            item = ArticleItem()
            item['titulo'] = post.css('h2::text').get()
            item['autor'] = post.css('.author::text').get()
            yield item

Pipeline personalizado para arquivos:

class FileSavePipeline:
    def __init__(self, caminho_arquivo):
        self.caminho = caminho_arquivo

    @classmethod
    def from_crawler(cls, crawler):
        caminho = crawler.settings.get('OUTPUT_FILE')
        return cls(caminho)

    def open_spider(self, spider):
        self.arquivo = open(self.caminho, 'w', encoding='utf-8')

    def process_item(self, item, spider):
        self.arquivo.write(f"{item['titulo']} - {item['autor']}\n")
        return item

    def close_spider(self, spider):
        self.arquivo.close()