Merge pull request #584 from surapuramakhil/ft_multi_job_portal_suppo…

…rt_refactoring

refactor changes (boy Scot) - for Making Project Multiple Job Portals Applier - Part 1

main.py

@@ -9,17 +9,24 @@
 from webdriver_manager.chrome import ChromeDriverManager
 from selenium.common.exceptions import WebDriverException
 from lib_resume_builder_AIHawk import Resume, FacadeManager, ResumeGenerator, StyleManager
+from typing import Optional
 from src.utils import chrome_browser_options
-from src.llm.llm_manager import GPTAnswerer
-from src.aihawk_authenticator import AIHawkAuthenticator
-from src.aihawk_bot_facade import AIHawkBotFacade
-from src.aihawk_job_manager import AIHawkJobManager
+
 from src.job_application_profile import JobApplicationProfile
 from loguru import logger
 
 # Suppress stderr only during specific operations
 original_stderr = sys.stderr
 
+# Add the src directory to the Python path
+sys.path.append(str(Path(__file__).resolve().parent / 'src'))
+
+from ai_hawk.authenticator import get_authenticator
+from ai_hawk.bot_facade import AIHawkBotFacade
+from ai_hawk.job_manager import AIHawkJobManager
+from ai_hawk.llm.llm_manager import GPTAnswerer
+
+
 class ConfigError(Exception):
     pass
 
@@ -172,7 +179,7 @@ def create_and_run_bot(parameters, llm_api_key):
         job_application_profile_object = JobApplicationProfile(plain_text_resume)
 
         browser = init_browser()
-        login_component = AIHawkAuthenticator(browser)
+        login_component = get_authenticator(driver=browser, platform='linkedin')
         apply_component = AIHawkJobManager(browser)
         gpt_answerer_component = GPTAnswerer(parameters, llm_api_key)
         bot = AIHawkBotFacade(login_component, apply_component)
@@ -195,7 +202,7 @@ def create_and_run_bot(parameters, llm_api_key):
 @click.command()
 @click.option('--resume', type=click.Path(exists=True, file_okay=True, dir_okay=False, path_type=Path), help="Path to the resume PDF file")
 @click.option('--collect', is_flag=True, help="Only collects data job information into data.json file")
-def main(collect: False, resume: Path = None):
+def main(collect: bool = False, resume: Optional[Path] = None):
     try:
         data_folder = Path("data_folder")
         secrets_file, config_file, plain_text_resume_file, output_folder = FileManager.validate_data_folder(data_folder)

pytest.ini

@@ -2,4 +2,5 @@
 minversion = 6.0
 addopts = --strict-markers --tb=short --cov=src --cov-report=term-missing
 testpaths =
-    tests
+    tests
+pythonpath = src

src/aihawk_authenticator.py → src/ai_hawk/authenticator.py

@@ -1,56 +1,75 @@
 import random
 import time
 
+from abc import ABC, abstractmethod
 from selenium.common.exceptions import NoSuchElementException, TimeoutException, NoAlertPresentException, TimeoutException, UnexpectedAlertPresentException
 from selenium.webdriver.common.by import By
 from selenium.webdriver.support import expected_conditions as EC
-from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support.wait import WebDriverWait
 
 from loguru import logger
 
+def get_authenticator(driver, platform):
+    if platform == 'linkedin':
+        return LinkedInAuthenticator(driver)
+    else:
+        raise NotImplementedError(f"Platform {platform} not implemented yet.")
 
-class AIHawkAuthenticator:
+class AIHawkAuthenticator(ABC):
 
-    def __init__(self, driver=None):
+    @property
+    def home_url(self):
+        pass
+
+    @abstractmethod
+    def navigate_to_login(self):
+        pass
+
+    @property
+    def is_logged_in(self):
+        pass
+
+    def __init__(self, driver):
         self.driver = driver
         logger.debug(f"AIHawkAuthenticator initialized with driver: {driver}")
 
     def start(self):
         logger.info("Starting Chrome browser to log in to AIHawk.")
-        if self.is_logged_in():
+        self.driver.get(self.home_url)
+        if self.is_logged_in:
             logger.info("User is already logged in. Skipping login process.")
             return
         else:
             logger.info("User is not logged in. Proceeding with login.")
             self.handle_login()
 
     def handle_login(self):
-        logger.info("Navigating to the AIHawk login page...")
-        self.driver.get("https://www.linkedin.com/login")
-        if 'feed' in self.driver.current_url:
-            logger.debug("User is already logged in.")
-            return
         try:
-            self.enter_credentials()
+            logger.info("Navigating to the AIHawk login page...")
+            self.navigate_to_login()
+            self.prompt_for_credentials()
         except NoSuchElementException as e:
             logger.error(f"Could not log in to AIHawk. Element not found: {e}")
