Web Scraping Avanzado con Scrapy y Playwright en Python

Web Scraping Avanzado con Scrapy y Playwright en Python

Para scraping serio a escala se necesita más que requests + BeautifulSoup. Scrapy es el framework de scraping más completo de Python; Playwright permite renderizar JavaScript en páginas modernas.

Scrapy — instalación y estructura del proyecto

pip install scrapy
scrapy startproject mi_scraper
cd mi_scraper
scrapy genspider productos tienda.com

mi_scraper/
├── scrapy.cfg
└── mi_scraper/
    ├── settings.py
    ├── items.py
    ├── middlewares.py
    ├── pipelines.py
    └── spiders/
        └── productos.py

Spider básico

# spiders/productos.py
import scrapy
from mi_scraper.items import ProductoItem

class ProductosSpider(scrapy.Spider):
    name = 'productos'
    allowed_domains = ['books.toscrape.com']
    start_urls = ['https://books.toscrape.com/catalogue/page-1.html']

    custom_settings = {
        'DOWNLOAD_DELAY': 1.5,          # esperar 1.5s entre peticiones
        'RANDOMIZE_DOWNLOAD_DELAY': True,
        'CONCURRENT_REQUESTS_PER_DOMAIN': 2,
    }

    def parse(self, response):
        # Extraer cada producto de la página
        for libro in response.css('article.product_pod'):
            item = ProductoItem()
            item['titulo'] = libro.css('h3 a::attr(title)').get()
            item['precio'] = libro.css('p.price_color::text').get()
            item['rating'] = libro.css('p.star-rating::attr(class)').get().split()[-1]
            item['url']    = response.urljoin(libro.css('h3 a::attr(href)').get())

            # Seguir al detalle del producto
            yield response.follow(item['url'], callback=self.parse_detalle,
                                  cb_kwargs={'item': item})

        # Paginación automática
        siguiente = response.css('li.next a::attr(href)').get()
        if siguiente:
            yield response.follow(siguiente, callback=self.parse)

    def parse_detalle(self, response, item):
        item['descripcion'] = response.css('#product_description + p::text').get('')
        item['stock']       = response.css('p.instock.availability::text').re_first(r'\d+')
        yield item

Items y validación

# items.py
import scrapy

class ProductoItem(scrapy.Item):
    titulo      = scrapy.Field()
    precio      = scrapy.Field()
    rating      = scrapy.Field()
    url         = scrapy.Field()
    descripcion = scrapy.Field()
    stock       = scrapy.Field()
    scrapeado   = scrapy.Field()

Pipelines — procesar y almacenar datos

# pipelines.py
import sqlite3
from datetime import datetime
from itemadapter import ItemAdapter

class ValidacionPipeline:
    """Filtra items incompletos."""
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if not adapter.get('titulo'):
            raise scrapy.exceptions.DropItem(f"Sin título: {item}")
        adapter['scrapeado'] = datetime.now().isoformat()
        return item

class LimpiezaPipeline:
    """Limpia y normaliza campos."""
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter.get('precio'):
            adapter['precio'] = adapter['precio'].replace('£', '').strip()
        return item

class SQLitePipeline:
    """Guarda en SQLite."""
    def open_spider(self, spider):
        self.conn = sqlite3.connect('productos.db')
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS productos (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                titulo      TEXT,
                precio      TEXT,
                rating      TEXT,
                url         TEXT UNIQUE,
                descripcion TEXT,
                stock       TEXT,
                scrapeado   TEXT
            )
        """)

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

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        self.conn.execute("""
            INSERT OR REPLACE INTO productos
            (titulo, precio, rating, url, descripcion, stock, scrapeado)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (
            adapter.get('titulo'), adapter.get('precio'),
            adapter.get('rating'), adapter.get('url'),
            adapter.get('descripcion'), adapter.get('stock'),
            adapter.get('scrapeado'),
        ))
        return item

# settings.py — activar pipelines en orden
ITEM_PIPELINES = {
    'mi_scraper.pipelines.ValidacionPipeline': 100,
    'mi_scraper.pipelines.LimpiezaPipeline':   200,
    'mi_scraper.pipelines.SQLitePipeline':     300,
}

# Exportar también a JSONL automáticamente
FEEDS = {
    'productos.jsonl': {'format': 'jsonlines', 'encoding': 'utf8'},
}

Middlewares — proxies y user agents

# middlewares.py
import random

USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/122.0',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/537.36',
    'Mozilla/5.0 (X11; Linux x86_64; rv:123.0) Gecko/20100101 Firefox/123.0',
]

PROXIES = [
    'http://proxy1:3128',
    'http://proxy2:3128',
]

class RotarUserAgentMiddleware:
    def process_request(self, request, spider):
        request.headers['User-Agent'] = random.choice(USER_AGENTS)

class RotarProxyMiddleware:
    def process_request(self, request, spider):
        request.meta['proxy'] = random.choice(PROXIES)

    def process_exception(self, request, exception, spider):
        # Reintentar con otro proxy si falla
        request.meta['proxy'] = random.choice(PROXIES)
        return request

