✨ Block ads (#74)

* ✨ Block ads

* Block ads in Pyppeteer

* Block ads in Selenium

- Use Selenium-Wire
- Expose WebDriver in Setup and Navigate handlers

* Update grouping description

* Bump version

* Tests

Author: roniemartinez

.github/workflows/python.yml

@@ -10,6 +10,7 @@ on:
   pull_request:
     branches:
       - master
+  workflow_dispatch:
 
 concurrency:
   group: ${{ github.ref }}

README.md

@@ -69,7 +69,7 @@ dude scrape --url "<url>" --output data.json path/to/script.py
 
 - Simple [Flask](https://github.com/pallets/flask)-inspired design - build a scraper with decorators.
 - Uses [Playwright](https://playwright.dev/python/) API - run your scraper in Chrome, Firefox and Webkit and leverage Playwright's powerful selector engine supporting CSS, XPath, text, regex, etc.
-- Data grouping - group related scraping data.
+- Data grouping - group related results.
 - URL pattern matching - run functions on specific URLs.
 - Priority - reorder functions based on priority.
 - Setup function - enable setup steps (clicking dialogs or login).

docs/advanced/01_setup.md

@@ -7,7 +7,7 @@ Setup handlers could perform any of the following:
 2. Click on dialogs buttons
 
 To create a Setup handler, you can pass `setup=True` parameter to `@select()` decorator. 
-The only difference with Setup and normal element handler is that setup functions should accept 2 parameters, the element matched by the selector and the Page object.
+The only difference with Setup and normal element handler is that setup functions should accept 2 parameters, the element matched by the selector and the Page object (or WebDriver object in Selenium).
 Click on the annotations (+ sign) for more details.
 
 === "Python"

docs/advanced/02_navigate.md

@@ -3,7 +3,7 @@
 Navigate handlers are used to move from page to page.
 
 To create a Navigate handler, you can pass `navigate=True` parameter to `@select()` decorator. 
-Like Setup handlers, Navigate handlers should accept 2 parameters, the element matched by the selector and the Page object.
+Like Setup handlers, Navigate handlers should accept 2 parameters, the element matched by the selector and the Page object (or WebDriver object in Selenium).
 Click on the annotations (+ sign) for more details.
 
 === "Python"

docs/features.md

@@ -2,7 +2,7 @@
 
 - Simple [Flask](https://github.com/pallets/flask)-inspired design - build a scraper with decorators.
 - Uses [Playwright](https://playwright.dev/python/) API - run your scraper in Chrome, Firefox and Webkit and leverage Playwright's powerful selector engine supporting CSS, XPath, text, regex, etc.
-- Data grouping - group related scraping data.
+- Data grouping - group related results.
 - URL pattern matching - run functions on specific URLs.
 - Priority - reorder functions based on priority.
 - Setup function - enable setup steps (clicking dialogs or login).

dude/optional/pyppeteer_scraper.py

@@ -3,8 +3,10 @@
 import logging
 from typing import Any, AsyncIterable, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union
 
+from braveblock import Adblocker
 from pyppeteer import launch
 from pyppeteer.element_handle import ElementHandle
+from pyppeteer.network_manager import Request
 from pyppeteer.page import Page
 
 from ..base import ScraperAbstract
@@ -19,6 +21,10 @@ class PyppeteerScraper(ScraperAbstract):
     Pyppeteer-based scraper
     """
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super(PyppeteerScraper, self).__init__(*args, **kwargs)
+        self.adblock = Adblocker()
+
     def run(
         self,
         urls: Sequence[str],
@@ -97,6 +103,21 @@ async def navigate_async(self, page: Page = None) -> bool:
                 return True
         return False
 
+    async def _block_url_if_needed(self, request: Request) -> Any:
+        url = request.url
+        source_url = (
+            request.headers.get("referer") or request.headers.get("origin") or request.headers.get("host") or url
+        )
+        if self.adblock.check_network_urls(
+            url=url,
+            source_url=source_url,
+            request_type=request.resourceType,
+        ):
+            logger.info("URL %s has been blocked.", url)
+            return await request.abort()
+        else:
+            return await request.continue_()
+
     async def _run_async(
         self,
         urls: Sequence[str],
@@ -106,7 +127,7 @@ async def _run_async(
         output: Optional[str],
         format: str,
     ) -> None:
-        launch_args: Dict[str, Any] = {"headless": headless, "args": ["--no-sandbox", "--disable-notifications"]}
+        launch_args: Dict[str, Any] = {"headless": headless, "args": ["--disable-notifications"]}
         if proxy:
             launch_args["args"] = [f"--proxy-server={proxy['server']}"]
 
@@ -116,6 +137,9 @@ async def _run_async(
         if proxy and proxy["username"] and proxy["password"]:
             await page.authenticate(credentials={"username": proxy["username"], "password": proxy["password"]})
 
+        await page.setRequestInterception(True)
+        page.on("request", lambda res: asyncio.ensure_future(self._block_url_if_needed(res)))
+
         for url in urls:
             await page.goto(url)
             logger.info("Loaded page %s", page.url)

dude/optional/selenium_scraper.py

@@ -4,14 +4,16 @@
 import os
 from typing import Any, AsyncIterable, Callable, Iterable, Optional, Sequence, Tuple, Union
 
-from selenium import webdriver
+from braveblock import Adblocker
 from selenium.webdriver.chrome.options import Options as ChromeOptions
 from selenium.webdriver.chrome.service import Service as ChromeService
 from selenium.webdriver.common.by import By
 from selenium.webdriver.firefox.options import Options as FirefoxOptions
 from selenium.webdriver.firefox.service import Service as FirefoxService
 from selenium.webdriver.remote.webdriver import WebDriver
 from selenium.webdriver.remote.webelement import WebElement
+from seleniumwire.request import Request
+from seleniumwire.webdriver import Chrome, Firefox
 from webdriver_manager.chrome import ChromeDriverManager
 from webdriver_manager.firefox import GeckoDriverManager
 from webdriver_manager.utils import ChromeType
@@ -27,6 +29,10 @@ class SeleniumScraper(ScraperAbstract):
     Selenium-based scraper
     """
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super(SeleniumScraper, self).__init__(*args, **kwargs)
+        self.adblock = Adblocker()
+
     def run(
         self,
         urls: Sequence[str],
@@ -100,7 +106,7 @@ def setup(self, driver: WebDriver = None) -> None:
         assert driver is not None
         for rule in self.get_setup_rules(driver.current_url):
             for element in self._get_elements(driver, rule.selector):
-                rule.handler(element)
+                rule.handler(element, driver)
 
     async def setup_async(self, driver: WebDriver = None) -> None:
         """
@@ -112,9 +118,9 @@ async def setup_async(self, driver: WebDriver = None) -> None:
         for rule in self.get_setup_rules(driver.current_url):
             for element in self._get_elements(driver, rule.selector):
                 if asyncio.iscoroutinefunction(rule.handler):
-                    await rule.handler(element)
+                    await rule.handler(element, driver)
                 else:
-                    rule.handler(element)
+                    rule.handler(element, driver)
 
     def navigate(self, driver: WebDriver = None) -> bool:
         """
@@ -125,7 +131,7 @@ def navigate(self, driver: WebDriver = None) -> bool:
         assert driver is not None
         for rule in self.get_navigate_rules(driver.current_url):
             for element in self._get_elements(driver, rule.selector):
-                rule.handler(element)
+                rule.handler(element, driver)
                 logger.info("Navigated to %s", driver.current_url)
                 return True
         return False
@@ -140,9 +146,9 @@ async def navigate_async(self, driver: WebDriver = None) -> bool:
         for rule in self.get_navigate_rules(driver.current_url):
             for element in self._get_elements(driver, rule.selector):
                 if asyncio.iscoroutinefunction(rule.handler):
-                    await rule.handler(element)
+                    await rule.handler(element, driver)
                 else:
-                    rule.handler(element)
+                    rule.handler(element, driver)
                 logger.info("Navigated to %s", driver.current_url)
                 return True
         return False
@@ -203,23 +209,40 @@ async def _run_async(
         driver.close()
         await self._save_async(format, output)
 
-    @staticmethod
-    def _get_driver(browser_type: str, headless: bool) -> WebDriver:
+    def _block_url_if_needed(self, request: Request) -> None:
+        url = request.url
+        source_url = (
+            request.headers.get("referer") or request.headers.get("origin") or request.headers.get("host") or url
+        )
+        if self.adblock.check_network_urls(
+            url=url,
+            source_url=source_url,
+            request_type=request.headers.get("sec-fetch-dest") or "other",
+        ):
+            logger.info("URL %s has been blocked.", url)
+            request.abort()
+
+    def _get_driver(self, browser_type: str, headless: bool) -> WebDriver:
         # TODO: Add more drivers: https://github.com/SergeyPirogov/webdriver_manager#webdriver-manager-for-python
         if browser_type == "firefox":
             executable_path = GeckoDriverManager().install()
             firefox_options = FirefoxOptions()
             firefox_options.headless = headless
             firefox_options.set_preference("dom.webnotifications.enabled", False)
-            return webdriver.Firefox(service=FirefoxService(executable_path=executable_path), options=firefox_options)
-
-        chrome_options = ChromeOptions()
-        chrome_options.headless = headless
-        chrome_options.add_argument("disable-notifications")
-        executable_path = ChromeDriverManager(
-            chrome_type=ChromeType.CHROMIUM, version=os.getenv("CHROMEDRIVER_VERSION", "latest")
-        ).install()
-        return webdriver.Chrome(service=ChromeService(executable_path=executable_path), options=chrome_options)
+            driver = Firefox(service=FirefoxService(executable_path=executable_path), options=firefox_options)
+        else:
+            chrome_options = ChromeOptions()
+            chrome_options.headless = headless
+            chrome_options.add_argument("disable-notifications")
+            executable_path = ChromeDriverManager(
+                chrome_type=ChromeType.CHROMIUM, version=os.getenv("CHROMEDRIVER_VERSION", "latest")
+            ).install()
+            driver = Chrome(service=ChromeService(executable_path=executable_path), options=chrome_options)
+
+        driver.implicitly_wait(10)
+        driver.request_interceptor = self._block_url_if_needed
+
+        return driver
 
     def collect_elements(self, driver: WebDriver = None) -> Iterable[Tuple[str, int, int, int, Any, Callable]]:
         """

dude/playwright_scraper.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Any, AsyncIterable, Callable, Dict, Iterable, Optional, Sequence, Tuple, Union
 
+from braveblock import Adblocker
 from playwright import async_api, sync_api
 from playwright.async_api import async_playwright
 from playwright.sync_api import sync_playwright
@@ -18,6 +19,10 @@ class PlaywrightScraper(ScraperAbstract):
     Playwright-based scraper
     """
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super(PlaywrightScraper, self).__init__(*args, **kwargs)
+        self.adblock = Adblocker()
+
     def run(
         self,
         urls: Sequence[str],
@@ -148,6 +153,23 @@ def _get_launch_kwargs(browser_type: str) -> Dict[str, Any]:
             args.append("--disable-notifications")
         return {"args": args, "firefox_user_prefs": {"dom.webnotifications.enabled": False}}
 
+    def _block_url_if_needed(self, route: Union[sync_api.Route, async_api.Route]) -> Any:
+        url = route.request.url
+        source_url = (
+            route.request.headers.get("referer")
+            or route.request.headers.get("origin")
+            or route.request.headers.get("host")
+            or url
+        )
+        if self.adblock.check_network_urls(
+            url=url,
+            source_url=source_url,
+            request_type=route.request.resource_type,
+        ):
+            logger.info("URL %s has been blocked.", url)
+            return route.abort()
+        return route.continue_()
+
     def _run_sync(
         self,
         urls: Sequence[str],
@@ -163,6 +185,7 @@ def _run_sync(
         with sync_playwright() as p:
             browser = p[browser_type].launch(headless=headless, proxy=proxy, **launch_kwargs)
             page = browser.new_page()
+            page.route("**/*", self._block_url_if_needed)
             self._scrape_sync(page, urls, pages)
             browser.close()
         self._save(format, output)
@@ -193,6 +216,7 @@ async def _run_async(
         async with async_playwright() as p:
             browser = await p[browser_type].launch(headless=headless, proxy=proxy, **launch_kwargs)
             page = await browser.new_page()
+            await page.route("**/*", self._block_url_if_needed)
             for url in urls:
                 await page.goto(url)
                 logger.info("Loaded page %s", page.url)

examples/dude.html

@@ -3,6 +3,7 @@
 <head>
     <meta charset="UTF-8">
     <title>A simple sandbox for dude</title>
+    <link rel="stylesheet" href="https://dude.ron.sh/blockme.css">
 </head>
 <body>
 <div class="custom-group">