refactor(scraper/web): add open_new_page option

Author: idiotWu

npiai/tools/scrapers/page_analyzer/app.py

@@ -36,11 +36,26 @@ class PageAnalyzer(BrowserTool):
         """
     )
 
-    force_captcha_detection: bool
+    _force_captcha_detection: bool
+    _open_new_page: bool
 
-    def __init__(self, force_captcha_detection: bool = False, **kwargs):
+    def __init__(
+        self,
+        force_captcha_detection: bool = False,
+        open_new_page=True,
+        **kwargs,
+    ):
+        """
+        Initialize the PageAnalyzer tool
+
+        Args:
+            force_captcha_detection: Whether to force the captcha detection when loading the page.
+            open_new_page: Whether to open a new page when analyzing the page. If set to False, the current page will be used.
+            **kwargs: BrowserTool arguments
+        """
         super().__init__(**kwargs)
-        self.force_captcha_detection = force_captcha_detection
+        self._force_captcha_detection = force_captcha_detection
+        self._open_new_page = open_new_page
 
     async def _validate_pagination(
         self,
@@ -268,14 +283,15 @@ async def support_infinite_scroll(
             url: URL of the page
             items_selector: CSS selector of the items on the page
         """
-        # use long wait time for pages to be fully loaded
-        await self.load_page(
-            ctx=ctx,
-            url=url,
-            timeout=3000,
-            wait_for_selector=items_selector,
-            force_capcha_detection=self.force_captcha_detection,
-        )
+        if self._open_new_page:
+            # use long wait time for pages to be fully loaded
+            await self.load_page(
+                ctx=ctx,
+                url=url,
+                timeout=3000,
+                wait_for_selector=items_selector,
+                force_capcha_detection=self._force_captcha_detection,
+            )
 
         return await self.playwright.page.evaluate(
             """
@@ -363,9 +379,12 @@ async def get_pagination_button(
             url: URL of the page
             items_selector: CSS selector of the items on the page
         """
-        await self.load_page(
-            ctx, url, force_capcha_detection=self.force_captcha_detection
-        )
+        if self._open_new_page:
+            await self.load_page(
+                ctx,
+                url,
+                force_capcha_detection=self._force_captcha_detection,
+            )
 
         # use latest page url in case of redirections
         page_url = await self.get_page_url()
@@ -480,9 +499,12 @@ async def infer_scraping_type(self, ctx: Context, url: str) -> ScrapingType:
             ctx: NPi Context
             url: URL of the page
         """
-        await self.load_page(
-            ctx, url, force_capcha_detection=self.force_captcha_detection
-        )
+        if self._open_new_page:
+            await self.load_page(
+                ctx,
+                url,
+                force_capcha_detection=self._force_captcha_detection,
+            )
 
         page_url = await self.get_page_url()
         page_title = await self.get_page_title()
@@ -564,12 +586,13 @@ async def infer_similar_items_selector(
             ctx: NPi Context
             url: URL of the page
         """
-        await self.load_page(
-            ctx,
-            url,
-            timeout=3000,
-            force_capcha_detection=self.force_captcha_detection,
-        )
+        if self._open_new_page:
+            await self.load_page(
+                ctx,
+                url,
+                timeout=3000,
+                force_capcha_detection=self._force_captcha_detection,
+            )
 
         # use latest page url in case of redirections
         page_url = await self.get_page_url()

npiai/tools/scrapers/web/app.py

@@ -36,9 +36,6 @@ class WebScraper(BaseScraper, BrowserTool):
     pagination_button_selector: str | None
     skip_item_hashes: Set[str] | None
 
-    # The maximum number of items to summarize in a single batch
-    _batch_size: int
-
     # all items loaded flag
     _all_items_loaded: bool = False
 
@@ -49,6 +46,9 @@ class WebScraper(BaseScraper, BrowserTool):
     # The list of hashes of items that have been skipped
     _matched_hashes: List[str]
 
+    # Whether to open a new page when start scraping
+    _open_new_page: bool
+
     def __init__(
         self,
         url: str,
@@ -58,6 +58,7 @@ def __init__(
         pagination_button_selector: str | None = None,
         skip_item_hashes: List[str] | None = None,
         headless: bool = True,
+        open_new_page: bool = True,
         playwright: PlaywrightContext = None,
     ):
         BaseScraper.__init__(self)
@@ -68,6 +69,7 @@ def __init__(
         self.items_selector = items_selector
         self.pagination_button_selector = pagination_button_selector
         self.skip_item_hashes = set(skip_item_hashes) if skip_item_hashes else None
+        self._open_new_page = open_new_page
         self._matched_hashes = []
         self._webpage_access_lock = asyncio.Lock()
 
@@ -76,13 +78,15 @@ def get_matched_hashes(self) -> List[str]:
 
     async def init_data(self, ctx: Context):
         self._matched_hashes = []
-        await self.load_page(
-            ctx=ctx,
-            url=self.url,
-            timeout=3000,
-            wait_for_selector=self.items_selector,
-            force_capcha_detection=True,
-        )
+
+        if self._open_new_page:
+            await self.load_page(
+                ctx=ctx,
+                url=self.url,
+                timeout=3000,
+                wait_for_selector=self.items_selector,
+                force_capcha_detection=True,
+            )
 
     async def next_items(
         self,

npiai/tools/scrapers/web/presets/linkedin/posts_scraper.py

@@ -34,6 +34,7 @@ def __init__(
         url: str,
         skip_item_hashes: List[str] | None = None,
         headless: bool = True,
+        open_new_page: bool = True,
         playwright: PlaywrightContext = None,
     ):
         super().__init__(
@@ -42,6 +43,7 @@ def __init__(
             items_selector=".fie-impression-container",
             skip_item_hashes=skip_item_hashes,
             headless=headless,
+            open_new_page=open_new_page,
             playwright=playwright,
         )