# settings.py
DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,  # desactivar el original
    'mi_scraper.middlewares.RotarUserAgentMiddleware': 400,
    'mi_scraper.middlewares.RotarProxyMiddleware':     350,
}

ROBOTSTXT_OBEY = True       # respetar robots.txt
AUTOTHROTTLE_ENABLED = True # ajuste automático de velocidad
AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0

Scrapy-Playwright — páginas con JavaScript

pip install scrapy-playwright
playwright install chromium

# settings.py
DOWNLOAD_HANDLERS = {
    'http':  'scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler',
    'https': 'scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler',
}
TWISTED_REACTOR = 'twisted.internet.asyncioreactor.AsyncioSelectorReactor'
PLAYWRIGHT_BROWSER_TYPE = 'chromium'
PLAYWRIGHT_LAUNCH_OPTIONS = {'headless': True}

# Spider con Playwright
import scrapy
from scrapy_playwright.page import PageMethod

class SPASpider(scrapy.Spider):
    name = 'spa_scraper'

    def start_requests(self):
        yield scrapy.Request(
            'https://ejemplo.com/productos-dinamicos',
            meta={
                'playwright': True,
                'playwright_include_page': True,   # acceder al objeto page
                'playwright_page_methods': [
                    # Esperar a que cargue el contenido dinámico
                    PageMethod('wait_for_selector', 'div.product-grid'),
                    # Hacer scroll para cargar lazy-loading
                    PageMethod('evaluate', 'window.scrollTo(0, document.body.scrollHeight)'),
                    PageMethod('wait_for_timeout', 2000),
                ],
            },
            callback=self.parse,
            errback=self.errback,
        )

    async def parse(self, response):
        page = response.meta.get('playwright_page')
        if page:
            await page.close()

        for producto in response.css('div.product-card'):
            yield {
                'nombre': producto.css('h2::text').get(),
                'precio': producto.css('.price::text').get(),
                'imagen': producto.css('img::attr(src)').get(),
            }

    async def errback(self, failure):
        page = failure.request.meta.get('playwright_page')
        if page:
            await page.close()
        self.logger.error(f"Error: {failure}")

Playwright standalone — scraping asíncrono

import asyncio
from playwright.async_api import async_playwright

async def scrape_producto(url: str) -> dict:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(
            user_agent='Mozilla/5.0 (compatible; MiBot/1.0)',
            viewport={'width': 1280, 'height': 720},
        )
        page = await context.new_page()

        # Bloquear recursos innecesarios (acelera la carga)
        await page.route('**/*.{png,jpg,gif,woff2,css}', lambda r: r.abort())

        await page.goto(url, wait_until='networkidle', timeout=30000)

        # Esperar elemento específico
        await page.wait_for_selector('h1.product-title', timeout=10000)

        # Extraer datos
        titulo = await page.text_content('h1.product-title')
        precio = await page.text_content('.price-tag')

        # Tomar screenshot para debug
        await page.screenshot(path='debug.png', full_page=True)

        await browser.close()
        return {'titulo': titulo.strip(), 'precio': precio.strip()}

# Scraping concurrente
async def scrape_muchos(urls: list[str]) -> list[dict]:
    semaforo = asyncio.Semaphore(3)  # máximo 3 páginas simultáneas

    async def scrape_uno(url):
        async with semaforo:
            return await scrape_producto(url)

    return await asyncio.gather(*[scrape_uno(u) for u in urls])

resultados = asyncio.run(scrape_muchos(['https://ejemplo.com/p1', 'https://ejemplo.com/p2']))

Ejecutar y monitorear Scrapy

# Ejecutar spider
scrapy crawl productos

# Exportar directamente
scrapy crawl productos -o salida.csv
scrapy crawl productos -o salida.json

# Estadísticas al final de la ejecución
scrapy crawl productos --set LOG_LEVEL=INFO

# Scrapy shell — explorar respuestas interactivamente
scrapy shell 'https://books.toscrape.com/'
# >>> response.css('title::text').get()
# >>> response.xpath('//h1/text()').get()

Buenas prácticas

• Respetar robots.txt y el Crawl-Delay del sitio — ROBOTSTXT_OBEY = True en Scrapy activa esto automáticamente.
• AUTOTHROTTLE_ENABLED = True ajusta la velocidad de scraping en función de la latencia del servidor, siendo cortés sin sacrificar velocidad.
• Scrapy para scraping a escala (miles de páginas); Playwright standalone para páginas individuales o flujos que requieren interacción (login, formularios, clicks).
• Guardar en JSONL (una línea por item) en lugar de JSON: puede leer y escribir incrementalmente sin cargar todo en memoria.
• ItemAdapter sobre el acceso directo al dict del item: compatible con Scrapy Items, dataclasses y attrs.
• Guardar screenshots de error en el errback para depurar páginas que fallan silenciosamente con Playwright.

Conversiones relacionadas

Conversiones frecuentes del catálogo:

• JPG a PNG
• MP4 a MP3
• PDF a DOCX
• PNG a JPG
• MP3 a WAV
• DOCX a PDF