-        self.handle_security_check()
+        self.handle_security_checks()
 
 
-    def enter_credentials(self):
+    def prompt_for_credentials(self):
         try:
             logger.debug("Enter credentials...")
-
             check_interval = 4  # Interval to log the current URL
             elapsed_time = 0
 
             while True:
+                # Bring the browser window to the front
+                current_window = self.driver.current_window_handle
+                self.driver.switch_to.window(current_window)
+
                 # Log current URL every 4 seconds and remind the user to log in
                 current_url = self.driver.current_url
                 logger.info(f"Please login on {current_url}")
 
                 # Check if the user is already on the feed page
-                if 'feed' in current_url:
+                if self.is_logged_in:
                     logger.debug("Login successful, redirected to feed page.")
                     break
                 else:
@@ -66,8 +85,20 @@ def enter_credentials(self):
         except TimeoutException:
             logger.error("Login form not found. Aborting login.")
 
+    @abstractmethod
+    def handle_security_checks(self):
+        pass
+
+class LinkedInAuthenticator(AIHawkAuthenticator):
 
-    def handle_security_check(self):
+    @property
+    def home_url(self):
+        return "https://www.linkedin.com"
+
+    def navigate_to_login(self):
+        return self.driver.get("https://www.linkedin.com/login")
+
+    def handle_security_checks(self):
         try:
             logger.debug("Handling security check...")
             WebDriverWait(self.driver, 10).until(
@@ -80,34 +111,12 @@ def handle_security_check(self):
             logger.info("Security check completed")
         except TimeoutException:
             logger.error("Security check not completed. Please try again later.")
-
+
+    @property
     def is_logged_in(self):
-        try:
-            self.driver.get('https://www.linkedin.com/feed')
-            logger.debug("Checking if user is logged in...")
-            WebDriverWait(self.driver, 3).until(
-                EC.presence_of_element_located((By.CLASS_NAME, 'share-box-feed-entry__trigger'))
-            )
-
-            # Check for the presence of the "Start a post" button
-            buttons = self.driver.find_elements(By.CLASS_NAME, 'share-box-feed-entry__trigger')
-            logger.debug(f"Found {len(buttons)} 'Start a post' buttons")
-
-            for i, button in enumerate(buttons):
-                logger.debug(f"Button {i + 1} text: {button.text.strip()}")
+        keywords = ['feed', 'mynetwork','jobs','messaging','notifications']
+        return any(item in self.driver.current_url for item in keywords) and 'linkedin.com' in self.driver.current_url
 
-            if any(button.text.strip().lower() == 'start a post' for button in buttons):
-                logger.info("Found 'Start a post' button indicating user is logged in.")
-                return True
-
-            profile_img_elements = self.driver.find_elements(By.XPATH, "//img[contains(@alt, 'Photo of')]")
-            if profile_img_elements:
-                logger.info("Profile image found. Assuming user is logged in.")
-                return True
-
-            logger.info("Did not find 'Start a post' button or profile image. User might not be logged in.")
-            return False
-
-        except TimeoutException:
-            logger.error("Page elements took too long to load or were not found.")
-            return False
+    def __init__(self, driver):
+        super().__init__(driver)
+        pass

src/aihawk_job_manager.py → src/ai_hawk/job_manager.py

@@ -9,10 +9,12 @@
 from selenium.common.exceptions import NoSuchElementException
 from selenium.webdriver.common.by import By
 
+
+from ai_hawk.linkedIn_easy_applier import AIHawkEasyApplier
 import src.utils as utils
 from app_config import MINIMUM_WAIT_TIME
 from src.job import Job
-from src.aihawk_easy_applier import AIHawkEasyApplier
+
 from loguru import logger
 import urllib.parse
 

tests/test_aihawk_easy_applier.py

@@ -1,6 +1,8 @@
 import pytest
 from unittest import mock
-from src.aihawk_easy_applier import AIHawkEasyApplier
+
+from ai_hawk.linkedIn_easy_applier import AIHawkEasyApplier
+
 
 
 @pytest.fixture

tests/test_aihawk_job_manager.py

@@ -3,7 +3,7 @@
 from pathlib import Path
 import os
 import pytest
-from src.aihawk_job_manager import AIHawkJobManager
+from ai_hawk.job_manager import AIHawkJobManager
 from selenium.common.exceptions import NoSuchElementException
 from loguru import logger
 

tests/test_aihawk_authenticator.py → tests/test_linkedIn_authenticator.py

@@ -1,11 +1,15 @@
+from httpx import get
+from numpy import place
 import pytest
 from selenium.webdriver.common.by import By
-from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support.wait import WebDriverWait
 from selenium.webdriver.support import expected_conditions as EC
-from src.aihawk_authenticator import AIHawkAuthenticator
+from ai_hawk.authenticator import AIHawkAuthenticator, LinkedInAuthenticator, get_authenticator
 from selenium.common.exceptions import NoSuchElementException, TimeoutException
 
 
+
+
 @pytest.fixture
 def mock_driver(mocker):
     """Fixture to mock the Selenium WebDriver."""
@@ -15,14 +19,14 @@ def mock_driver(mocker):
 @pytest.fixture
 def authenticator(mock_driver):
     """Fixture to initialize AIHawkAuthenticator with a mocked driver."""
-    return AIHawkAuthenticator(mock_driver)
+    return get_authenticator(mock_driver, platform='linkedin') 
 
 
 def test_handle_login(mocker, authenticator):
     """Test handling the AIHawk login process."""
     mocker.patch.object(authenticator.driver, 'get')
-    mocker.patch.object(authenticator, 'enter_credentials')
-    mocker.patch.object(authenticator, 'handle_security_check')
+    mocker.patch.object(authenticator, 'prompt_for_credentials')
+    mocker.patch.object(authenticator, 'handle_security_checks')
 
     # Mock current_url as a regular return value, not PropertyMock
     mocker.patch.object(authenticator.driver, 'current_url',
@@ -32,8 +36,8 @@ def test_handle_login(mocker, authenticator):
 
     authenticator.driver.get.assert_called_with(
         'https://www.linkedin.com/login')
-    authenticator.enter_credentials.assert_called_once()
-    authenticator.handle_security_check.assert_called_once()
+    authenticator.prompt_for_credentials.assert_called_once()
+    authenticator.handle_security_checks.assert_called_once()
 
 
 def test_enter_credentials_success(mocker, authenticator):
@@ -45,28 +49,33 @@ def test_enter_credentials_success(mocker, authenticator):
     mocker.patch.object(authenticator.driver, 'find_element',
                         return_value=password_mock)
 
-
-
-
-
-
-def test_is_logged_in_true(mocker, authenticator):
-    """Test if the user is logged in."""
-    buttons_mock = mocker.Mock()
-    buttons_mock.text = "Start a post"
-    mocker.patch.object(WebDriverWait, 'until')
-    mocker.patch.object(authenticator.driver, 'find_elements',
-                        return_value=[buttons_mock])
-
-    assert authenticator.is_logged_in() is True
-
-
-def test_is_logged_in_false(mocker, authenticator):
-    """Test if the user is not logged in."""
-    mocker.patch.object(WebDriverWait, 'until')
-    mocker.patch.object(authenticator.driver, 'find_elements', return_value=[])
-
-    assert authenticator.is_logged_in() is False
+def test_is_logged_in_true(mock_driver):
+    # Mock the current_url to simulate a logged-in state
+    mock_driver.current_url = "https://www.linkedin.com/feed/"
+    authenticator = LinkedInAuthenticator(mock_driver)
+
+    assert authenticator.is_logged_in == True
+
+def test_is_logged_in_false(mock_driver):
+    # Mock the current_url to simulate a logged-out state
+    mock_driver.current_url = "https://www.linkedin.com/login"
+    authenticator = LinkedInAuthenticator(mock_driver)
+
+    assert authenticator.is_logged_in == False
+
+def test_is_logged_in_partial_keyword(mock_driver):
+    # Mock the current_url to simulate a URL containing a keyword but not logged in
+    mock_driver.current_url = "https://www.linkedin.com/jobs/search/"
+    authenticator = LinkedInAuthenticator(mock_driver)
+
+    assert authenticator.is_logged_in == True
+
+def test_is_logged_in_no_linkedin(mock_driver):
+    # Mock the current_url to simulate a URL not related to LinkedIn
+    mock_driver.current_url = "https://www.example.com/feed/"
+    authenticator = LinkedInAuthenticator(mock_driver)
+
+    assert authenticator.is_logged_in == False
 
 
 def test_handle_security_check_success(mocker, authenticator):
@@ -76,7 +85,7 @@ def test_handle_security_check_success(mocker, authenticator):
         mocker.Mock()   # Security check completion
     ])
 
-    authenticator.handle_security_check()
+    authenticator.handle_security_checks()
 
     # Verify WebDriverWait is called with EC.url_contains for both the challenge and feed
     WebDriverWait(authenticator.driver, 10).until.assert_any_call(mocker.ANY)
@@ -87,7 +96,7 @@ def test_handle_security_check_timeout(mocker, authenticator):
     """Test handling security check timeout."""
     mocker.patch.object(WebDriverWait, 'until', side_effect=TimeoutException)
 
-    authenticator.handle_security_check()
+    authenticator.handle_security_checks()
 
     # Verify WebDriverWait is called with EC.url_contains for the challenge
     WebDriverWait(authenticator.driver, 10).until.assert_any_call(mocker.ANY)