Python Web Crawler: From a Script to Scrapy

# pip install requests beautifulsoup4
import json
import time
from collections import deque
from urllib.parse import urljoin, urldefrag, urlparse
from urllib.robotparser import RobotFileParser

import requests
from bs4 import BeautifulSoup

START = "https://books.toscrape.com/"
MAX_PAGES = 100
DELAY = 1.0                                   # seconds between requests
UA = "Mozilla/5.0 (compatible; demo-crawler/1.0)"

host = urlparse(START).netloc
robots = RobotFileParser(urljoin(START, "/robots.txt"))
robots.read()
session = requests.Session()
session.headers["User-Agent"] = UA

queue, seen, pages = deque([START]), {START}, []
while queue and len(pages) < MAX_PAGES:
    url = queue.popleft()
    if not robots.can_fetch(UA, url):
        continue
    try:
        resp = session.get(url, timeout=20)
    except requests.RequestException as exc:
        print("error", url, exc)
        continue
    if resp.status_code != 200 or "text/html" not in resp.headers.get("Content-Type", ""):
        continue

    soup = BeautifulSoup(resp.text, "html.parser")
    pages.append({"url": url, "title": soup.title.get_text(strip=True) if soup.title else ""})

    for a in soup.select("a[href]"):
        link = urldefrag(urljoin(url, a["href"]))[0]     # absolute, without #fragment
        if urlparse(link).netloc == host and link not in seen:
            seen.add(link)
            queue.append(link)
    time.sleep(DELAY)

json.dump(pages, open("pages.json", "w"), indent=2)
print(len(pages), "pages crawled")

next_link = soup.select_one("li.next a")
if next_link:
    queue.append(urljoin(url, next_link["href"]))          # the next list page
for a in soup.select("article.product_pod h3 a"):
    queue.append(urljoin(url, a["href"]))                   # each item on this page

# pip install httpx beautifulsoup4
import asyncio
from urllib.parse import urljoin, urldefrag, urlparse
import httpx
from bs4 import BeautifulSoup

START, MAX_PAGES, CONCURRENCY = "https://books.toscrape.com/", 300, 5
host = urlparse(START).netloc

async def crawl():
    seen, queue, done = {START}, asyncio.Queue(), []
    await queue.put(START)
    async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:
        async def worker():
            while True:
                url = await queue.get()
                try:
                    if len(done) >= MAX_PAGES:
                        continue                         # drain the queue once the limit is hit
                    r = await client.get(url)
                    if "text/html" in r.headers.get("content-type", ""):
                        soup = BeautifulSoup(r.text, "html.parser")
                        done.append(url)
                        for a in soup.select("a[href]"):
                            link = urldefrag(urljoin(url, a["href"]))[0]
                            if urlparse(link).netloc == host and link not in seen:
                                seen.add(link)
                                queue.put_nowait(link)
                    await asyncio.sleep(0.5)            # per-worker pause
                except httpx.HTTPError:
                    pass
                finally:
                    queue.task_done()
        workers = [asyncio.create_task(worker()) for _ in range(CONCURRENCY)]
        await queue.join()
        for w in workers:
            w.cancel()
    print(len(done), "pages")

asyncio.run(crawl())

# pip install scrapy  ->  scrapy runspider books_crawler.py -O books.json
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule

class BooksCrawler(CrawlSpider):
    name = "books"
    allowed_domains = ["books.toscrape.com"]
    start_urls = ["https://books.toscrape.com/"]
    custom_settings = {"ROBOTSTXT_OBEY": True, "DOWNLOAD_DELAY": 1, "CLOSESPIDER_PAGECOUNT": 500}
    rules = (
        Rule(LinkExtractor(allow=r"catalogue/page-\d+\.html")),                        # follow list pages
        Rule(LinkExtractor(restrict_css="article.product_pod h3"), callback="parse_book"),  # open each book
    )

    def parse_book(self, response):
        yield {
            "title": response.css("h1::text").get(),
            "price": response.css("p.price_color::text").get(),
            "url": response.url,
        }

# pip install 'crawlee[beautifulsoup]'
import asyncio
from crawlee.crawlers import BeautifulSoupCrawler, BeautifulSoupCrawlingContext

async def main():
    crawler = BeautifulSoupCrawler(max_requests_per_crawl=50)

    @crawler.router.default_handler
    async def handle(context: BeautifulSoupCrawlingContext) -> None:
        await context.push_data({"url": context.request.url, "title": context.soup.title.string})
        await context.enqueue_links()                 # same-site links by default

    await crawler.run(["https://books.toscrape.com/"])

asyncio.run(main())