:gem: [Feature] SearchAPIApp: Concurrent fetch urls and extract contents

apis/search_api.py

@@ -10,9 +10,9 @@ from typing import Union
 from sse_starlette.sse import EventSourceResponse, ServerSentEvent
 from utils.logger import logger
 from networks.google_searcher import GoogleSearcher
-from networks.webpage_fetcher import WebpageFetcher
+from networks.webpage_fetcher import BatchWebpageFetcher
 from documents.query_results_extractor import QueryResultsExtractor
-from documents.webpage_content_extractor import WebpageContentExtractor
+from documents.webpage_content_extractor import BatchWebpageContentExtractor
 from utils.logger import logger
 
 
@@ -43,7 +43,7 @@ class SearchAPIApp:
             default=["web"],
             description="(list[str]) Types of search results: `web`, `image`, `videos`, `news`",
         )
-        extract_content: bool = Field(
+        extract_webpage: bool = Field(
             default=False,
             description="(bool) Enable extracting main text contents from webpage, will add `text` filed in each `query_result` dict",
         )
@@ -73,24 +73,57 @@ class SearchAPIApp:
             queries_search_results.append(query_search_results)
         logger.note(queries_search_results)
 
-        if item.extract_content:
-            webpage_fetcher = WebpageFetcher()
-            webpage_content_extractor = WebpageContentExtractor()
-            for query_idx, query_search_result in enumerate(queries_search_results):
-                for query_result_idx, query_result in enumerate(
-                    query_search_result["query_results"]
-                ):
-                    webpage_html_path = webpage_fetcher.fetch(
-                        query_result["url"],
-                        overwrite=item.overwrite_webpage_html,
-                        output_parent=query_search_result["query"],
-                    )
-                    extracted_content = webpage_content_extractor.extract(
-                        webpage_html_path
-                    )
-                    queries_search_results[query_idx]["query_results"][
-                        query_result_idx
-                    ]["text"] = extracted_content
+        if item.extract_webpage:
+            queries_search_results = self.extract_webpages(
+                queries_search_results,
+                overwrite_webpage_html=item.overwrite_webpage_html,
+            )
+        return queries_search_results
+
+    def extract_webpages(self, queries_search_results, overwrite_webpage_html=False):
+        for query_idx, query_search_results in enumerate(queries_search_results):
+            # Fetch webpages with urls
+            batch_webpage_fetcher = BatchWebpageFetcher()
+            urls = [
+                query_result["url"]
+                for query_result in query_search_results["query_results"]
+            ]
+            url_and_html_path_list = batch_webpage_fetcher.fetch(
+                urls,
+                overwrite=overwrite_webpage_html,
+                output_parent=query_search_results["query"],
+            )
+            html_paths = [
+                url_and_html_path["output_path"]
+                for url_and_html_path in url_and_html_path_list
+            ]
+
+            # Extract webpage contents from htmls
+            batch_webpage_content_extractor = BatchWebpageContentExtractor()
+            html_path_and_extracted_content_list = (
+                batch_webpage_content_extractor.extract(html_paths)
+            )
+
+            # Write extracted contents (as 'text' field) to query_search_results
+            url_and_extracted_content_dict = {}
+
+            for item in url_and_html_path_list:
+                url = item["url"]
+                html_path = item["output_path"]
+                extracted_content = html_path_and_extracted_content_list[
+                    html_paths.index(html_path)
+                ]["extracted_content"]
+                url_and_extracted_content_dict[url] = extracted_content
+
+            for query_result_idx, query_result in enumerate(
+                query_search_results["query_results"]
+            ):
+                url = query_result["url"]
+                extracted_content = url_and_extracted_content_dict[url]
+                queries_search_results[query_idx]["query_results"][query_result_idx][
+                    "text"
+                ] = extracted_content
+
         return queries_search_results
 
     def setup_routes(self):