harambe-sdk 0.18.31

Data extraction SDK for Playwright 🐒🍌

🦍 Harambe Web extraction SDK 🦍

Harambe

Harambe is the extraction SDK for Reworkd. It provides a simple interface for interacting with the web. It provides a unified interface and runtime for both manual and automatically created web extractors

---

• Setup and Installation
• Folder Structure
• Example Scraper
  • Detail Only Scraper
  • Listing Scraper
  • Using Cache
• Running a Scraper
• Submitting a PR

---

Setup and Installation

To install Harambe, clone the repository and install the requirements. All requirements are managed via poetry.

git clone https://github.com/reworkd/harambe.git
poetry install

Folder Structure

The scrapers folder contains all the scrapers. The harambe folder contains the SDK and utility functions.

Example Scraper

Generally scrapers come in two types, listing and detail scrapers. Listing scrapers are used to collect a list of items to scrape. Detail scrapers are used to scrape the details of a single item.

If all the items that you want to scrape are available on a single page, then you can use a detail scraper to scrape all the items. If the items are spread across multiple pages, then you will need to use a listing scraper to collect the items and then use a detail scraper to scrape the details of each item.

ALL scrapers must be decorated with SDK.scraper. This decorator registers the scraper with the SDK and provides the SDK with the necessary information to run the scraper.

Detail Only Scraper

Shown below is an example detail scraper. The context parameter is used to pass data from the listing scraper to the detail scraper. In this example, the context parameter is used to pass the phone

import asyncio
from typing import Any

from playwright.async_api import Page

from harambe import SDK
from harambe import PlaywrightUtils as Pu

SELECTORS = {
    "last_page": "",
    "list_view": "//div[@class='et_pb_blurb_content']",
    "name": "//h4/*[self::span or self::a]",
    "fax": ">Fax.*?strong>(.*?)<br>",
    # etc...
}


# Annotation registers the scraper with the SDK
@SDK.scraper(domain="https://apprhs.org/our-locations/", stage="detail")
async def scrape(sdk: SDK, url: str, *args: Any, **kwargs: Any) -> None:
    page: Page = sdk.page

    locations = await page.locator(SELECTORS["list_view"]).all()
    for location in locations:        
        # Save the data to the database or file
        await sdk.save_data(
            {
                "name": await Pu.get_text(location, SELECTORS["name"]),
                "fax": await Pu.parse_by_regex(location, SELECTORS["fax"]),
                # etc...
            }
        )


if __name__ == "__main__":
    asyncio.run(SDK.run(scrape, "https://apprhs.org/our-locations/"))

Listing Scraper

Shown below is an example listing scraper. Use SDK.enqueue to to add urls that will need be scraped by the detail scraper. The context parameter is used to pass data from the listing scraper to the detail scraper.

import asyncio
from typing import Any

from playwright.async_api import Page

from harambe import SDK

SELECTORS = {}


@SDK.scraper(domain="https://example.org", stage="listing")
async def scrape(sdk: SDK, url: str, *args: Any, **kwargs: Any) -> None:
    page: Page = sdk.page

    for url in [
        "https://example.org/1",
        "https://example.org/2",
        "https://example.org/3",
    ]:  # Imagine these are locators
        await sdk.enqueue(
            url,
            context={
                "phone": "123-456-7890",
                # Some data from the listing page that we want to pass to the detail page, (optional)
                "foo": "bar",
                "baz": "qux",
            },
        )


@SDK.scraper(domain="https://example.org", stage="detail")
async def scrape_detail(sdk: SDK, url: str, context: Any) -> None:
    page: Page = sdk.page

    # Grab all properties from the context
    detail = {**context}

    detail["fax"] = "123-456-7890"  # Some data grabbed from the detail page
    detail["type"] = "Hospital"  # Some data grabbed from the detail page
    await sdk.save_data(detail)  # Save the data to the database


if __name__ == "__main__":
    asyncio.run(SDK.run(scrape, "https://navicenthealth.org/locations"))
    asyncio.run(SDK.run_from_file(scrape_detail))

Using Cache

The code below is an example detail scraper that relies on HAR cache that it creates during initial run, subsequently using it as source of data to improve speed and consume less bandwidth.

import asyncio
import os.path
from typing import Any

from playwright.async_api import Page

from harambe import SDK
from harambe import PlaywrightUtils as Pu

HAR_FILE_PATH = "bananas.har"
SELECTORS = {
    "last_page": "",
    "list_view": "//div[@class='et_pb_blurb_content']",
    "name": "//h4/*[self::span or self::a]",
    "fax": ">Fax.*?strong>(.*?)<br>",
    # etc...
}


async def setup(sdk: SDK) -> None:
    page: Page = sdk.page

    already_cached = os.path.isfile(HAR_FILE_PATH)

    if already_cached:
        await page.route_from_har(HAR_FILE_PATH, not_found="fallback")
    else:
        await page.route_from_har(HAR_FILE_PATH, not_found="fallback", update=True)


# Annotation registers the scraper with the SDK
@SDK.scraper(domain="https://apprhs.org/our-locations/", stage="detail")
async def scrape(sdk: SDK, url: str, *args: Any, **kwargs: Any) -> None:
    page: Page = sdk.page

    locations = await page.locator(SELECTORS["list_view"]).all()
    for location in locations:
        # Save the data to the database or file
        await sdk.save_data(
            {
                "name": await Pu.get_text(location, SELECTORS["name"]),
                "fax": await Pu.parse_by_regex(location, SELECTORS["fax"]),
                # etc...
            }
        )


if __name__ == "__main__":
    asyncio.run(SDK.run(scrape, "https://apprhs.org/our-locations/", setup=setup))

Running a Scraper

You can use poetry to run a scraper. The run command takes the scraper function and the url to scrape. The run_from_file command takes the scraper function and the path to the file containing the urls to scrape.

poetry run python poetry run python scrapers/medical/apprhs.py 

Submitting a PR

Before submitting a PR, please run the following commands to ensure that your code is formatted correctly.

make FORMAT LINT

Happy extraction! 🦍