IVYXSTUDIO · IVY NODE

Web Scraping

ivy.node.web-scraping · v0.1.0

ivyx✓

Fetches a page, and the pages after it, and returns the rows a content function reads from each. The content and next-page functions are handles another node or a fixture provides.

#BeautifulSoup#HTML parsing

Inputs

Field	Type	Description
urlrequired	string	The first page to fetch.
content_processorrequired	Not declared	A function (soup) -> list of row objects for one page.
next_page_rulerequired	Not declared	A function (soup, url) -> the next page's url, or None to stop.

Outputs

Field	Type	Description
scraped_dictionaryrequired	array	The rows every page produced, in page order.

Source

python

# Input preparation
inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]
url = inp["url"]
next_page_rule = inp["next_page_rule"]
content_processor = inp["content_processor"]

# Compute
import requests
from bs4 import BeautifulSoup
from typing import Callable, Optional, List, Dict

class WebScraping:
    """
    A flexible web scraping class where all content and pagination logic
    is injected from outside.
    """

    def __init__(
        self, 
        url: str, 
        next_page_rule: Callable[[BeautifulSoup, str], Optional[str]],
        content_processor: Callable[[BeautifulSoup], List[Dict]]
    ):
        """
        Initialize the Scraper with external rules for pagination and content processing.

        :param url: The starting URL for scraping.
        :param next_page_rule: A callable to determine the next page URL.
        :param content_processor: A callable to process the content of each page.
        """
        self.url = url
        self.next_page_rule = next_page_rule
        self.content_processor = content_processor

    def scrape(self) -> List[Dict]:
        """
        Main scraping method that handles pagination and content processing.

        :return: A list of dictionaries containing the scraped data.
        """
        url = self.url
        all_data = []

        while url:
            print(f"Scraping URL: {url}")
            response = requests.get(url, timeout=30)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, 'html.parser')

            # Process the current page's content
            page_data = self.content_processor(soup)
            all_data.extend(page_data)

            # Get the next page URL using the injected rule
            url = self.next_page_rule(soup, url)

        return all_data

web_scraping = WebScraping(
    url=url,
    next_page_rule=next_page_rule,
    content_processor=content_processor
)

scraped_dictionary = web_scraping.scrape()

# Output collection (runner reads __ivy_ctx__)
out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out["scraped_dictionary"] = scraped_dictionary

Tests

Requires: python:3.9, requests, beautifulsoup4

one-page
A page with two headings yields two rows and no next page.
not-found
A page that answers 404 fails the step.