Done implementing base of requests_driver.py and chrome_driver.py (#6) (#10)

Author: kggold4

requirements.txt

@@ -3,4 +3,5 @@ selenium~=4.5.0
 setuptools==65.5.1
 requests~=2.28.2
 bs4~=0.0.1
-beautifulsoup4~=4.12.0
+beautifulsoup4~=4.12.0
+lxml~=4.9.2

scrapers/scraper_component/chrome_driver.py

@@ -1,8 +1,10 @@
 from datetime import datetime
 from time import sleep
+from typing import List
 
-from selenium.common import InvalidArgumentException, NoSuchElementException, TimeoutException
+from selenium.common import InvalidArgumentException, NoSuchElementException, TimeoutException, WebDriverException
 from selenium.webdriver import ActionChains, Keys
+from selenium.webdriver.chrome.options import Options
 
 from logger import get_current_logger, log_function
 from scrapers.scraper_component.utils.driver_consts import BrowserConsts, MainConsts
@@ -11,6 +13,8 @@
     create_path_if_needed, kill_browser_childes
 from selenium import webdriver
 
+from scrapers.scraper_component.utils.element import Element
+
 
 class ChromeDriver(BaseDriverInterface):
     def __init__(self, browser_type: str = BrowserConsts.CHROME, browser_profile_path: str = None,
@@ -73,13 +77,19 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     @log_function
     def __init_chrome_driver__(self):
         try:
-            options = webdriver.ChromeOptions()
+            chrome_options = Options()
+            chrome_options.add_argument('--no-sandbox')
+            chrome_options.add_argument('--disable-dev-shm-usage')
+            chrome_options.add_argument(argument=f"user-data-dir={self.browser_profile_path}")
             if self.headless:
-                options.add_argument('--headless')
-            options.add_argument('--no-sandbox')
-            options.add_argument('--disable-dev-shm-usage')
-            options.add_argument(argument=f"user-data-dir={self.browser_profile_path}")
-            self._driver = webdriver.Chrome(executable_path=self.webdriver_path, options=options)
+                chrome_options.add_argument("--headless")
+                chrome_options.add_argument('start-maximized')
+                chrome_options.add_argument('disable-infobars')
+                chrome_options.add_argument("--disable-extensions")
+            start_time = datetime.now()
+            self._driver = webdriver.Chrome(executable_path=self.webdriver_path, options=chrome_options)
+            end_time = datetime.now()
+            self.logger.info(f"Init chrome driver in {(end_time - start_time).total_seconds()} seconds")
         except Exception as e:
             if "executable needs to be in path" in str(e).lower():
                 self.logger.error(f"PATH Error")
@@ -93,31 +103,49 @@ def __init_chrome_driver__(self):
     @log_function
     def exit(self):
         self._driver.quit()
-        self.logger.info(f"ChromeDriver exit")
+        self.logger.info(f"Exit Chrome Driver")
 
     @log_function
     def get_url(self, url: str):
-        try:
-            self._driver.get(url)
-        except InvalidArgumentException:
-            self.logger.error(f"Error getting url: '{url}' - invalid url input format, please give full correct format")
-            self.exit()
+        for trie in range(MainConsts.GET_URL_TRIES):
+            try:
+                self.logger.debug(f"Trying to get page url: `{url}` NO. {trie + 1}/{MainConsts.GET_URL_TRIES}")
+                self._driver.get(url)
+                self.logger.info(f"Get to page url: `{url}`")
+                return
+            except InvalidArgumentException:
+                desc = f"Error getting url: '{url}' - invalid url input format, please give full correct format"
+                self.__error_and_exit(desc)
+            except WebDriverException as e:
+                if "ERR_CONNECTION_RESET" in str(e):
+                    continue
+                desc = f"Error getting to page url: `{url}` - {str(e)}"
+                self.__error_and_exit(desc)
+            except Exception as e:
+                desc = f"Error getting to page url: `{url}` - {str(e)}"
+                self.__error_and_exit(desc)
+        self.__error_and_exit(f"Error getting to page url: `{url}` after {MainConsts.GET_URL_TRIES} tries")
+
+    @log_function
+    def __error_and_exit(self, desc):
+        self.logger.error(desc)
+        self.exit()
 
     @log_function
     def get_current_url(self) -> str:
-        return self._driver.current_url
+        return self._driver.current_url if self._driver.current_url not in BrowserConsts.NEW_TAB_URLS else None
 
     @log_function
     def get_title(self) -> str:
-        return self._driver.title
+        return self._driver.title if self._driver.title != BrowserConsts.NEW_TAB_TITLE and self._driver.title else None
 
     @log_function
-    def find_element(self, by, value):
-        return self._driver.find_element(by=by, value=value)
+    def find_element(self, by, value) -> Element:
+        return Element(read_element=self._driver.find_element(by=by, value=value))
 
     @log_function
-    def find_elements(self, by, value):
-        return self._driver.find_elements(by=by, value=value)
+    def find_elements(self, by, value) -> List[Element]:
+        return [Element(read_element=element) for element in self._driver.find_elements(by=by, value=value)]
 
     @log_function
     def wait_until_object_appears(self, by, value, timeout: int = MainConsts.DEFAULT_ELEMENT_TIMEOUT):

scrapers/scraper_component/interfaces/base_driver_interface.py

@@ -1,3 +1,8 @@
+from typing import List
+
+from scrapers.scraper_component.utils.element import Element
+
+
 class BaseDriverInterface:
     def get_url(self, url: str):
         raise NotImplementedError
@@ -8,8 +13,11 @@ def get_current_url(self) -> str:
     def get_title(self) -> str:
         raise NotImplementedError
 
-    def find_element(self, by, value):
+    def find_element(self, by, value) -> Element:
+        raise NotImplementedError
+
+    def find_elements(self, by, value) -> List[Element]:
         raise NotImplementedError
 
-    def find_elements(self, by, value):
+    def exit(self):
         raise NotImplementedError

scrapers/scraper_component/interfaces/element_interface.py

@@ -0,0 +1,12 @@
+class ElementInterface:
+    def get_text(self) -> str:
+        raise NotImplementedError
+
+    def get_tag_name(self) -> str:
+        raise NotImplementedError
+
+    def get_attribute(self, attribute: str) -> str:
+        raise NotImplementedError
+
+    def is_hidden(self) -> bool:
+        raise NotImplementedError

scrapers/scraper_component/requests_driver.py

@@ -1,49 +1,105 @@
 import requests
 from bs4 import BeautifulSoup
+from selenium.common import NoSuchElementException
+from selenium.webdriver.common.by import By
 
-from logger import get_current_logger
-from scrapers.scraper_component.utils.exceptions import PageNotFoundException
+from logger import get_current_logger, log_function
+from scrapers.scraper_component.utils.driver_consts import MainConsts
+from scrapers.scraper_component.utils.element import Element
+from scrapers.scraper_component.utils.exceptions import PageNotFoundException, AttributeNameException
 from scrapers.scraper_component.interfaces.base_driver_interface import BaseDriverInterface
 
+from urllib.request import urlopen
+from lxml import etree
+
 from typing import List
 
 
 class RequestsDriver(BaseDriverInterface):
-    def __init__(self):
+    def __init__(self, headless: bool = False):
         self.logger = get_current_logger()
         self.url = None
         self._current_soup_page = None
+        self.headless = headless
 
+    @log_function
+    def exit(self):
+        self.url = None
+        self._current_soup_page = None
+        self.logger.info(f"Exit Request Driver")
+
+    @log_function
     def _get_page_tag_names(self) -> List[str]:
         return list({tag.name for tag in self._current_soup_page.findAll()})
 
+    @log_function
     def get_url(self, url: str):
         try:
-            page_res = requests.get(url)
-            self._current_soup_page = BeautifulSoup(page_res.text, 'html.parser')
-            self.url = page_res.url
-            self.logger.info(f"Get to page url: `{self.url}`")
+            for trie in range(MainConsts.GET_URL_TRIES):
+                self.logger.debug(f"Trying to get page url: `{url}` NO. {trie + 1}/{MainConsts.GET_URL_TRIES}")
+                page_res = requests.get(url, timeout=MainConsts.REQUEST_TIMEOUT)
+                if page_res.status_code == 200:
+                    self._current_soup_page = BeautifulSoup(page_res.text, 'html.parser')
+                    self.url = page_res.url
+                    self.logger.info(f"Get to page url: `{url}`")
+                    return
+            raise PageNotFoundException(f"Error getting page url: `{self.url}` after {MainConsts.GET_URL_TRIES} tries")
         except Exception as e:
             desc = f"Error getting url: `{url}` - `{e}`"
             self.logger.error(desc)
             raise PageNotFoundException(desc)
 
+    @log_function
     def get_current_url(self) -> str:
         return self.url
 
+    @log_function
     def get_title(self) -> str:
-        return self._current_soup_page.title.text
-
-    def find_element(self, by, value):
-        # return self._current_soup_page.find(name=tag_name, attrs={by: value})
-        pass
+        return self._current_soup_page.title.text if self._current_soup_page else None
 
-    def find_elements(self, by, value):
-        pass
+    @log_function
+    def find_element(self, by, value) -> Element:
+        element = None
+        try:
+            if by == By.ID:
+                element = self._current_soup_page.find(attrs={"id": value})
+            elif by == By.CLASS_NAME:
+                element = self._current_soup_page.find(attrs={"class": value})
+            elif by == By.XPATH:
+                htmlparser = etree.HTMLParser()
+                response = urlopen(self.url)
+                tree = etree.parse(response, htmlparser)
+                element = tree.xpath(value)[0]
+            else:
+                raise AttributeNameException(f"Cannot find element by: `{by}`")
+        except Exception as e:
+            self.__raise_no_such_element_exception(by=by, value=value, exception=e)
+        if element is None:
+            self.__raise_no_such_element_exception(by=by, value=value, exception=NoSuchElementException)
+        return Element(read_element=element)
 
+    def __raise_no_such_element_exception(self, by, value, exception):
+        desc = f"Cannot find element by: `{by}` with value: `{value}` - {str(exception)}"
+        self.logger.error(desc)
+        raise NoSuchElementException(desc)
 
-if __name__ == '__main__':
-    rd = RequestsDriver()
-    rd.get_url("https://www.bbc.com/")
-    element = rd.find_element(by="class", value="block-link__overlay-link")
-    print(element)
+    @log_function
+    def find_elements(self, by, value) -> List[Element]:
+        elements = None
+        try:
+            if by == By.ID:
+                elements = self._current_soup_page.findAll(attrs={"id": value})
+            elif by == By.CLASS_NAME:
+                elements = self._current_soup_page.findAll(attrs={"class": value})
+            elif by == By.XPATH:
+                htmlparser = etree.HTMLParser()
+                response = urlopen(self.url)
+                tree = etree.parse(response, htmlparser)
+                elements = tree.xpath(value)
+            else:
+                raise AttributeNameException(f"Cannot find element by: `{by}`")
+        except Exception as e:
+            self.__raise_no_such_element_exception(by=by, value=value, exception=e)
+        if elements is None:
+            return list()
+        return [Element(read_element=element) for element in elements]