Advanced Web Scraping with Scrapy and Playwright in Python

Advanced Web Scraping with Scrapy and Playwright in Python

For serious scraping at scale you need more than requests + BeautifulSoup. Scrapy is Python's most complete scraping framework; Playwright handles JavaScript-rendered modern pages.

Scrapy — installation and project structure

pip install scrapy
scrapy startproject my_scraper
cd my_scraper
scrapy genspider products store.com

my_scraper/
├── scrapy.cfg
└── my_scraper/
    ├── settings.py
    ├── items.py
    ├── middlewares.py
    ├── pipelines.py
    └── spiders/
        └── products.py

Basic spider

# spiders/products.py
import scrapy
from my_scraper.items import ProductItem

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

    custom_settings = {
        'DOWNLOAD_DELAY': 1.5,
        'RANDOMIZE_DOWNLOAD_DELAY': True,
        'CONCURRENT_REQUESTS_PER_DOMAIN': 2,
    }

    def parse(self, response):
        for book in response.css('article.product_pod'):
            item = ProductItem()
            item['title']  = book.css('h3 a::attr(title)').get()
            item['price']  = book.css('p.price_color::text').get()
            item['rating'] = book.css('p.star-rating::attr(class)').get().split()[-1]
            item['url']    = response.urljoin(book.css('h3 a::attr(href)').get())
            yield response.follow(item['url'], callback=self.parse_detail,
                                  cb_kwargs={'item': item})

        # Automatic pagination
        next_page = response.css('li.next a::attr(href)').get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

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

Items

# items.py
import scrapy

class ProductItem(scrapy.Item):
    title       = scrapy.Field()
    price       = scrapy.Field()
    rating      = scrapy.Field()
    url         = scrapy.Field()
    description = scrapy.Field()
    stock       = scrapy.Field()
    scraped_at  = scrapy.Field()

Pipelines — process and store data

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

class ValidationPipeline:
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if not adapter.get('title'):
            raise scrapy.exceptions.DropItem(f"Missing title: {item}")
        adapter['scraped_at'] = datetime.now().isoformat()
        return item

class CleaningPipeline:
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter.get('price'):
            adapter['price'] = adapter['price'].replace('£', '').strip()
        return item

class SQLitePipeline:
    def open_spider(self, spider):
        self.conn = sqlite3.connect('products.db')
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS products (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                title       TEXT,
                price       TEXT,
                rating      TEXT,
                url         TEXT UNIQUE,
                description TEXT,
                stock       TEXT,
                scraped_at  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 products
            (title, price, rating, url, description, stock, scraped_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (
            adapter.get('title'),   adapter.get('price'),
            adapter.get('rating'),  adapter.get('url'),
            adapter.get('description'), adapter.get('stock'),
            adapter.get('scraped_at'),
        ))
        return item

# settings.py — enable pipelines in order
ITEM_PIPELINES = {
    'my_scraper.pipelines.ValidationPipeline': 100,
    'my_scraper.pipelines.CleaningPipeline':   200,
    'my_scraper.pipelines.SQLitePipeline':     300,
}
FEEDS = {
    'products.jsonl': {'format': 'jsonlines', 'encoding': 'utf8'},
}

Middlewares — user agent and proxy rotation

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 RotateUserAgentMiddleware:
    def process_request(self, request, spider):
        request.headers['User-Agent'] = random.choice(USER_AGENTS)

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

    def process_exception(self, request, exception, spider):
        request.meta['proxy'] = random.choice(PROXIES)
        return request

# settings.py
DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,
    'my_scraper.middlewares.RotateUserAgentMiddleware': 400,
    'my_scraper.middlewares.RotateProxyMiddleware':     350,
}
ROBOTSTXT_OBEY = True
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0

scrapy-playwright — JavaScript pages

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}

import scrapy
from scrapy_playwright.page import PageMethod

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

    def start_requests(self):
        yield scrapy.Request(
            'https://example.com/dynamic-products',
            meta={
                'playwright': True,
                'playwright_include_page': True,
                'playwright_page_methods': [
                    PageMethod('wait_for_selector', 'div.product-grid'),
                    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 product in response.css('div.product-card'):
            yield {
                'name':  product.css('h2::text').get(),
                'price': product.css('.price::text').get(),
                'image': product.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 — async scraping

import asyncio
from playwright.async_api import async_playwright

async def scrape_product(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; MyBot/1.0)',
            viewport={'width': 1280, 'height': 720},
        )
        page = await context.new_page()

        # Block unnecessary resources (speeds up loading)
        await page.route('**/*.{png,jpg,gif,woff2,css}', lambda r: r.abort())

        await page.goto(url, wait_until='networkidle', timeout=30_000)
        await page.wait_for_selector('h1.product-title', timeout=10_000)

        title = await page.text_content('h1.product-title')
        price = await page.text_content('.price-tag')

        await page.screenshot(path='debug.png', full_page=True)
        await browser.close()
        return {'title': title.strip(), 'price': price.strip()}

# Concurrent scraping
async def scrape_many(urls: list[str]) -> list[dict]:
    sem = asyncio.Semaphore(3)  # max 3 simultaneous pages

    async def scrape_one(url):
        async with sem:
            return await scrape_product(url)

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

results = asyncio.run(scrape_many(['https://example.com/p1', 'https://example.com/p2']))

Run and monitor Scrapy

# Run spider
scrapy crawl products

# Export directly
scrapy crawl products -o output.csv
scrapy crawl products -o output.json

# Set log level
scrapy crawl products --set LOG_LEVEL=INFO

# Interactive shell — explore responses
scrapy shell 'https://books.toscrape.com/'
# >>> response.css('title::text').get()
# >>> response.xpath('//h1/text()').get()

Best practices

• Respect robots.txt and Crawl-Delay — ROBOTSTXT_OBEY = True in Scrapy handles this automatically.
• AUTOTHROTTLE_ENABLED = True adjusts crawl speed based on server latency — polite without sacrificing throughput.
• Use Scrapy for scale (thousands of pages); use Playwright standalone for single pages or flows requiring interaction (login, forms, clicks).
• Store as JSONL (one item per line) instead of JSON — reads and writes incrementally without loading everything into memory.
• ItemAdapter over direct dict access — compatible with Scrapy Items, dataclasses, and attrs objects.
• Screenshot on error in errback to debug pages that fail silently with Playwright.

Related conversions

Frequent conversions across the catalogue:

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