afiqzudinhadi/linkedIn_auto_jobs_applier_with_AI
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'v3' into requirements-file

Browse source
This commit is contained in:
Thomas 2024-09-13 15:05:26 -06:00 • committed by GitHub
commit 35d0caec6b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 1674 additions and 275 deletions
Download patch file Download diff file Expand all files Collapse all files

BIN
.DS_Store vendored Normal file
View file

Binary file not shown.

13
.gitignore vendored
View file

@ -26,7 +26,9 @@ share/python-wheels/
.installed.cfg
*.egg
MANIFEST
chrome_profile/*
data_folder/*
answers.json
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
@ -148,12 +150,3 @@ venv.bak/
# Mono Auto Generated Files
mono_crash.*
# Project Specific
data_folder/output/
generated_cv/
chrome_profile/
virtual/
answers.json
# MacOS
.DS_Store

4
app_config.py Normal file
View file

@ -0,0 +1,4 @@
# LOGGING
MINIMUM_LOG_LEVEL = "DEBUG"
MINIMUM_WAIT_TIME = 60 * 15

4
data_folder/config.yaml
View file

@ -48,5 +48,5 @@ job_applicants_threshold:
max_applicants: 100
llm_model_type: openai
llm_model: gpt-4o
llm_api_url: https://api.pawan.krd/cosmosrp/v1
llm_model: gpt-4o-mini
# llm_api_url: https://api.pawan.krd/cosmosrp/v1 this field is optional

25
main.py
View file

@ -7,14 +7,15 @@ import click
from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService
from webdriver_manager.chrome import ChromeDriverManager
from selenium.common.exceptions import WebDriverException, TimeoutException
from selenium.common.exceptions import WebDriverException
from lib_resume_builder_AIHawk import Resume,StyleManager,FacadeManager,ResumeGenerator
from src.utils import chrome_browser_options
from src.gpt import GPTAnswerer
from src.llm.llm_manager import GPTAnswerer
from src.linkedIn_authenticator import LinkedInAuthenticator
from src.linkedIn_bot_facade import LinkedInBotFacade
from src.linkedIn_job_manager import LinkedInJobManager
from src.job_application_profile import JobApplicationProfile
from loguru import logger
# Suppress stderr
sys.stderr = open(os.devnull, 'w')
@ -181,7 +182,7 @@ def create_and_run_bot(email, password, parameters, llm_api_key):
bot.start_login()
bot.start_apply()
except WebDriverException as e:
print(f"WebDriver error occurred: {e}")
logger.error(f"WebDriver error occurred: {e}")
except Exception as e:
raise RuntimeError(f"Error running the bot: {str(e)}")
@ -201,20 +202,20 @@ def main(resume: Path = None):
create_and_run_bot(email, password, parameters, llm_api_key)
except ConfigError as ce:
print(f"Configuration error: {str(ce)}")
print("Refer to the configuration guide for troubleshooting: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
logger.error(f"Configuration error: {str(ce)}")
logger.error(f"Refer to the configuration guide for troubleshooting: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration {str(ce)}")
except FileNotFoundError as fnf:
print(f"File not found: {str(fnf)}")
print("Ensure all required files are present in the data folder.")
print("Refer to the file setup guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
logger.error(f"File not found: {str(fnf)}")
logger.error("Ensure all required files are present in the data folder.")
logger.error("Refer to the file setup guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
except RuntimeError as re:
print(f"Runtime error: {str(re)}")
logger.error(f"Runtime error: {str(re)}")
print("Refer to the configuration and troubleshooting guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
logger.error("Refer to the configuration and troubleshooting guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
except Exception as e:
print(f"An unexpected error occurred: {str(e)}")
print("Refer to the general troubleshooting guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
logger.error(f"An unexpected error occurred: {str(e)}")
logger.error("Refer to the general troubleshooting guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
if __name__ == "__main__":
main()

5
pytest.ini Normal file
View file

@ -0,0 +1,5 @@
[pytest]
minversion = 6.0
addopts = --strict-markers --tb=short --cov=src --cov-report=term-missing
testpaths =
tests

15
resume_yaml_generator.py
View file

@ -6,6 +6,7 @@ from typing import Dict, Any
import re
from jsonschema import validate, ValidationError
from pdfminer.high_level import extract_text
from loguru import logger
def load_yaml(file_path: str) -> Dict[str, Any]:
with open(file_path, 'r') as file:
@ -119,7 +120,7 @@ def generate_report(validation_result: Dict[str, Any], output_file: str):
report += "YAML is not valid. Errors:\n"
report += validation_result["errors"] + "\n"
print(report)
logger.debug(report)
def pdf_to_text(pdf_path: str) -> str:
return extract_text(pdf_path)
@ -137,24 +138,24 @@ def main():
# Check if input is PDF or TXT
if args.input.lower().endswith('.pdf'):
resume_text = pdf_to_text(args.input)
print(f"PDF resume converted to text successfully.")
logger.debug(f"PDF resume converted to text successfully.")
else:
resume_text = load_resume_text(args.input)
generated_yaml = generate_yaml_from_resume(resume_text, schema, api_key)
save_yaml(generated_yaml, args.output)
print(f"Resume YAML generated and saved to {args.output}")
logger.debug(f"Resume YAML generated and saved to {args.output}")
validation_result = validate_yaml(generated_yaml, schema)
if validation_result["valid"]:
print("YAML is valid and conforms to the schema.")
logger.debug("YAML is valid and conforms to the schema.")
else:
print("YAML is not valid. Errors:")
print(validation_result["errors"])
logger.error("YAML is not valid. Errors:")
logger.error(validation_result["errors"])
except Exception as e:
print(f"An error occurred: {e}")
logger.error(f"An error occurred: {e}")
if __name__ == "__main__":
main()

12
src/job.py
View file

@ -1,6 +1,6 @@
from dataclasses import dataclass
from src.utils import logger
from loguru import logger
@dataclass
@ -16,22 +16,22 @@ class Job:
recruiter_link: str = ""
def set_summarize_job_description(self, summarize_job_description):
logger.debug("Setting summarized job description: %s", summarize_job_description)
logger.debug(f"Setting summarized job description: {summarize_job_description}")
self.summarize_job_description = summarize_job_description
def set_job_description(self, description):
logger.debug("Setting job description: %s", description)
logger.debug(f"Setting job description: {description}")
self.description = description
def set_recruiter_link(self, recruiter_link):
logger.debug("Setting recruiter link: %s", recruiter_link)
logger.debug(f"Setting recruiter link: {recruiter_link}")
self.recruiter_link = recruiter_link
def formatted_job_information(self):
"""
Formats the job information as a markdown string.
"""
logger.debug("Formatting job information for job: %s at %s", self.title, self.company)
logger.debug(f"Formatting job information for job: {self.title} at {self.company}")
job_information = f"""
# Job Description
## Job Information
@ -44,5 +44,5 @@ class Job:
{self.description or 'No description provided.'}
"""
formatted_information = job_information.strip()
logger.debug("Formatted job information: %s", formatted_information)
logger.debug(f"Formatted job information: {formatted_information}")
return formatted_information

62
src/job_application_profile.py
View file

@ -2,7 +2,7 @@ from dataclasses import dataclass
import yaml
from src.utils import logger
from loguru import logger
@dataclass
@ -58,106 +58,106 @@ class JobApplicationProfile:
logger.debug("Initializing JobApplicationProfile with provided YAML string")
try:
data = yaml.safe_load(yaml_str)
logger.debug("YAML data successfully parsed: %s", data)
logger.debug(f"YAML data successfully parsed: {data}")
except yaml.YAMLError as e:
logger.error("Error parsing YAML file: %s", e)
logger.error(f"Error parsing YAML file: {e}")
raise ValueError("Error parsing YAML file.") from e
except Exception as e:
logger.error("Unexpected error occurred while parsing the YAML file: %s", e)
logger.error(f"Unexpected error occurred while parsing the YAML file: {e}")
raise RuntimeError("An unexpected error occurred while parsing the YAML file.") from e
if not isinstance(data, dict):
logger.error("YAML data must be a dictionary, received: %s", type(data))
logger.error(f"YAML data must be a dictionary, received: {type(data)}")
raise TypeError("YAML data must be a dictionary.")
# Process self_identification
try:
logger.debug("Processing self_identification")
self.self_identification = SelfIdentification(**data['self_identification'])
logger.debug("self_identification processed: %s", self.self_identification)
logger.debug(f"self_identification processed: {self.self_identification}")
except KeyError as e:
logger.error("Required field %s is missing in self_identification data.", e)
logger.error(f"Required field {e} is missing in self_identification data.")
raise KeyError(f"Required field {e} is missing in self_identification data.") from e
except TypeError as e:
logger.error("Error in self_identification data: %s", e)
logger.error(f"Error in self_identification data: {e}")
raise TypeError(f"Error in self_identification data: {e}") from e
except AttributeError as e:
logger.error("Attribute error in self_identification processing: %s", e)
logger.error(f"Attribute error in self_identification processing: {e}")
raise AttributeError("Attribute error in self_identification processing.") from e
except Exception as e:
logger.error("An unexpected error occurred while processing self_identification: %s", e)
logger.error(f"An unexpected error occurred while processing self_identification: {e}")
raise RuntimeError("An unexpected error occurred while processing self_identification.") from e
# Process legal_authorization
try:
logger.debug("Processing legal_authorization")
self.legal_authorization = LegalAuthorization(**data['legal_authorization'])
logger.debug("legal_authorization processed: %s", self.legal_authorization)
logger.debug(f"legal_authorization processed: {self.legal_authorization}")
except KeyError as e:
logger.error("Required field %s is missing in legal_authorization data.", e)
logger.error(f"Required field {e} is missing in legal_authorization data.")
raise KeyError(f"Required field {e} is missing in legal_authorization data.") from e
except TypeError as e:
logger.error("Error in legal_authorization data: %s", e)
logger.error(f"Error in legal_authorization data: {e}")
raise TypeError(f"Error in legal_authorization data: {e}") from e
except AttributeError as e:
logger.error("Attribute error in legal_authorization processing: %s", e)
logger.error(f"Attribute error in legal_authorization processing: {e}")
raise AttributeError("Attribute error in legal_authorization processing.") from e
except Exception as e:
logger.error("An unexpected error occurred while processing legal_authorization: %s", e)
logger.error(f"An unexpected error occurred while processing legal_authorization: {e}")
raise RuntimeError("An unexpected error occurred while processing legal_authorization.") from e
# Process work_preferences
try:
logger.debug("Processing work_preferences")
self.work_preferences = WorkPreferences(**data['work_preferences'])
logger.debug("work_preferences processed: %s", self.work_preferences)
logger.debug(f"Work_preferences processed: {self.work_preferences}")
except KeyError as e:
logger.error("Required field %s is missing in work_preferences data.", e)
logger.error(f"Required field {e} is missing in work_preferences data.")
raise KeyError(f"Required field {e} is missing in work_preferences data.") from e
except TypeError as e:
logger.error("Error in work_preferences data: %s", e)
logger.error(f"Error in work_preferences data: {e}")
raise TypeError(f"Error in work_preferences data: {e}") from e
except AttributeError as e:
logger.error("Attribute error in work_preferences processing: %s", e)
logger.error(f"Attribute error in work_preferences processing: {e}")
raise AttributeError("Attribute error in work_preferences processing.") from e
except Exception as e:
logger.error("An unexpected error occurred while processing work_preferences: %s", e)
logger.error(f"An unexpected error occurred while processing work_preferences: {e}")
raise RuntimeError("An unexpected error occurred while processing work_preferences.") from e
# Process availability
try:
logger.debug("Processing availability")
self.availability = Availability(**data['availability'])
logger.debug("availability processed: %s", self.availability)
logger.debug(f"Availability processed: {self.availability}")
except KeyError as e:
logger.error("Required field %s is missing in availability data.", e)
logger.error(f"Required field {e} is missing in availability data.")
raise KeyError(f"Required field {e} is missing in availability data.") from e
except TypeError as e:
logger.error("Error in availability data: %s", e)
logger.error(f"Error in availability data: {e}")
raise TypeError(f"Error in availability data: {e}") from e
except AttributeError as e:
logger.error("Attribute error in availability processing: %s", e)
logger.error(f"Attribute error in availability processing: {e}")
raise AttributeError("Attribute error in availability processing.") from e
except Exception as e:
logger.error("An unexpected error occurred while processing availability: %s", e)
logger.error(f"An unexpected error occurred while processing availability: {e}")
raise RuntimeError("An unexpected error occurred while processing availability.") from e
# Process salary_expectations
try:
logger.debug("Processing salary_expectations")
self.salary_expectations = SalaryExpectations(**data['salary_expectations'])
logger.debug("salary_expectations processed: %s", self.salary_expectations)
logger.debug(f"salary_expectations processed: {self.salary_expectations}")
except KeyError as e:
logger.error("Required field %s is missing in salary_expectations data.", e)
logger.error(f"Required field {e} is missing in salary_expectations data.")
raise KeyError(f"Required field {e} is missing in salary_expectations data.") from e
except TypeError as e:
logger.error("Error in salary_expectations data: %s", e)
logger.error(f"Error in salary_expectations data: {e}")
raise TypeError(f"Error in salary_expectations data: {e}") from e
except AttributeError as e:
logger.error("Attribute error in salary_expectations processing: %s", e)
logger.error(f"Attribute error in salary_expectations processing: {e}")
raise AttributeError("Attribute error in salary_expectations processing.") from e
except Exception as e:
logger.error("An unexpected error occurred while processing salary_expectations: %s", e)
logger.error(f"An unexpected error occurred while processing salary_expectations: {e}")
raise RuntimeError("An unexpected error occurred while processing salary_expectations.") from e
logger.debug("JobApplicationProfile initialization completed successfully.")
@ -173,5 +173,5 @@ class JobApplicationProfile:
f"Work Preferences:\n{format_dataclass(self.work_preferences)}\n\n"
f"Availability: {self.availability.notice_period}\n\n"
f"Salary Expectations: {self.salary_expectations.salary_range_usd}\n\n")
logger.debug("String representation generated: %s", formatted_str)
logger.debug(f"String representation generated: {formatted_str}")
return formatted_str

29
src/linkedIn_authenticator.py
View file

@ -6,7 +6,7 @@ from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from src.utils import logger
from loguru import logger
class LinkedInAuthenticator:
@ -15,12 +15,12 @@ class LinkedInAuthenticator:
self.driver = driver
self.email = ""
self.password = ""
logger.debug("LinkedInAuthenticator initialized with driver: %s", driver)
logger.debug(f"LinkedInAuthenticator initialized with driver: {driver}")
def set_secrets(self, email, password):
self.email = email
self.password = password
logger.debug("Secrets set with email: %s", email)
logger.debug(f"Secrets set with email: {email}")
def start(self):
logger.info("Starting Chrome browser to log in to LinkedIn.")
@ -40,13 +40,13 @@ class LinkedInAuthenticator:
logger.info("Navigating to the LinkedIn login page...")
self.driver.get("https://www.linkedin.com/login")
if 'feed' in self.driver.current_url:
print("User is already logged in.")
logger.debug("User is already logged in.")
return
try:
self.enter_credentials()
self.submit_login_form()
except NoSuchElementException as e:
logger.error("Could not log in to LinkedIn. Element not found: %s", e)
logger.error(f"Could not log in to LinkedIn. Element not found: {e}")
time.sleep(random.uniform(3, 5))
self.handle_security_check()
@ -57,13 +57,13 @@ class LinkedInAuthenticator:
EC.presence_of_element_located((By.ID, "username"))
)
email_field.send_keys(self.email)
logger.debug("Email entered: %s", self.email)
logger.debug(f"Email entered: {self.email}")
password_field = self.driver.find_element(By.ID, "password")
password_field.send_keys(self.password)
logger.debug("Password entered.")
except TimeoutException:
logger.error("Login form not found. Aborting login.")
print("Login form not found. Aborting login.")
logger.error("Login form not found. Aborting login.")
def submit_login_form(self):
try:
@ -73,7 +73,6 @@ class LinkedInAuthenticator:
logger.debug("Login form submitted.")
except NoSuchElementException:
logger.error("Login button not found. Please verify the page structure.")
print("Login button not found. Please verify the page structure.")
def handle_security_check(self):
try:
@ -82,22 +81,19 @@ class LinkedInAuthenticator:
EC.url_contains('https://www.linkedin.com/checkpoint/challengesV2/')
)
logger.warning("Security checkpoint detected. Please complete the challenge.")
print("Security checkpoint detected. Please complete the challenge.")
WebDriverWait(self.driver, 300).until(
EC.url_contains('https://www.linkedin.com/feed/')
)
logger.info("Security check completed")
print("Security check completed")
except TimeoutException:
logger.error("Security check not completed within the timeout.")
print("Security check not completed. Please try again later.")
logger.error("Security check not completed. Please try again later.")
def is_logged_in(self):
# target_url = 'https://www.linkedin.com/feed'
#
# # Navigate to the target URL if not already there
# if self.driver.current_url != target_url:
# logger.debug("Navigating to target URL: %s", target_url)
# logger.debug(f"Navigating to target URL: {target_url}")
# self.driver.get(target_url)
try:
@ -109,10 +105,10 @@ class LinkedInAuthenticator:
# 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("Found %d 'Start a post' buttons", len(buttons))
logger.debug(f"Found {len(buttons)} 'Start a post' buttons")
for i, button in enumerate(buttons):
logger.debug("Button %d text: %s", i + 1, button.text.strip())
logger.debug(f"Button {i + 1} text: {button.text.strip()}")
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.")
@ -132,11 +128,10 @@ class LinkedInAuthenticator:
def wait_for_page_load(self, timeout=10):
try:
logger.debug("Waiting for page to load with timeout: %s seconds", timeout)
logger.debug(f"Waiting for page to load with timeout: {timeout} seconds")
WebDriverWait(self.driver, timeout).until(
lambda d: d.execute_script('return document.readyState') == 'complete'
)
logger.debug("Page load completed.")
except TimeoutException:
logger.error("Page load timed out.")
print("Page load timed out.")

12
src/linkedIn_bot_facade.py
View file

@ -1,4 +1,4 @@
from src.utils import logger
from loguru import logger
class LinkedInBotState:
@ -16,10 +16,10 @@ class LinkedInBotState:
self.logged_in = False
def validate_state(self, required_keys):
logger.debug("Validating LinkedInBotState with required keys: %s", required_keys)
logger.debug(f"Validating LinkedInBotState with required keys: {required_keys}")
for key in required_keys:
if not getattr(self, key):
logger.error("State validation failed: %s is not set", key)
logger.error(f"State validation failed: {key} is not set")
raise ValueError(f"{key.replace('_', ' ').capitalize()} must be set before proceeding.")
logger.debug("State validation passed")
@ -87,11 +87,11 @@ class LinkedInBotFacade:
logger.debug("Apply process started successfully")
def _validate_non_empty(self, value, name):
logger.debug("Validating that %s is not empty", name)
logger.debug(f"Validating that {name} is not empty")
if not value:
logger.error("Validation failed: %s is empty", name)
logger.error(f"Validation failed: {name} is empty")
raise ValueError(f"{name} cannot be empty.")
logger.debug("Validation passed for %s", name)
logger.debug(f"Validation passed for {name}")
def _ensure_job_profile_and_resume_set(self):
logger.debug("Ensuring job profile and resume are set")

72
src/linkedIn_easy_applier.py
View file

@ -19,7 +19,7 @@ from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import Select, WebDriverWait
import src.utils as utils
from src.utils import logger
from loguru import logger
class LinkedInEasyApplier:
@ -39,7 +39,7 @@ class LinkedInEasyApplier:
def _load_questions_from_json(self) -> List[dict]:
output_file = 'answers.json'
logger.debug("Loading questions from JSON file: %s", output_file)
logger.debug(f"Loading questions from JSON file: {output_file}")
try:
with open(output_file, 'r') as f:
try:
@ -56,7 +56,7 @@ class LinkedInEasyApplier:
return []
except Exception:
tb_str = traceback.format_exc()
logger.error("Error loading questions data from JSON file: %s", tb_str)
logger.error(f"Error loading questions data from JSON file: {tb_str}")
raise Exception(f"Error loading questions data from JSON file: \nTraceback:\n{tb_str}")
def check_for_premium_redirect(self, job: Any, max_attempts=3):
@ -73,18 +73,32 @@ class LinkedInEasyApplier:
current_url = self.driver.current_url
if "linkedin.com/premium" in current_url:
logger.error("Failed to return to job page after %d attempts. Cannot apply for the job.", max_attempts)
logger.error(f"Failed to return to job page after {max_attempts} attempts. Cannot apply for the job.")
raise Exception(
f"Redirected to LinkedIn Premium page and failed to return after {max_attempts} attempts. Job application aborted.")
def apply_to_job(self, job: Any) -> None:
"""
Starts the process of applying to a job.
:param job: A job object with the job details.
:return: None
"""
logger.debug(f"Applying to job: {job}")
try:
self.job_apply(job)
logger.info(f"Successfully applied to job: {job.title}")
except Exception as e:
logger.error(f"Failed to apply to job: {job.title}, error: {str(e)}")
raise e
def job_apply(self, job: Any):
logger.debug("Starting job application for job: %s", job)
logger.debug(f"Starting job application for job: {job}")
try:
self.driver.get(job.link)
logger.debug("Navigated to job link: %s", job.link)
logger.debug(f"Navigated to job link: {job.link}")
except Exception as e:
logger.error("Failed to navigate to job link: %s, error: %s", job.link, str(e))
logger.error(f"Failed to navigate to job link: {job.link}, error: {str(e)}")
raise
time.sleep(random.uniform(3, 5))
@ -104,12 +118,12 @@ class LinkedInEasyApplier:
logger.debug("Retrieving job description")
job_description = self._get_job_description()
job.set_job_description(job_description)
logger.debug("Job description set: %s", job_description[:100])
logger.debug(f"Job description set: {job_description[:100]}")
logger.debug("Retrieving recruiter link")
recruiter_link = self._get_job_recruiter()
job.set_recruiter_link(recruiter_link)
logger.debug("Recruiter link set: %s", recruiter_link)
logger.debug(f"Recruiter link set: {recruiter_link}")
logger.debug("Attempting to click 'Easy Apply' button")
actions = ActionChains(self.driver)
@ -121,12 +135,12 @@ class LinkedInEasyApplier:
logger.debug("Filling out application form")
self._fill_application_form(job)
logger.debug("Job application process completed successfully for job: %s", job)
logger.debug(f"Job application process completed successfully for job: {job}")
except Exception as e:
tb_str = traceback.format_exc()
logger.error("Failed to apply to job: %s. Error traceback: %s", job, tb_str)
logger.error(f"Failed to apply to job: {job}, error: {tb_str}")
logger.debug("Discarding application due to failure")
self._discard_application()
@ -202,7 +216,7 @@ class LinkedInEasyApplier:
attempt += 1
page_source = self.driver.page_source
logger.error("No clickable 'Easy Apply' button found after 2 attempts. Page source:\n%s", page_source)
logger.error(f"No clickable 'Easy Apply' button found after 2 attempts. Page source:\n{page_source}")
raise Exception("No clickable 'Easy Apply' button found")
def _get_job_description(self) -> str:
@ -222,11 +236,11 @@ class LinkedInEasyApplier:
return description
except NoSuchElementException:
tb_str = traceback.format_exc()
logger.error("Job description not found: %s", tb_str)
logger.error(f"Job description not found: {tb_str}")
raise Exception(f"Job description not found: \nTraceback:\n{tb_str}")
except Exception:
tb_str = traceback.format_exc()
logger.error("Error getting Job description: %s", tb_str)
logger.error(f"Error getting Job description: {tb_str}")
raise Exception(f"Error getting Job description: \nTraceback:\n{tb_str}")
def _get_job_recruiter(self):
@ -243,13 +257,13 @@ class LinkedInEasyApplier:
if recruiter_elements:
recruiter_element = recruiter_elements[0]
recruiter_link = recruiter_element.get_attribute('href')
logger.debug("Job recruiter link retrieved successfully: %s", recruiter_link)
logger.debug(f"Job recruiter link retrieved successfully: {recruiter_link}")
return recruiter_link
else:
logger.debug("No recruiter link found in the hiring team section")
return ""
except Exception as e:
logger.warning("Failed to retrieve recruiter information: %s", e)
logger.warning(f"Failed to retrieve recruiter information: {e}")
return ""
def _scroll_page(self) -> None:
@ -259,7 +273,7 @@ class LinkedInEasyApplier:
utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=True)
def _fill_application_form(self, job):
logger.debug("Filling out application form for job: %s", job)
logger.debug(f"Filling out application form for job: {job}")
while True:
self.fill_up(job)
if self._next_or_submit():
@ -289,13 +303,13 @@ class LinkedInEasyApplier:
By.XPATH, "//label[contains(.,'to stay up to date with their page.')]")
follow_checkbox.click()
except Exception as e:
logger.warning("Failed to unfollow company: %s", e)
logger.debug(f"Failed to unfollow company: {e}")
def _check_for_errors(self) -> None:
logger.debug("Checking for form errors")
error_elements = self.driver.find_elements(By.CLASS_NAME, 'artdeco-inline-feedback--error')
if error_elements:
logger.error("Form submission failed with errors: %s", [e.text for e in error_elements])
logger.error(f"Form submission failed with errors: {error_elements}")
raise Exception(f"Failed answering or file upload. {str([e.text for e in error_elements])}")
def _discard_application(self) -> None:
@ -306,10 +320,10 @@ class LinkedInEasyApplier:
self.driver.find_elements(By.CLASS_NAME, 'artdeco-modal__confirm-dialog-btn')[0].click()
time.sleep(random.uniform(3, 5))
except Exception as e:
logger.warning("Failed to discard application: %s", e)
logger.warning(f"Failed to discard application: {e}")
def fill_up(self, job) -> None:
logger.debug("Filling up form sections for job: %s", job)
logger.debug(f"Filling up form sections for job: {job}")
try:
easy_apply_content = WebDriverWait(self.driver, 10).until(
@ -372,7 +386,7 @@ class LinkedInEasyApplier:
def _is_upload_field(self, element: WebElement) -> bool:
is_upload = bool(element.find_elements(By.XPATH, ".//input[@type='file']"))
logger.debug("Element is upload field: %s", is_upload)
logger.debug(f"Element is upload field: {is_upload}")
return is_upload
def _handle_upload_fields(self, element: WebElement, job) -> None:
@ -784,16 +798,16 @@ class LinkedInEasyApplier:
field_type = field.get_attribute('type').lower()
field_id = field.get_attribute("id").lower()
is_numeric = 'numeric' in field_id or field_type == 'number' or ('text' == field_type and 'numeric' in field_id)
logger.debug("Field type: %s, Field ID: %s, Is numeric: %s", field_type, field_id, is_numeric)
logger.debug(f"Field type: {field_type}, Field ID: {field_id}, Is numeric: {is_numeric}")
return is_numeric
def _enter_text(self, element: WebElement, text: str) -> None:
logger.debug("Entering text: %s", text)
logger.debug(f"Entering text: {text}")
element.clear()
element.send_keys(text)
def _select_radio(self, radios: List[WebElement], answer: str) -> None:
logger.debug("Selecting radio option: %s", answer)
logger.debug(f"Selecting radio option: {answer}")
for radio in radios:
if answer in radio.text.lower():
radio.find_element(By.TAG_NAME, 'label').click()
@ -801,14 +815,14 @@ class LinkedInEasyApplier:
radios[-1].find_element(By.TAG_NAME, 'label').click()
def _select_dropdown_option(self, element: WebElement, text: str) -> None:
logger.debug("Selecting dropdown option: %s", text)
logger.debug(f"Selecting dropdown option: {text}")
select = Select(element)
select.select_by_visible_text(text)
def _save_questions_to_json(self, question_data: dict) -> None:
output_file = 'answers.json'
question_data['question'] = self._sanitize_text(question_data['question'])
logger.debug("Saving question data to JSON: %s", question_data)
logger.debug(f"Saving question data to JSON: {question_data}")
try:
try:
with open(output_file, 'r') as f:
@ -828,11 +842,11 @@ class LinkedInEasyApplier:
logger.debug("Question data saved successfully to JSON")
except Exception:
tb_str = traceback.format_exc()
logger.error("Error saving questions data to JSON file: %s", tb_str)
logger.error(f"Error saving questions data to JSON file: {tb_str}")
raise Exception(f"Error saving questions data to JSON file: \nTraceback:\n{tb_str}")
def _sanitize_text(self, text: str) -> str:
sanitized_text = text.lower().strip().replace('"', '').replace('\\', '')
sanitized_text = re.sub(r'[\x00-\x1F\x7F]', '', sanitized_text).replace('\n', ' ').replace('\r', '').rstrip(',')
logger.debug("Sanitized text: %s", sanitized_text)
logger.debug(f"Sanitized text: {sanitized_text}")
return sanitized_text

75
src/linkedIn_job_manager.py
View file

@ -10,9 +10,10 @@ from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
import src.utils as utils
from app_config import MINIMUM_WAIT_TIME
from src.job import Job
from src.linkedIn_easy_applier import LinkedInEasyApplier
from src.utils import logger
from loguru import logger
class EnvironmentKeys:
@ -20,19 +21,18 @@ class EnvironmentKeys:
logger.debug("Initializing EnvironmentKeys")
self.skip_apply = self._read_env_key_bool("SKIP_APPLY")
self.disable_description_filter = self._read_env_key_bool("DISABLE_DESCRIPTION_FILTER")
logger.debug("EnvironmentKeys initialized: skip_apply=%s, disable_description_filter=%s",
self.skip_apply, self.disable_description_filter)
logger.debug(f"EnvironmentKeys initialized: skip_apply={self.skip_apply}, disable_description_filter={self.disable_description_filter}")
@staticmethod
def _read_env_key(key: str) -> str:
value = os.getenv(key, "")
logger.debug("Read environment key %s: %s", key, value)
logger.debug(f"Read environment key {key}: {value}")
return value
@staticmethod
def _read_env_key_bool(key: str) -> bool:
value = os.getenv(key) == "True"
logger.debug("Read environment key %s as bool: %s", key, value)
logger.debug(f"Read environment key {key} as bool: {value}")
return value
@ -79,27 +79,27 @@ class LinkedInJobManager:
searches = list(product(self.positions, self.locations))
random.shuffle(searches)
page_sleep = 0
minimum_time = 60 * 15
minimum_time = MINIMUM_WAIT_TIME
minimum_page_time = time.time() + minimum_time
for position, location in searches:
location_url = "&location=" + location
job_page_number = -1
utils.printyellow(f"Starting the search for {position} in {location}.")
logger.debug(f"Starting the search for {position} in {location}.")
try:
while True:
page_sleep += 1
job_page_number += 1
utils.printyellow(f"Going to job page {job_page_number}")
logger.debug(f"Going to job page {job_page_number}")
self.next_job_page(position, location_url, job_page_number)
time.sleep(random.uniform(1.5, 3.5))
utils.printyellow("Starting the application process for this page...")
logger.debug("Starting the application process for this page...")
try:
jobs = self.get_jobs_from_page()
if not jobs:
utils.printyellow("No more jobs found on this page. Exiting loop.")
logger.debug("No more jobs found on this page. Exiting loop.")
break
except Exception as e:
logger.error(f"Failed to retrieve jobs: {e}")
@ -108,11 +108,10 @@ class LinkedInJobManager:
try:
self.apply_jobs()
except Exception as e:
logger.error("Error during job application: %s", e)
utils.printred(f"Error during job application: {e}")
logger.error(f"Error during job application: {e}")
continue
utils.printyellow("Applying to jobs on this page has been completed!")
logger.debug("Applying to jobs on this page has been completed!")
time_left = minimum_page_time - time.time()
@ -126,10 +125,8 @@ class LinkedInJobManager:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
utils.printyellow("User skipped waiting.")
else:
logger.debug(f"Sleeping for {time_left} seconds as user chose not to skip.")
utils.printyellow(f"Sleeping for {time_left} seconds.")
time.sleep(time_left)
minimum_page_time = time.time() + minimum_time
@ -144,15 +141,12 @@ class LinkedInJobManager:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
utils.printyellow("User skipped waiting.")
else:
logger.debug(f"Sleeping for {sleep_time} seconds.")
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.")
time.sleep(sleep_time)
page_sleep += 1
except Exception as e:
logger.error("Unexpected error during job search: %s", e)
utils.printred(f"Unexpected error: {e}")
logger.error(f"Unexpected error during job search: {e}")
continue
time_left = minimum_page_time - time.time()
@ -166,10 +160,8 @@ class LinkedInJobManager:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
utils.printyellow("User skipped waiting.")
else:
logger.debug(f"Sleeping for {time_left} seconds as user chose not to skip.")
utils.printyellow(f"Sleeping for {time_left} seconds.")
time.sleep(time_left)
minimum_page_time = time.time() + minimum_time
@ -184,10 +176,8 @@ class LinkedInJobManager:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
utils.printyellow("User skipped waiting.")
else:
logger.debug(f"Sleeping for {sleep_time} seconds.")
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.")
time.sleep(sleep_time)
page_sleep += 1
@ -197,7 +187,6 @@ class LinkedInJobManager:
no_jobs_element = self.driver.find_element(By.CLASS_NAME, 'jobs-search-two-pane__no-results-banner--expand')
if 'No matching jobs found' in no_jobs_element.text or 'unfortunately, things aren' in self.driver.page_source.lower():
utils.printyellow("No matching jobs found on this page.")
logger.debug("No matching jobs found on this page, skipping.")
return []
@ -212,7 +201,6 @@ class LinkedInJobManager:
job_list_elements = self.driver.find_elements(By.CLASS_NAME, 'scaffold-layout__list-container')[
0].find_elements(By.CLASS_NAME, 'jobs-search-results__list-item')
if not job_list_elements:
utils.printyellow("No job class elements found on page.")
logger.debug("No job class elements found on page, skipping.")
return []
@ -230,7 +218,6 @@ class LinkedInJobManager:
try:
no_jobs_element = self.driver.find_element(By.CLASS_NAME, 'jobs-search-two-pane__no-results-banner--expand')
if 'No matching jobs found' in no_jobs_element.text or 'unfortunately, things aren' in self.driver.page_source.lower():
utils.printyellow("No matching jobs found on this page, moving to next.")
logger.debug("No matching jobs found on this page, skipping")
return
except NoSuchElementException:
@ -244,7 +231,6 @@ class LinkedInJobManager:
0].find_elements(By.CLASS_NAME, 'jobs-search-results__list-item')
if not job_list_elements:
utils.printyellow("No job class elements found on page, moving to next page.")
logger.debug("No job class elements found on page, skipping")
return
@ -287,8 +273,6 @@ class LinkedInJobManager:
if applicants_count is not None:
# Perform the threshold check for applicants count
if applicants_count < self.min_applicants or applicants_count > self.max_applicants:
utils.printyellow(
f"Skipping {job.title} at {job.company} due to applicants count: {applicants_count}")
logger.debug(f"Skipping {job.title} at {job.company}, applicants count: {applicants_count}")
self.write_to_file(job, "skipped_due_to_applicants")
continue # Skip this job if applicants count is outside the threshold
@ -314,8 +298,7 @@ class LinkedInJobManager:
logger.debug(f"Continuing with job application for {job.title} at {job.company}")
if self.is_blacklisted(job.title, job.company, job.link):
utils.printyellow(f"Blacklisted {job.title} at {job.company}, skipping...")
logger.debug("Job blacklisted: %s at %s", job.title, job.company)
logger.debug(f"Job blacklisted: {job.title} at {job.company}")
self.write_to_file(job, "skipped")
continue
if self.is_already_applied_to_job(job.title, job.company, job.link):
@ -328,15 +311,14 @@ class LinkedInJobManager:
if job.apply_method not in {"Continue", "Applied", "Apply"}:
self.easy_applier_component.job_apply(job)
self.write_to_file(job, "success")
logger.debug("Applied to job: %s at %s", job.title, job.company)
logger.debug(f"Applied to job: {job.title} at {job.company}")
except Exception as e:
logger.error("Failed to apply for %s at %s: %s", job.title, job.company, e)
utils.printred(f"Failed to apply for {job.title} at {job.company}: {e}")
logger.error(f"Failed to apply for {job.title} at {job.company}: {e}")
self.write_to_file(job, "failed")
continue
def write_to_file(self, job, file_name):
logger.debug("Writing job application result to file: %s", file_name)
logger.debug(f"Writing job application result to file: {file_name}")
pdf_path = Path(job.pdf_path).resolve()
pdf_path = pdf_path.as_uri()
data = {
@ -351,19 +333,19 @@ class LinkedInJobManager:
if not file_path.exists():
with open(file_path, 'w', encoding='utf-8') as f:
json.dump([data], f, indent=4)
logger.debug("Job data written to new file: %s", file_path)
logger.debug(f"Job data written to new file: {file_name}")
else:
with open(file_path, 'r+', encoding='utf-8') as f:
try:
existing_data = json.load(f)
except json.JSONDecodeError:
logger.error("JSON decode error in file: %s", file_path)
logger.error(f"JSON decode error in file: {file_path}")
existing_data = []
existing_data.append(data)
f.seek(0)
json.dump(existing_data, f, indent=4)
f.truncate()
logger.debug("Job data appended to existing file: %s", file_path)
logger.debug(f"Job data appended to existing file: {file_name}")
def get_base_search_url(self, parameters):
logger.debug("Constructing base search URL")
@ -388,11 +370,11 @@ class LinkedInJobManager:
url_parts.append("f_LF=f_AL") # Easy Apply
base_url = "&".join(url_parts)
full_url = f"?{base_url}{date_param}"
logger.debug("Base search URL constructed: %s", full_url)
logger.debug(f"Base search URL constructed: {full_url}")
return full_url
def next_job_page(self, position, location, job_page):
logger.debug("Navigating to next job page: %s in %s, page %d", position, location, job_page)
logger.debug(f"Navigating to next job page: {position} in {location}, page {job_page}")
self.driver.get(
f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}{location}&start={job_page * 25}")
@ -403,39 +385,36 @@ class LinkedInJobManager:
job_title = job_tile.find_element(By.CLASS_NAME, 'job-card-list__title').text
link = job_tile.find_element(By.CLASS_NAME, 'job-card-list__title').get_attribute('href').split('?')[0]
company = job_tile.find_element(By.CLASS_NAME, 'job-card-container__primary-description').text
logger.debug("Job information extracted: %s at %s", job_title, company)
logger.debug(f"Job information extracted: {job_title} at {company}")
except NoSuchElementException:
utils.printyellow("Some job information (title, link, or company) is missing.")
logger.warning("Some job information (title, link, or company) is missing.")
try:
job_location = job_tile.find_element(By.CLASS_NAME, 'job-card-container__metadata-item').text
except NoSuchElementException:
utils.printyellow("Job location is missing.")
logger.warning("Job location is missing.")
try:
apply_method = job_tile.find_element(By.CLASS_NAME, 'job-card-container__apply-method').text
except NoSuchElementException:
apply_method = "Applied"
utils.printyellow("Apply method not found, assuming 'Applied'.")
logger.warning("Apply method not found, assuming 'Applied'.")
return job_title, company, job_location, link, apply_method
def is_blacklisted(self, job_title, company, link):
logger.debug("Checking if job is blacklisted: %s at %s", job_title, company)
logger.debug(f"Checking if job is blacklisted: {job_title} at {company}")
job_title_words = job_title.lower().split(' ')
title_blacklisted = any(word in job_title_words for word in self.title_blacklist)
company_blacklisted = company.strip().lower() in (word.strip().lower() for word in self.company_blacklist)
link_seen = link in self.seen_jobs
is_blacklisted = title_blacklisted or company_blacklisted or link_seen
logger.debug("Job blacklisted status: %s", is_blacklisted)
logger.debug(f"Job blacklisted status: {is_blacklisted}")
return title_blacklisted or company_blacklisted or link_seen
def is_already_applied_to_job(self, job_title, company, link):
link_seen = link in self.seen_jobs
if link_seen:
utils.printyellow(f"Already applied to job: {job_title} at {company}, skipping...")
logger.debug(f"Already applied to job: {job_title} at {company}, skipping...")
return link_seen
def is_already_applied_to_company(self, company):
@ -451,7 +430,7 @@ class LinkedInJobManager:
existing_data = json.load(f)
for applied_job in existing_data:
if applied_job['company'].strip().lower() == company.strip().lower():
utils.printyellow(
logger.debug(
f"Already applied at {company} (once per company policy), skipping...")
return True
except json.JSONDecodeError:

279
src/linkedin-api.py
View file

@ -1,59 +1,59 @@
import logging
from typing import Dict, List
from typing import Optional, Union, Literal
from urllib.parse import urlencode
from linkedin_api import Linkedin
from typing import Optional, Union, Literal
from urllib.parse import quote, urlencode, parse_qs, urlparse
# import logging
import json
from loguru import logger
# set log to all debug
logging.basicConfig(level=logging.INFO)
# logging.basicConfig(level=logging.INFO)
class LinkedInEvolvedAPI(Linkedin):
already_applied_jobs: List[str] = []
def __init__(self, username, password):
super().__init__(username, password)
def search_jobs(
self,
keywords: Optional[str] = None,
companies: Optional[List[str]] = None,
experience: Optional[
List[
Union[
Literal["1"],
Literal["2"],
Literal["3"],
Literal["4"],
Literal["5"],
Literal["6"],
]
self,
keywords: Optional[str] = None,
companies: Optional[List[str]] = None,
experience: Optional[
List[
Union[
Literal["1"],
Literal["2"],
Literal["3"],
Literal["4"],
Literal["5"],
Literal["6"],
]
] = None,
job_type: Optional[
List[
Union[
Literal["F"],
Literal["C"],
Literal["P"],
Literal["T"],
Literal["I"],
Literal["V"],
Literal["O"],
]
]
] = None,
job_type: Optional[
List[
Union[
Literal["F"],
Literal["C"],
Literal["P"],
Literal["T"],
Literal["I"],
Literal["V"],
Literal["O"],
]
] = None,
job_title: Optional[List[str]] = None,
industries: Optional[List[str]] = None,
location_name: Optional[str] = None,
remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None,
listed_at: None | int = None,
distance: Optional[int] = None,
easy_apply: Optional[bool] = True,
limit=-1,
offset=0,
**kwargs,
]
] = None,
job_title: Optional[List[str]] = None,
industries: Optional[List[str]] = None,
location_name: Optional[str] = None,
remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None,
listed_at: None | int = None,
distance: Optional[int] = None,
easy_apply: Optional[bool] = True,
limit=-1,
offset=0,
**kwargs,
) -> List[Dict]:
"""Perform a LinkedIn search for jobs.
@ -155,21 +155,21 @@ class LinkedInEvolvedAPI(Linkedin):
e["job_id"] = trackingUrn
if e.get("$type") == "com.linkedin.voyager.dash.jobs.JobPosting":
new_data.append(e)
if not new_data:
break
results.extend(new_data)
if (
(-1 < limit <= len(results))
or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS
(-1 < limit <= len(results))
or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS
) or len(elements) == 0:
break
self.logger.debug(f"results grew to {len(results)}")
return results
def get_fields_for_easy_apply(self, job_id: str) -> List[Dict]:
def get_fields_for_easy_apply(self,job_id: str) -> List[Dict]:
"""Get fields needed for easy apply jobs.
:param job_id: Job ID
@ -182,12 +182,14 @@ class LinkedInEvolvedAPI(Linkedin):
cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
headers: Dict[str, str] = self._headers()
headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
headers["Cookie"] = cookie_str
headers["Connection"] = "keep-alive"
default_params = {
"decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67",
"jobPostingUrn": f"urn:li:fsd_jobPosting:{job_id}",
@ -216,26 +218,26 @@ class LinkedInEvolvedAPI(Linkedin):
except ValueError:
self.logger.error("Failed to parse JSON response")
return []
form_components = []
for item in data.get("included", []):
if 'formComponent' in item:
if 'formComponent' in item:
urn = item['urn']
try:
title = item['title']['text']
except TypeError:
title = urn
form_component_type = list(item['formComponent'].keys())[0]
form_component_details = item['formComponent'][form_component_type]
component_info = {
'title': title,
'urn': urn,
'formComponentType': form_component_type,
}
if 'textSelectableOptions' in form_component_details:
options = [
opt['optionText']['text'] for opt in form_component_details['textSelectableOptions']
@ -243,18 +245,18 @@ class LinkedInEvolvedAPI(Linkedin):
component_info['selectableOptions'] = options
elif 'selectableOptions' in form_component_details:
options = [
opt['textSelectableOption']['optionText']['text']
opt['textSelectableOption']['optionText']['text']
for opt in form_component_details['selectableOptions']
]
component_info['selectableOptions'] = options
form_components.append(component_info)
return form_components
def apply_to_job(self, job_id: str, fields: dict, followCompany: bool = True) -> bool:
def apply_to_job(self,job_id: str, fields: dict, followCompany: bool = True) -> bool:
return False
# ToDo: Implement apply to job parser first
# How need to be implemented:
# 1. Get fields for easy apply job from the previous method (get_fields_for_easy_apply)
@ -264,11 +266,11 @@ class LinkedInEvolvedAPI(Linkedin):
# {'title': 'Quanti anni di esperienza di lavoro hai con Router?', 'urn': 'urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4013860791,9478711764,numeric)', 'formComponentType': 'singleLineTextFormComponent', 'response': '5'}
# To fill, you can temporary use input() function to get the data from the user manually for testing purposes (for the further implementation, the question will be asked to AI implementation and automatically filled)
# Build a working payload.
# EXAMPLE OF WORKING PAYLOAD
# 4005350454 is job_id, so need to be replaced with the job_id
# {
#{
# "followCompany": true,
# "responses": [
# {
@ -348,29 +350,172 @@ class LinkedInEvolvedAPI(Linkedin):
# }
# ],
# "trackingId": ""
# }
#}
# Push the commit to the repository and create a pull request to the v3 branch.
def create_request_pdf(self, filename: str) -> str | None:
"""
Create a PDF file with the request data.
:param filename: Name of the file
:type filename: str | None
:return: URL of the file uploaded to the LinkedIn.
"""
cookies = self.client.session.cookies.get_dict()
cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
headers: Dict[str, str] = self._headers()
headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
headers["Cookie"] = cookie_str
headers["Connection"] = "keep-alive"
default_params = {
'action': 'requestUrl'
}
res = self._post(
f"/voyagerJobsDashAmbryUploadUrls",
headers=headers,
cookies=cookies,
json={"contentType":"PDF","filename":"200.pdf","maxSizeBytes":18810},
params=default_params
)
match res.status_code:
case 200:
parse_res = res.json()
url = parse_res['data']['value']
logger.info(url)
return url
case _:
self.logger.error("Failed to create a request PDF")
return None
def upload_resume_via_ambry(self, url: str, cv_path: str) -> bool | str:
"""
Upload resume via Ambry.
:param url: URL of the file uploaded to the LinkedIn.
:type url: str
:param raw_cv: Raw CV data
:type raw_cv: bytes
:return: PDF hash.pdf or false
"""
binary_cv: bytes = self.file_to_binary(cv_path)
cookies = self.client.session.cookies.get_dict()
cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
headers: Dict[str, str] = self._headers()
headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
headers["Cookie"] = cookie_str
headers["Connection"] = "keep-alive"
ambry_url = url.replace("https://www.linkedin.com", "")
res = self._post(
ambry_url,
base_request=True,
headers=headers,
cookies=cookies,
data=binary_cv,
)
match res.status_code:
case 201:
return res.headers['Location']
case _:
self.logger.error("Failed to upload resume via Ambry")
return False
def confirm_upload_resume(self, cv_hash: str) -> bool:
"""
Upload resume.
:param cv_hash: PDF hash
:type cv_hash: str
:return: True if success, False if failed
"""
cookies = self.client.session.cookies.get_dict()
cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
headers: Dict[str, str] = self._headers()
headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
headers["Cookie"] = cookie_str
headers["Connection"] = "keep-alive"
json_data = {'entityUrn': f'urn:li:fsd_resume:{cv_hash}'}
res = self._post(
"/voyagerJobsDashResumes",
headers=headers,
cookies=cookies,
json=json_data,
)
match res.status_code:
case 201:
return True
case _:
self.logger.error("Failed to upload resume")
return False
def file_to_binary(self, file_path):
with open(file_path, 'rb') as file:
binary_data = file.read()
return binary_data
def set_job_as_applied(self, job_id: str) -> None:
self.already_applied_jobs.append(job_id)
def upload_linkedin_resume(self, cv_path: str) -> str | bool:
url = self.create_request_pdf("resume.pdf")
if url:
cv_hash = self.upload_resume_via_ambry(url, cv_path)
if cv_hash:
self.confirm_upload_resume(cv_hash)
return cv_hash
return False
## EXAMPLE USAGE
if __name__ == "__main__":
api: LinkedInEvolvedAPI = LinkedInEvolvedAPI(username="", password="")
jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=100, easy_apply=True, offset=1,
listed_at=None)
jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=100, easy_apply=True, offset=1, listed_at=None)
for job in jobs:
job_id: str = job["job_id"]
print(f"Job ID: {job_id}")
continue
resume: str = api.upload_linkedin_resume("resume.pdf")
if isinstance(resume, bool):
logger.error("Failed to upload resume")
continue
elif isinstance(resume, str):
logger.info(f"Resume uploaded with hash {resume}")
else:
logger.error("Unknown error")
continue
if job_id in api.already_applied_jobs:
logging.info(f"Already applied to job {job_id}, skipping it")
logger.info(f"Already applied to job {job_id}, skipping it")
continue
fields = api.get_fields_for_easy_apply(job_id)
for field in fields:
print(field)
logger.info(field)
break

577
src/llm/llm_manager.py Normal file
View file

@ -0,0 +1,577 @@
import json
import os
import re
import textwrap
import time
from abc import ABC, abstractmethod
from datetime import datetime
from pathlib import Path
from typing import Dict, List
from typing import Union
import httpx
from Levenshtein import distance
from dotenv import load_dotenv
from langchain_core.messages import BaseMessage
from langchain_core.messages.ai import AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompt_values import StringPromptValue
from langchain_core.prompts import ChatPromptTemplate
import src.strings as strings
from loguru import logger
load_dotenv()
class AIModel(ABC):
@abstractmethod
def invoke(self, prompt: str) -> str:
pass
class OpenAIModel(AIModel):
def __init__(self, api_key: str, llm_model: str):
from langchain_openai import ChatOpenAI
self.model = ChatOpenAI(model_name=llm_model, openai_api_key=api_key,
temperature=0.4)
def invoke(self, prompt: str) -> BaseMessage:
logger.debug("Invoking OpenAI API")
response = self.model.invoke(prompt)
return response
class ClaudeModel(AIModel):
def __init__(self, api_key: str, llm_model: str):
from langchain_anthropic import ChatAnthropic
self.model = ChatAnthropic(model=llm_model, api_key=api_key,
temperature=0.4)
def invoke(self, prompt: str) -> BaseMessage:
response = self.model.invoke(prompt)
logger.debug("Invoking Claude API")
return response
class OllamaModel(AIModel):
def __init__(self, llm_model: str, llm_api_url: str):
from langchain_ollama import ChatOllama
if len(llm_api_url) > 0:
logger.debug(f"Using Ollama with API URL: {llm_api_url}")
self.model = ChatOllama(model=llm_model, base_url=llm_api_url)
else:
self.model = ChatOllama(model=llm_model)
def invoke(self, prompt: str) -> BaseMessage:
response = self.model.invoke(prompt)
return response
class GeminiModel(AIModel):
def __init__(self, api_key:str, llm_model: str):
from langchain_google_genai import ChatGoogleGenerativeAI
self.model = ChatGoogleGenerativeAI(model=llm_model, google_api_key=api_key)
def invoke(self, prompt: str) -> BaseMessage:
response = self.model.invoke(prompt)
return response
class AIAdapter:
def __init__(self, config: dict, api_key: str):
self.model = self._create_model(config, api_key)
def _create_model(self, config: dict, api_key: str) -> AIModel:
llm_model_type = config['llm_model_type']
llm_model = config['llm_model']
llm_api_url = config.get('llm_api_url', "")
logger.debug(f"Using {llm_model_type} with {llm_model}")
if llm_model_type == "openai":
return OpenAIModel(api_key, llm_model)
elif llm_model_type == "claude":
return ClaudeModel(api_key, llm_model)
elif llm_model_type == "ollama":
return OllamaModel(llm_model, llm_api_url)
elif llm_model_type == "gemini":
return GeminiModel(api_key, llm_model)
else:
raise ValueError(f"Unsupported model type: {llm_model_type}")
def invoke(self, prompt: str) -> str:
return self.model.invoke(prompt)
class LLMLogger:
def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel, GeminiModel]):
self.llm = llm
logger.debug(f"LLMLogger successfully initialized with LLM: {llm}")
@staticmethod
def log_request(prompts, parsed_reply: Dict[str, Dict]):
logger.debug("Starting log_request method")
logger.debug(f"Prompts received: {prompts}")
logger.debug(f"Parsed reply received: {parsed_reply}")
try:
calls_log = os.path.join(
Path("data_folder/output"), "open_ai_calls.json")
logger.debug(f"Logging path determined: {calls_log}")
except Exception as e:
logger.error(f"Error determining the log path: {str(e)}")
raise
if isinstance(prompts, StringPromptValue):
logger.debug("Prompts are of type StringPromptValue")
prompts = prompts.text
logger.debug(f"Prompts converted to text: {prompts}")
elif isinstance(prompts, Dict):
logger.debug("Prompts are of type Dict")
try:
prompts = {
f"prompt_{i + 1}": prompt.content
for i, prompt in enumerate(prompts.messages)
}
logger.debug(f"Prompts converted to dictionary: {prompts}")
except Exception as e:
logger.error(f"Error converting prompts to dictionary: {str(e)}")
raise
else:
logger.debug("Prompts are of unknown type, attempting default conversion")
try:
prompts = {
f"prompt_{i + 1}": prompt.content
for i, prompt in enumerate(prompts.messages)
}
logger.debug(f"Prompts converted to dictionary using default method: {prompts}")
except Exception as e:
logger.error(f"Error converting prompts using default method: {str(e)}")
raise
try:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
logger.debug(f"Current time obtained: {current_time}")
except Exception as e:
logger.error(f"Error obtaining current time: {str(e)}")
raise
try:
token_usage = parsed_reply["usage_metadata"]
output_tokens = token_usage["output_tokens"]
input_tokens = token_usage["input_tokens"]
total_tokens = token_usage["total_tokens"]
logger.debug(f"Token usage - Input: {input_tokens}, Output: {output_tokens}, Total: {total_tokens}")
except KeyError as e:
logger.error(f"KeyError in parsed_reply structure: {str(e)}")
raise
try:
model_name = parsed_reply["response_metadata"]["model_name"]
logger.debug(f"Model name: {model_name}")
except KeyError as e:
logger.error(f"KeyError in response_metadata: {str(e)}")
raise
try:
prompt_price_per_token = 0.00000015
completion_price_per_token = 0.0000006
total_cost = (input_tokens * prompt_price_per_token) + \
(output_tokens * completion_price_per_token)
logger.debug(f"Total cost calculated: {total_cost}")
except Exception as e:
logger.error(f"Error calculating total cost: {str(e)}")
raise
try:
log_entry = {
"model": model_name,
"time": current_time,
"prompts": prompts,
"replies": parsed_reply["content"],
"total_tokens": total_tokens,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_cost": total_cost,
}
logger.debug(f"Log entry created: {log_entry}")
except KeyError as e:
logger.error(f"Error creating log entry: missing key {str(e)} in parsed_reply")
raise
try:
with open(calls_log, "a", encoding="utf-8") as f:
json_string = json.dumps(
log_entry, ensure_ascii=False, indent=4)
f.write(json_string + "\n")
logger.debug(f"Log entry written to file: {calls_log}")
except Exception as e:
logger.error(f"Error writing log entry to file: {str(e)}")
raise
class LoggerChatModel:
def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel, GeminiModel]):
self.llm = llm
logger.debug(f"LoggerChatModel successfully initialized with LLM: {llm}")
def __call__(self, messages: List[Dict[str, str]]) -> str:
logger.debug(f"Entering __call__ method with messages: {messages}")
while True:
try:
logger.debug("Attempting to call the LLM with messages")
reply = self.llm.invoke(messages)
logger.debug(f"LLM response received: {reply}")
parsed_reply = self.parse_llmresult(reply)
logger.debug(f"Parsed LLM reply: {parsed_reply}")
LLMLogger.log_request(
prompts=messages, parsed_reply=parsed_reply)
logger.debug("Request successfully logged")
return reply
except httpx.HTTPStatusError as e:
logger.error(f"HTTPStatusError encountered: {str(e)}")
if e.response.status_code == 429:
retry_after = e.response.headers.get('retry-after')
retry_after_ms = e.response.headers.get('retry-after-ms')
if retry_after:
wait_time = int(retry_after)
logger.warning(
f"Rate limit exceeded. Waiting for {wait_time} seconds before retrying (extracted from 'retry-after' header)...")
time.sleep(wait_time)
elif retry_after_ms:
wait_time = int(retry_after_ms) / 1000.0
logger.warning(
f"Rate limit exceeded. Waiting for {wait_time} seconds before retrying (extracted from 'retry-after-ms' header)...")
time.sleep(wait_time)
else:
wait_time = 30
logger.warning(
f"'retry-after' header not found. Waiting for {wait_time} seconds before retrying (default)...")
time.sleep(wait_time)
else:
logger.error(f"HTTP error occurred with status code: {e.response.status_code}, waiting 30 seconds before retrying")
time.sleep(30)
except Exception as e:
logger.error(f"Unexpected error occurred: {str(e)}")
logger.info(
"Waiting for 30 seconds before retrying due to an unexpected error.")
time.sleep(30)
continue
def parse_llmresult(self, llmresult: AIMessage) -> Dict[str, Dict]:
logger.debug(f"Parsing LLM result: {llmresult}")
try:
content = llmresult.content
response_metadata = llmresult.response_metadata
id_ = llmresult.id
usage_metadata = llmresult.usage_metadata
parsed_result = {
"content": content,
"response_metadata": {
"model_name": response_metadata.get("model_name", ""),
"system_fingerprint": response_metadata.get("system_fingerprint", ""),
"finish_reason": response_metadata.get("finish_reason", ""),
"logprobs": response_metadata.get("logprobs", None),
},
"id": id_,
"usage_metadata": {
"input_tokens": usage_metadata.get("input_tokens", 0),
"output_tokens": usage_metadata.get("output_tokens", 0),
"total_tokens": usage_metadata.get("total_tokens", 0),
},
}
logger.debug(f"Parsed LLM result successfully: {parsed_result}")
return parsed_result
except KeyError as e:
logger.error(
f"KeyError while parsing LLM result: missing key {str(e)}")
raise
except Exception as e:
logger.error(
f"Unexpected error while parsing LLM result: {str(e)}")
raise
class GPTAnswerer:
def __init__(self, config, llm_api_key):
self.ai_adapter = AIAdapter(config, llm_api_key)
self.llm_cheap = LoggerChatModel(self.ai_adapter)
@property
def job_description(self):
return self.job.description
@staticmethod
def find_best_match(text: str, options: list[str]) -> str:
logger.debug(f"Finding best match for text: '{text}' in options: {options}")
distances = [
(option, distance(text.lower(), option.lower())) for option in options
]
best_option = min(distances, key=lambda x: x[1])[0]
logger.debug(f"Best match found: {best_option}")
return best_option
@staticmethod
def _remove_placeholders(text: str) -> str:
logger.debug(f"Removing placeholders from text: {text}")
text = text.replace("PLACEHOLDER", "")
return text.strip()
@staticmethod
def _preprocess_template_string(template: str) -> str:
logger.debug("Preprocessing template string")
return textwrap.dedent(template)
def set_resume(self, resume):
logger.debug(f"Setting resume: {resume}")
self.resume = resume
def set_job(self, job):
logger.debug(f"Setting job: {job}")
self.job = job
self.job.set_summarize_job_description(
self.summarize_job_description(self.job.description))
def set_job_application_profile(self, job_application_profile):
logger.debug(f"Setting job application profile: {job_application_profile}")
self.job_application_profile = job_application_profile
def summarize_job_description(self, text: str) -> str:
logger.debug(f"Summarizing job description: {text}")
strings.summarize_prompt_template = self._preprocess_template_string(
strings.summarize_prompt_template
)
prompt = ChatPromptTemplate.from_template(
strings.summarize_prompt_template)
chain = prompt | self.llm_cheap | StrOutputParser()
output = chain.invoke({"text": text})
logger.debug(f"Summary generated: {output}")
return output
def _create_chain(self, template: str):
logger.debug(f"Creating chain with template: {template}")
prompt = ChatPromptTemplate.from_template(template)
return prompt | self.llm_cheap | StrOutputParser()
def answer_question_textual_wide_range(self, question: str) -> str:
logger.debug(f"Answering textual question: {question}")
chains = {
"personal_information": self._create_chain(strings.personal_information_template),
"self_identification": self._create_chain(strings.self_identification_template),
"legal_authorization": self._create_chain(strings.legal_authorization_template),
"work_preferences": self._create_chain(strings.work_preferences_template),
"education_details": self._create_chain(strings.education_details_template),
"experience_details": self._create_chain(strings.experience_details_template),
"projects": self._create_chain(strings.projects_template),
"availability": self._create_chain(strings.availability_template),
"salary_expectations": self._create_chain(strings.salary_expectations_template),
"certifications": self._create_chain(strings.certifications_template),
"languages": self._create_chain(strings.languages_template),
"interests": self._create_chain(strings.interests_template),
"cover_letter": self._create_chain(strings.coverletter_template),
}
section_prompt = """
You are assisting a bot designed to automatically apply for jobs on LinkedIn. The bot receives various questions about job applications and needs to determine the most relevant section of the resume to provide an accurate response.
For the following question: '{question}', determine which section of the resume is most relevant.
Respond with exactly one of the following options:
- Personal information
- Self Identification
- Legal Authorization
- Work Preferences
- Education Details
- Experience Details
- Projects
- Availability
- Salary Expectations
- Certifications
- Languages
- Interests
- Cover letter
Here are detailed guidelines to help you choose the correct section:
1. **Personal Information**:
- **Purpose**: Contains your basic contact details and online profiles.
- **Use When**: The question is about how to contact you or requests links to your professional online presence.
- **Examples**: Email address, phone number, LinkedIn profile, GitHub repository, personal website.
2. **Self Identification**:
- **Purpose**: Covers personal identifiers and demographic information.
- **Use When**: The question pertains to your gender, pronouns, veteran status, disability status, or ethnicity.
- **Examples**: Gender, pronouns, veteran status, disability status, ethnicity.
3. **Legal Authorization**:
- **Purpose**: Details your work authorization status and visa requirements.
- **Use When**: The question asks about your ability to work in specific countries or if you need sponsorship or visas.
- **Examples**: Work authorization in EU and US, visa requirements, legally allowed to work.
4. **Work Preferences**:
- **Purpose**: Specifies your preferences regarding work conditions and job roles.
- **Use When**: The question is about your preferences for remote work, in-person work, relocation, and willingness to undergo assessments or background checks.
- **Examples**: Remote work, in-person work, open to relocation, willingness to complete assessments.
5. **Education Details**:
- **Purpose**: Contains information about your academic qualifications.
- **Use When**: The question concerns your degrees, universities attended, GPA, and relevant coursework.
- **Examples**: Degree, university, GPA, field of study, exams.
6. **Experience Details**:
- **Purpose**: Details your professional work history and key responsibilities.
- **Use When**: The question pertains to your job roles, responsibilities, and achievements in previous positions.
- **Examples**: Job positions, company names, key responsibilities, skills acquired.
7. **Projects**:
- **Purpose**: Highlights specific projects you have worked on.
- **Use When**: The question asks about particular projects, their descriptions, or links to project repositories.
- **Examples**: Project names, descriptions, links to project repositories.
8. **Availability**:
- **Purpose**: Provides information on your availability for new roles.
- **Use When**: The question is about how soon you can start a new job or your notice period.
- **Examples**: Notice period, availability to start.
9. **Salary Expectations**:
- **Purpose**: Covers your expected salary range.
- **Use When**: The question pertains to your salary expectations or compensation requirements.
- **Examples**: Desired salary range.
10. **Certifications**:
- **Purpose**: Lists your professional certifications or licenses.
- **Use When**: The question involves your certifications or qualifications from recognized organizations.
- **Examples**: Certification names, issuing bodies, dates of validity.
11. **Languages**:
- **Purpose**: Describes the languages you can speak and your proficiency levels.
- **Use When**: The question asks about your language skills or proficiency in specific languages.
- **Examples**: Languages spoken, proficiency levels.
12. **Interests**:
- **Purpose**: Details your personal or professional interests.
- **Use When**: The question is about your hobbies, interests, or activities outside of work.
- **Examples**: Personal hobbies, professional interests.
13. **Cover Letter**:
- **Purpose**: Contains your personalized cover letter or statement.
- **Use When**: The question involves your cover letter or specific written content intended for the job application.
- **Examples**: Cover letter content, personalized statements.
Provide only the exact name of the section from the list above with no additional text.
"""
prompt = ChatPromptTemplate.from_template(section_prompt)
chain = prompt | self.llm_cheap | StrOutputParser()
output = chain.invoke({"question": question})
match = re.search(
r"(Personal information|Self Identification|Legal Authorization|Work Preferences|Education "
r"Details|Experience Details|Projects|Availability|Salary "
r"Expectations|Certifications|Languages|Interests|Cover letter)",
output, re.IGNORECASE)
if not match:
raise ValueError(
"Could not extract section name from the response.")
section_name = match.group(1).lower().replace(" ", "_")
if section_name == "cover_letter":
chain = chains.get(section_name)
output = chain.invoke(
{"resume": self.resume, "job_description": self.job_description})
logger.debug(f"Cover letter generated: {output}")
return output
resume_section = getattr(self.resume, section_name, None) or getattr(self.job_application_profile, section_name,
None)
if resume_section is None:
logger.error(
f"Section '{section_name}' not found in either resume or job_application_profile.")
raise ValueError(f"Section '{section_name}' not found in either resume or job_application_profile.")
chain = chains.get(section_name)
if chain is None:
logger.error(f"Chain not defined for section '{section_name}'")
raise ValueError(f"Chain not defined for section '{section_name}'")
output = chain.invoke(
{"resume_section": resume_section, "question": question})
logger.debug(f"Question answered: {output}")
return output
def answer_question_numeric(self, question: str, default_experience: int = 3) -> int:
logger.debug(f"Answering numeric question: {question}")
func_template = self._preprocess_template_string(
strings.numeric_question_template)
prompt = ChatPromptTemplate.from_template(func_template)
chain = prompt | self.llm_cheap | StrOutputParser()
output_str = chain.invoke(
{"resume_educations": self.resume.education_details, "resume_jobs": self.resume.experience_details,
"resume_projects": self.resume.projects, "question": question})
logger.debug(f"Raw output for numeric question: {output_str}")
try:
output = self.extract_number_from_string(output_str)
logger.debug(f"Extracted number: {output}")
except ValueError:
logger.warning(
f"Failed to extract number, using default experience: {default_experience}")
output = default_experience
return output
def extract_number_from_string(self, output_str):
logger.debug(f"Extracting number from string: {output_str}")
numbers = re.findall(r"\d+", output_str)
if numbers:
logger.debug(f"Numbers found: {numbers}")
return int(numbers[0])
else:
logger.error("No numbers found in the string")
raise ValueError("No numbers found in the string")
def answer_question_from_options(self, question: str, options: list[str]) -> str:
logger.debug(f"Answering question from options: {question}")
func_template = self._preprocess_template_string(
strings.options_template)
prompt = ChatPromptTemplate.from_template(func_template)
chain = prompt | self.llm_cheap | StrOutputParser()
output_str = chain.invoke(
{"resume": self.resume, "question": question, "options": options})
logger.debug(f"Raw output for options question: {output_str}")
best_option = self.find_best_match(output_str, options)
logger.debug(f"Best option determined: {best_option}")
return best_option
def resume_or_cover(self, phrase: str) -> str:
logger.debug(
f"Determining if phrase refers to resume or cover letter: {phrase}")
prompt_template = """
Given the following phrase, respond with only 'resume' if the phrase is about a resume, or 'cover' if it's about a cover letter.
If the phrase contains only one word 'upload', consider it as 'cover'.
If the phrase contains 'upload resume', consider it as 'resume'.
Do not provide any additional information or explanations.
phrase: {phrase}
"""
prompt = ChatPromptTemplate.from_template(prompt_template)
chain = prompt | self.llm_cheap | StrOutputParser()
response = chain.invoke({"phrase": phrase})
logger.debug(f"Response for resume_or_cover: {response}")
if "resume" in response:
return "resume"
elif "cover" in response:
return "cover"
else:
return "resume"

79
src/utils.py
View file

@ -1,41 +1,48 @@
import logging
import os
import random
import sys
import time
from selenium import webdriver
from loguru import logger
from app_config import MINIMUM_LOG_LEVEL
log_file = "app_log.log"
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(log_file, mode='a', encoding='utf-8'),
logging.StreamHandler()
],
force=True # This will reset the root logger's handlers and apply the new configuration
)
# TODO: REMOVE THE FOLLOWING BLOCK: No need as Loguru handles everything by default
# logging.basicConfig(
# level=logging.INFO,
# format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
# handlers=[
# logging.FileHandler(log_file, mode='a', encoding='utf-8'),
# logging.StreamHandler()
# ],
# force=True # This will reset the root logger's handlers and apply the new configuration
# )
logger = logging.getLogger(__name__)
file_handler = logging.FileHandler(log_file, mode='a', encoding='utf-8')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.setLevel(logging.INFO)
if MINIMUM_LOG_LEVEL in ["DEBUG", "TRACE", "INFO", "WARNING", "ERROR", "CRITICAL"]:
logger.remove()
logger.add(sys.stderr, level=MINIMUM_LOG_LEVEL)
else:
logger.warning(f"Invalid log level: {MINIMUM_LOG_LEVEL}. Defaulting to DEBUG.")
logger.remove()
logger.add(sys.stderr, level="DEBUG")
chromeProfilePath = os.path.join(os.getcwd(), "chrome_profile", "linkedin_profile")
def ensure_chrome_profile():
logger.debug("Ensuring Chrome profile exists at path: %s", chromeProfilePath)
logger.debug(f"Ensuring Chrome profile exists at path: {chromeProfilePath}")
profile_dir = os.path.dirname(chromeProfilePath)
if not os.path.exists(profile_dir):
os.makedirs(profile_dir)
logger.debug("Created directory for Chrome profile: %s", profile_dir)
logger.debug(f"Created directory for Chrome profile: {profile_dir}")
if not os.path.exists(chromeProfilePath):
os.makedirs(chromeProfilePath)
logger.debug("Created Chrome profile directory: %s", chromeProfilePath)
logger.debug(f"Created Chrome profile directory: {chromeProfilePath}")
return chromeProfilePath
@ -43,13 +50,12 @@ def is_scrollable(element):
scroll_height = element.get_attribute("scrollHeight")
client_height = element.get_attribute("clientHeight")
scrollable = int(scroll_height) > int(client_height)
logger.debug("Element scrollable check: scrollHeight=%s, clientHeight=%s, scrollable=%s", scroll_height,
client_height, scrollable)
logger.debug(f"Element scrollable check: scrollHeight={scroll_height}, clientHeight={client_height}, scrollable={scrollable}")
return scrollable
def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse=False):
logger.debug("Starting slow scroll: start=%d, end=%d, step=%d, reverse=%s", start, end, step, reverse)
logger.debug(f"Starting slow scroll: start={start}, end={end}, step={step}, reverse={reverse}")
if reverse:
start, end = end, start
@ -60,19 +66,17 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse
raise ValueError("Step cannot be zero.")
max_scroll_height = int(scrollable_element.get_attribute("scrollHeight"))
current_scroll_position = int(scrollable_element.get_attribute("scrollTop"))
logger.debug("Max scroll height of the element: %d", max_scroll_height)
logger.debug("Current scroll position: %d", current_scroll_position)
current_scroll_position = int(float(scrollable_element.get_attribute("scrollTop")))
logger.debug(f"Max scroll height of the element: {max_scroll_height}")
logger.debug(f"Current scroll position: {current_scroll_position}")
if reverse:
if current_scroll_position < start:
start = current_scroll_position
logger.debug("Adjusted start position for upward scroll: %d", start)
logger.debug(f"Adjusted start position for upward scroll: {start}")
else:
if end > max_scroll_height:
logger.warning("End value exceeds the scroll height. Adjusting end to %d", max_scroll_height)
logger.warning(f"End value exceeds the scroll height. Adjusting end to {max_scroll_height}")
end = max_scroll_height
script_scroll_to = "arguments[0].scrollTop = arguments[1];"
@ -81,12 +85,10 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse
if scrollable_element.is_displayed():
if not is_scrollable(scrollable_element):
logger.warning("The element is not scrollable.")
print("The element is not scrollable.")
return
if (step > 0 and start >= end) or (step < 0 and start <= end):
logger.warning("No scrolling will occur due to incorrect start/end values.")
print("No scrolling will occur due to incorrect start/end values.")
return
position = start
@ -94,15 +96,14 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse
while (step > 0 and position < end) or (step < 0 and position > end):
if position == previous_position:
# Avoid re-scrolling to the same position
logger.debug("Stopping scroll as position hasn't changed: %d", position)
logger.debug(f"Stopping scroll as position hasn't changed: {position}")
break
try:
driver.execute_script(script_scroll_to, scrollable_element, position)
logger.debug("Scrolled to position: %d", position)
logger.debug(f"Scrolled to position: {position}")
except Exception as e:
logger.error("Error during scrolling: %s", e)
print(f"Error during scrolling: {e}")
logger.error(f"Error during scrolling: {e}")
previous_position = position
position += step
@ -114,14 +115,12 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse
# Ensure the final scroll position is correct
driver.execute_script(script_scroll_to, scrollable_element, end)
logger.debug("Scrolled to final position: %d", end)
logger.debug(f"Scrolled to final position: {end}")
time.sleep(0.5)
else:
logger.warning("The element is not visible.")
print("The element is not visible.")
except Exception as e:
logger.error("Exception occurred during scrolling: %s", e)
print(f"Exception occurred: {e}")
logger.error(f"Exception occurred during scrolling: {e}")
def chrome_browser_options():
@ -159,7 +158,7 @@ def chrome_browser_options():
profile_dir = os.path.basename(chromeProfilePath)
options.add_argument('--user-data-dir=' + initial_path)
options.add_argument("--profile-directory=" + profile_dir)
logger.debug("Using Chrome profile directory: %s", chromeProfilePath)
logger.debug(f"Using Chrome profile directory: {chromeProfilePath}")
else:
options.add_argument("--incognito")
logger.debug("Using Chrome in incognito mode")

0
tests/__init__.py Normal file
View file

153
tests/test_job_application_profile.py Normal file
View file

@ -0,0 +1,153 @@
import pytest
from src.job_application_profile import JobApplicationProfile
@pytest.fixture
def valid_yaml():
"""Valid YAML string for initializing JobApplicationProfile."""
return """
self_identification:
gender: Male
pronouns: He/Him
veteran: No
disability: No
ethnicity: Asian
legal_authorization:
eu_work_authorization: "Yes"
us_work_authorization: "Yes"
requires_us_visa: "No"
legally_allowed_to_work_in_us: "Yes"
requires_us_sponsorship: "No"
requires_eu_visa: "No"
legally_allowed_to_work_in_eu: "Yes"
requires_eu_sponsorship: "No"
work_preferences:
remote_work: "Yes"
in_person_work: "No"
open_to_relocation: "Yes"
willing_to_complete_assessments: "Yes"
willing_to_undergo_drug_tests: "Yes"
willing_to_undergo_background_checks: "Yes"
availability:
notice_period: "2 weeks"
salary_expectations:
salary_range_usd: "80000-120000"
"""
@pytest.fixture
def missing_field_yaml():
"""YAML string missing a required field (self_identification)."""
return """
legal_authorization:
eu_work_authorization: "Yes"
us_work_authorization: "Yes"
requires_us_visa: "No"
legally_allowed_to_work_in_us: "Yes"
requires_us_sponsorship: "No"
requires_eu_visa: "No"
legally_allowed_to_work_in_eu: "Yes"
requires_eu_sponsorship: "No"
work_preferences:
remote_work: "Yes"
in_person_work: "No"
open_to_relocation: "Yes"
willing_to_complete_assessments: "Yes"
willing_to_undergo_drug_tests: "Yes"
willing_to_undergo_background_checks: "Yes"
availability:
notice_period: "2 weeks"
salary_expectations:
salary_range_usd: "80000-120000"
"""
@pytest.fixture
def invalid_type_yaml():
"""YAML string with an invalid type for a field."""
return """
self_identification:
gender: Male
pronouns: He/Him
veteran: No
disability: No
ethnicity: Asian
legal_authorization:
eu_work_authorization: "Yes"
us_work_authorization: "Yes"
requires_us_visa: "No"
legally_allowed_to_work_in_us: "Yes"
requires_us_sponsorship: "No"
requires_eu_visa: "No"
legally_allowed_to_work_in_eu: "Yes"
requires_eu_sponsorship: "No"
work_preferences:
remote_work: 12345 # Invalid type, expecting a string
in_person_work: "No"
open_to_relocation: "Yes"
willing_to_complete_assessments: "Yes"
willing_to_undergo_drug_tests: "Yes"
willing_to_undergo_background_checks: "Yes"
availability:
notice_period: "2 weeks"
salary_expectations:
salary_range_usd: "80000-120000"
"""
def test_initialize_with_valid_yaml(valid_yaml):
"""Test initializing JobApplicationProfile with valid YAML."""
profile = JobApplicationProfile(valid_yaml)
# Check that the profile fields are correctly initialized
assert profile.self_identification.gender == "Male"
assert profile.self_identification.pronouns == "He/Him"
assert profile.legal_authorization.eu_work_authorization == "Yes"
assert profile.work_preferences.remote_work == "Yes"
assert profile.availability.notice_period == "2 weeks"
assert profile.salary_expectations.salary_range_usd == "80000-120000"
def test_initialize_with_missing_field(missing_field_yaml):
"""Test initializing JobApplicationProfile with missing required fields."""
with pytest.raises(KeyError) as excinfo:
JobApplicationProfile(missing_field_yaml)
assert "self_identification" in str(excinfo.value)
def test_initialize_with_invalid_yaml():
"""Test initializing JobApplicationProfile with invalid YAML."""
invalid_yaml_str = """
self_identification:
gender: Male
pronouns: He/Him
veteran: No
disability: No
ethnicity: Asian
legal_authorization:
eu_work_authorization: "Yes"
us_work_authorization: "Yes"
requires_us_visa: "No"
legally_allowed_to_work_in_us: "Yes"
requires_us_sponsorship: "No"
requires_eu_visa: "No"
legally_allowed_to_work_in_eu: "Yes"
requires_eu_sponsorship: "No"
work_preferences:
remote_work: "Yes"
in_person_work: "No"
availability:
notice_period: "2 weeks"
salary_expectations:
salary_range_usd: "80000-120000"
""" # Missing fields in work_preferences
with pytest.raises(TypeError):
JobApplicationProfile(invalid_yaml_str)
def test_str_representation(valid_yaml):
"""Test the string representation of JobApplicationProfile."""
profile = JobApplicationProfile(valid_yaml)
profile_str = str(profile)
assert "Self Identification:" in profile_str
assert "Legal Authorization:" in profile_str
assert "Work Preferences:" in profile_str
assert "Availability:" in profile_str
assert "Salary Expectations:" in profile_str
assert "Male" in profile_str
assert "80000-120000" in profile_str

158
tests/test_linkedIn_authenticator.py Normal file
View file

@ -0,0 +1,158 @@
import pytest
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from src.linkedIn_authenticator import LinkedInAuthenticator
from selenium.common.exceptions import NoSuchElementException, TimeoutException
@pytest.fixture
def mock_driver(mocker):
"""Fixture to mock the Selenium WebDriver."""
return mocker.Mock()
@pytest.fixture
def authenticator(mock_driver):
"""Fixture to initialize LinkedInAuthenticator with a mocked driver."""
return LinkedInAuthenticator(mock_driver)
def test_set_secrets(authenticator):
"""Test setting secrets (email, password)."""
authenticator.set_secrets("test@example.com", "password123")
assert authenticator.email == "test@example.com"
assert authenticator.password == "password123"
def test_start_logged_in(mocker, authenticator):
"""Test starting LinkedIn when already logged in."""
mocker.patch.object(authenticator, 'is_logged_in', return_value=True)
mocker.patch.object(authenticator.driver, 'get')
mocker.patch("time.sleep") # Avoid waiting during the test
authenticator.start()
authenticator.driver.get.assert_called_with('https://www.linkedin.com/feed')
authenticator.is_logged_in.assert_called_once()
assert authenticator.driver.get.call_count == 1
def test_start_not_logged_in(mocker, authenticator):
"""Test starting LinkedIn when not logged in."""
mocker.patch.object(authenticator, 'is_logged_in', return_value=False)
mocker.patch.object(authenticator, 'handle_login')
mocker.patch.object(authenticator.driver, 'get')
mocker.patch("time.sleep")
authenticator.start()
authenticator.driver.get.assert_called_with('https://www.linkedin.com/feed')
authenticator.handle_login.assert_called_once()
def test_handle_login(mocker, authenticator):
"""Test handling the LinkedIn login process."""
mocker.patch.object(authenticator.driver, 'get')
mocker.patch.object(authenticator, 'enter_credentials')
mocker.patch.object(authenticator, 'submit_login_form')
mocker.patch.object(authenticator, 'handle_security_check')
# Mock current_url as a regular return value, not PropertyMock
mocker.patch.object(authenticator.driver, 'current_url', return_value='https://www.linkedin.com/login')
authenticator.handle_login()
authenticator.driver.get.assert_called_with('https://www.linkedin.com/login')
authenticator.enter_credentials.assert_called_once()
authenticator.submit_login_form.assert_called_once()
authenticator.handle_security_check.assert_called_once()
def test_enter_credentials_success(mocker, authenticator):
"""Test entering credentials."""
email_mock = mocker.Mock()
password_mock = mocker.Mock()
mocker.patch.object(WebDriverWait, 'until', return_value=email_mock)
mocker.patch.object(authenticator.driver, 'find_element', return_value=password_mock)
authenticator.set_secrets("test@example.com", "password123")
authenticator.enter_credentials()
email_mock.send_keys.assert_called_once_with("test@example.com")
password_mock.send_keys.assert_called_once_with("password123")
def test_enter_credentials_timeout(mocker, authenticator):
"""Test entering credentials with a TimeoutException."""
mocker.patch.object(WebDriverWait, 'until', side_effect=TimeoutException)
authenticator.set_secrets("test@example.com", "password123")
authenticator.enter_credentials()
authenticator.driver.find_element.assert_not_called() # Password input should not be accessed if email fails
def test_submit_login_form_success(mocker, authenticator):
"""Test submitting the login form."""
login_button_mock = mocker.Mock()
mocker.patch.object(authenticator.driver, 'find_element', return_value=login_button_mock)
authenticator.submit_login_form()
login_button_mock.click.assert_called_once()
def test_submit_login_form_no_button(mocker, authenticator):
"""Test submitting the login form when the login button is not found."""
mocker.patch.object(authenticator.driver, 'find_element', side_effect=NoSuchElementException)
authenticator.submit_login_form()
authenticator.driver.find_element.assert_called_once_with(By.XPATH, '//button[@type="submit"]')
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_handle_security_check_success(mocker, authenticator):
"""Test handling security check successfully."""
mocker.patch.object(WebDriverWait, 'until', side_effect=[
mocker.Mock(), # Security checkpoint detection
mocker.Mock() # Security check completion
])
authenticator.handle_security_check()
# Verify WebDriverWait is called with EC.url_contains for both the challenge and feed
WebDriverWait(authenticator.driver, 10).until.assert_any_call(mocker.ANY)
WebDriverWait(authenticator.driver, 300).until.assert_any_call(mocker.ANY)
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()
# Verify WebDriverWait is called with EC.url_contains for the challenge
WebDriverWait(authenticator.driver, 10).until.assert_any_call(mocker.ANY)

14
tests/test_linkedIn_bot_facade.py Normal file
View file

@ -0,0 +1,14 @@
import pytest
# from src.linkedIn_job_manager import JobManager
@pytest.fixture
def job_manager():
"""Fixture for JobManager."""
return None # Replace with valid instance or mock later
def test_bot_functionality(job_manager):
"""Test LinkedIn bot facade."""
# Example: test job manager interacts with the bot facade correctly
job = {"title": "Software Engineer"}
# job_manager.some_method_to_apply(job)
assert job is not None # Placeholder for actual test

97
tests/test_linkedIn_easy_applier.py Normal file
View file

@ -0,0 +1,97 @@
import pytest
from unittest import mock
from src.linkedIn_easy_applier import LinkedInEasyApplier
@pytest.fixture
def mock_driver():
"""Fixture to mock Selenium WebDriver."""
return mock.Mock()
@pytest.fixture
def mock_gpt_answerer():
"""Fixture to mock GPT Answerer."""
return mock.Mock()
@pytest.fixture
def mock_resume_generator_manager():
"""Fixture to mock Resume Generator Manager."""
return mock.Mock()
@pytest.fixture
def easy_applier(mock_driver, mock_gpt_answerer, mock_resume_generator_manager):
"""Fixture to initialize LinkedInEasyApplier with mocks."""
return LinkedInEasyApplier(
driver=mock_driver,
resume_dir="/path/to/resume",
set_old_answers=[('Question 1', 'Answer 1', 'Type 1')],
gpt_answerer=mock_gpt_answerer,
resume_generator_manager=mock_resume_generator_manager
)
def test_initialization(mocker, easy_applier):
"""Test that LinkedInEasyApplier is initialized correctly."""
# Mock os.path.exists to return True
mocker.patch('os.path.exists', return_value=True)
easy_applier = LinkedInEasyApplier(
driver=mocker.Mock(),
resume_dir="/path/to/resume",
set_old_answers=[('Question 1', 'Answer 1', 'Type 1')],
gpt_answerer=mocker.Mock(),
resume_generator_manager=mocker.Mock()
)
assert easy_applier.resume_path == "/path/to/resume"
assert len(easy_applier.set_old_answers) == 1
assert easy_applier.gpt_answerer is not None
assert easy_applier.resume_generator_manager is not None
def test_apply_to_job_success(mocker, easy_applier):
"""Test successfully applying to a job."""
mock_job = mock.Mock()
# Mock job_apply so we don't actually try to apply
mocker.patch.object(easy_applier, 'job_apply')
easy_applier.apply_to_job(mock_job)
easy_applier.job_apply.assert_called_once_with(mock_job)
def test_apply_to_job_failure(mocker, easy_applier):
"""Test failure while applying to a job."""
mock_job = mock.Mock()
mocker.patch.object(easy_applier, 'job_apply',
side_effect=Exception("Test error"))
with pytest.raises(Exception, match="Test error"):
easy_applier.apply_to_job(mock_job)
easy_applier.job_apply.assert_called_once_with(mock_job)
def test_check_for_premium_redirect_no_redirect(mocker, easy_applier):
"""Test that check_for_premium_redirect works when there's no redirect."""
mock_job = mock.Mock()
easy_applier.driver.current_url = "https://www.linkedin.com/jobs/view/1234"
easy_applier.check_for_premium_redirect(mock_job)
easy_applier.driver.get.assert_not_called()
def test_check_for_premium_redirect_with_redirect(mocker, easy_applier):
"""Test that check_for_premium_redirect handles LinkedIn Premium redirects."""
mock_job = mock.Mock()
easy_applier.driver.current_url = "https://www.linkedin.com/premium"
mock_job.link = "https://www.linkedin.com/jobs/view/1234"
with pytest.raises(Exception, match="Redirected to LinkedIn Premium page and failed to return"):
easy_applier.check_for_premium_redirect(mock_job)
# Verify that it attempted to return to the job page 3 times
assert easy_applier.driver.get.call_count == 3

168
tests/test_linkedIn_job_manager.py Normal file
View file

@ -0,0 +1,168 @@
from src.job import Job
from unittest import mock
from pathlib import Path
import os
import pytest
from src.linkedIn_job_manager import LinkedInJobManager
from selenium.common.exceptions import NoSuchElementException
from loguru import logger
@pytest.fixture
def job_manager(mocker):
"""Fixture to create a LinkedInJobManager instance with mocked driver."""
mock_driver = mocker.Mock()
return LinkedInJobManager(mock_driver)
def test_initialization(job_manager):
"""Test LinkedInJobManager initialization."""
assert job_manager.driver is not None
assert job_manager.set_old_answers == set()
assert job_manager.easy_applier_component is None
def test_set_parameters(mocker, job_manager):
"""Test setting parameters for the LinkedInJobManager."""
# Mocking os.path.exists to return True for the resume path
mocker.patch('pathlib.Path.exists', return_value=True)
params = {
'company_blacklist': ['Company A', 'Company B'],
'title_blacklist': ['Intern', 'Junior'],
'positions': ['Software Engineer', 'Data Scientist'],
'locations': ['New York', 'San Francisco'],
'apply_once_at_company': True,
'uploads': {'resume': '/path/to/resume'}, # Resume path provided here
'outputFileDirectory': '/path/to/output',
'job_applicants_threshold': {
'min_applicants': 5,
'max_applicants': 50
},
'remote': False,
'distance': 50,
'date': {'all time': True}
}
job_manager.set_parameters(params)
# Normalize paths to handle platform differences (e.g., Windows vs Unix-like systems)
assert str(job_manager.resume_path) == os.path.normpath('/path/to/resume')
assert str(job_manager.output_file_directory) == os.path.normpath(
'/path/to/output')
def next_job_page(self, position, location, job_page):
logger.debug(f"Navigating to next job page: {position} in {location}, page {job_page}")
self.driver.get(
f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}&location={location}&start={job_page * 25}")
def test_get_jobs_from_page_no_jobs(mocker, job_manager):
"""Test get_jobs_from_page when no jobs are found."""
mocker.patch.object(job_manager.driver, 'find_element',
side_effect=NoSuchElementException)
jobs = job_manager.get_jobs_from_page()
assert jobs == []
def test_get_jobs_from_page_with_jobs(mocker, job_manager):
"""Test get_jobs_from_page when job elements are found."""
# Mock the no_jobs_element to behave correctly
mock_no_jobs_element = mocker.Mock()
mock_no_jobs_element.text = "No matching jobs found"
# Mocking the find_element to return the mock no_jobs_element
mocker.patch.object(job_manager.driver, 'find_element',
return_value=mock_no_jobs_element)
# Mock the page_source
mocker.patch.object(job_manager.driver, 'page_source',
return_value="some page content")
# Ensure jobs are returned as empty list due to "No matching jobs found"
jobs = job_manager.get_jobs_from_page()
assert jobs == [] # No jobs expected due to "No matching jobs found"
def test_apply_jobs_with_no_jobs(mocker, job_manager):
"""Test apply_jobs when no jobs are found."""
# Mocking find_element to return a mock element that simulates no jobs
mock_element = mocker.Mock()
mock_element.text = "No matching jobs found"
# Mock the driver to simulate the page source
mocker.patch.object(job_manager.driver, 'page_source', return_value="")
# Mock the driver to return the mock element when find_element is called
mocker.patch.object(job_manager.driver, 'find_element',
return_value=mock_element)
# Call apply_jobs and ensure no exceptions are raised
job_manager.apply_jobs()
# Ensure it attempted to find the job results list
assert job_manager.driver.find_element.call_count == 1
def test_apply_jobs_with_jobs(mocker, job_manager):
"""Test apply_jobs when jobs are present."""
# Mock no_jobs_element to simulate the absence of "No matching jobs found" banner
no_jobs_element = mocker.Mock()
no_jobs_element.text = "" # Empty text means "No matching jobs found" is not present
mocker.patch.object(job_manager.driver, 'find_element',
return_value=no_jobs_element)
# Mock the page_source to simulate what the page looks like when jobs are present
mocker.patch.object(job_manager.driver, 'page_source',
return_value="some job content")
# Mock the outer find_elements (scaffold-layout__list-container)
container_mock = mocker.Mock()
# Mock the inner find_elements to return job list items
job_element_mock = mocker.Mock()
# Simulating two job items
job_elements_list = [job_element_mock, job_element_mock]
# Return the container mock, which itself returns the job elements list
container_mock.find_elements.return_value = job_elements_list
mocker.patch.object(job_manager.driver, 'find_elements',
return_value=[container_mock])
# Mock the extract_job_information_from_tile method to return sample job info
mocker.patch.object(job_manager, 'extract_job_information_from_tile', return_value=(
"Title", "Company", "Location", "Apply", "Link"))
# Mock other methods like is_blacklisted, is_already_applied_to_job, and is_already_applied_to_company
mocker.patch.object(job_manager, 'is_blacklisted', return_value=False)
mocker.patch.object(
job_manager, 'is_already_applied_to_job', return_value=False)
mocker.patch.object(
job_manager, 'is_already_applied_to_company', return_value=False)
# Mock the LinkedInEasyApplier component
job_manager.easy_applier_component = mocker.Mock()
# Mock the output_file_directory as a valid Path object
job_manager.output_file_directory = Path("/mocked/path/to/output")
# Mock Path.exists() to always return True (so no actual file system interaction is needed)
mocker.patch.object(Path, 'exists', return_value=True)
# Mock the open function to prevent actual file writing
mock_open = mocker.mock_open()
mocker.patch('builtins.open', mock_open)
# Run the apply_jobs method
job_manager.apply_jobs()
# Assertions
assert job_manager.driver.find_elements.call_count == 1
# Called for each job element
assert job_manager.extract_job_information_from_tile.call_count == 2
# Called for each job element
assert job_manager.easy_applier_component.job_apply.call_count == 2
mock_open.assert_called() # Ensure that the open function was called

96
tests/test_utils.py Normal file
View file

@ -0,0 +1,96 @@
# tests/test_utils.py
import pytest
import os
import time
from unittest import mock
from selenium.webdriver.remote.webelement import WebElement
from src.utils import ensure_chrome_profile, is_scrollable, scroll_slow, chrome_browser_options, printred, printyellow
# Mocking logging to avoid actual file writing
@pytest.fixture(autouse=True)
def mock_logger(mocker):
mocker.patch("src.utils.logger")
# Test ensure_chrome_profile function
def test_ensure_chrome_profile(mocker):
mocker.patch("os.path.exists", return_value=False) # Pretend directory doesn't exist
mocker.patch("os.makedirs") # Mock making directories
# Call the function
profile_path = ensure_chrome_profile()
# Verify that os.makedirs was called twice to create the directory
assert profile_path.endswith("linkedin_profile")
assert os.path.exists.called
assert os.makedirs.called
# Test is_scrollable function
def test_is_scrollable(mocker):
mock_element = mocker.Mock(spec=WebElement)
mock_element.get_attribute.side_effect = lambda attr: "1000" if attr == "scrollHeight" else "500"
# Call the function
scrollable = is_scrollable(mock_element)
# Check the expected outcome
assert scrollable is True
mock_element.get_attribute.assert_any_call("scrollHeight")
mock_element.get_attribute.assert_any_call("clientHeight")
# Test scroll_slow function
def test_scroll_slow(mocker):
mock_driver = mocker.Mock()
mock_element = mocker.Mock(spec=WebElement)
# Mock element's attributes for scrolling
mock_element.get_attribute.side_effect = lambda attr: "2000" if attr == "scrollHeight" else "0"
mock_element.is_displayed.return_value = True
mocker.patch("time.sleep") # Mock time.sleep to avoid waiting
# Call the function
scroll_slow(mock_driver, mock_element, start=0, end=1000, step=100, reverse=False)
# Ensure that scrolling happened multiple times
assert mock_driver.execute_script.called
mock_element.is_displayed.assert_called_once()
def test_scroll_slow_element_not_scrollable(mocker):
mock_driver = mocker.Mock()
mock_element = mocker.Mock(spec=WebElement)
# Mock the attributes so the element is not scrollable
mock_element.get_attribute.side_effect = lambda attr: "1000" if attr == "scrollHeight" else "1000"
mock_element.is_displayed.return_value = True
scroll_slow(mock_driver, mock_element, start=0, end=1000, step=100)
# Ensure it detected non-scrollable element
mock_driver.execute_script.assert_not_called()
# Test chrome_browser_options function
def test_chrome_browser_options(mocker):
mocker.patch("src.utils.ensure_chrome_profile")
mocker.patch("os.path.dirname", return_value="/mocked/path")
mocker.patch("os.path.basename", return_value="profile_directory")
mock_options = mocker.Mock()
mocker.patch("selenium.webdriver.ChromeOptions", return_value=mock_options)
# Call the function
options = chrome_browser_options()
# Ensure options were set
assert mock_options.add_argument.called
assert options == mock_options
# Test printred and printyellow functions
def test_printred(mocker):
mocker.patch("builtins.print")
printred("Test")
print.assert_called_once_with("\033[91mTest\033[0m")
def test_printyellow(mocker):
mocker.patch("builtins.print")
printyellow("Test")
print.assert_called_once_with("\033[93mTest\033[0m")