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

Loguru Integration: Better logs

Browse source
This commit is contained in:
tapas-joshi 2024-09-12 21:30:35 -04:00
parent c27bf2c715
commit f77706d579
14 changed files with 851 additions and 374 deletions
Download patch file Download diff file Expand all files Collapse all files

1
app_config.py Normal file
View file

@ -0,0 +1 @@
MINIMUM_LOG_LEVEL="DEBUG"

25
main.py
View file

@ -7,14 +7,15 @@ import click
from selenium import webdriver from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService from selenium.webdriver.chrome.service import Service as ChromeService
from webdriver_manager.chrome import ChromeDriverManager 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 lib_resume_builder_AIHawk import Resume,StyleManager,FacadeManager,ResumeGenerator
from src.utils import chrome_browser_options 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_authenticator import LinkedInAuthenticator
from src.linkedIn_bot_facade import LinkedInBotFacade from src.linkedIn_bot_facade import LinkedInBotFacade
from src.linkedIn_job_manager import LinkedInJobManager from src.linkedIn_job_manager import LinkedInJobManager
from src.job_application_profile import JobApplicationProfile from src.job_application_profile import JobApplicationProfile
from loguru import logger
# Suppress stderr # Suppress stderr
sys.stderr = open(os.devnull, 'w') 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_login()
bot.start_apply() bot.start_apply()
except WebDriverException as e: except WebDriverException as e:
print(f"WebDriver error occurred: {e}") logger.error(f"WebDriver error occurred: {e}")
except Exception as e: except Exception as e:
raise RuntimeError(f"Error running the bot: {str(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) create_and_run_bot(email, password, parameters, llm_api_key)
except ConfigError as ce: except ConfigError as ce:
print(f"Configuration error: {str(ce)}") logger.error(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"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: except FileNotFoundError as fnf:
print(f"File not found: {str(fnf)}") logger.error(f"File not found: {str(fnf)}")
print("Ensure all required files are present in the data folder.") logger.error("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("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: 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: except Exception as e:
print(f"An unexpected error occurred: {str(e)}") logger.error(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("Refer to the general troubleshooting guide: https://github.com/feder-cr/LinkedIn_AIHawk_automatic_job_application/blob/main/readme.md#configuration")
if __name__ == "__main__": if __name__ == "__main__":
main() main()

1
requirements.txt
View file

@ -25,3 +25,4 @@ httpx~=0.27.2
python-dotenv~=1.0.1 python-dotenv~=1.0.1
PyYAML~=6.0.2 PyYAML~=6.0.2
pytest>=8.3.3 pytest>=8.3.3
loguru==0.7.2

15
resume_yaml_generator.py
View file

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

12
src/job.py
View file

@ -1,6 +1,6 @@
from dataclasses import dataclass from dataclasses import dataclass
from src.utils import logger from loguru import logger
@dataclass @dataclass
@ -16,22 +16,22 @@ class Job:
recruiter_link: str = "" recruiter_link: str = ""
def set_summarize_job_description(self, summarize_job_description): 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 self.summarize_job_description = summarize_job_description
def set_job_description(self, description): def set_job_description(self, description):
logger.debug("Setting job description: %s", description) logger.debug(f"Setting job description: {description}")
self.description = description self.description = description
def set_recruiter_link(self, recruiter_link): 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 self.recruiter_link = recruiter_link
def formatted_job_information(self): def formatted_job_information(self):
""" """
Formats the job information as a markdown string. 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_information = f"""
# Job Description # Job Description
## Job Information ## Job Information
@ -44,5 +44,5 @@ class Job:
{self.description or 'No description provided.'} {self.description or 'No description provided.'}
""" """
formatted_information = job_information.strip() formatted_information = job_information.strip()
logger.debug("Formatted job information: %s", formatted_information) logger.debug(f"Formatted job information: {formatted_information}")
return formatted_information return formatted_information

62
src/job_application_profile.py
View file

@ -2,7 +2,7 @@ from dataclasses import dataclass
import yaml import yaml
from src.utils import logger from loguru import logger
@dataclass @dataclass
@ -58,106 +58,106 @@ class JobApplicationProfile:
logger.debug("Initializing JobApplicationProfile with provided YAML string") logger.debug("Initializing JobApplicationProfile with provided YAML string")
try: try:
data = yaml.safe_load(yaml_str) 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: 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 raise ValueError("Error parsing YAML file.") from e
except Exception as 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 raise RuntimeError("An unexpected error occurred while parsing the YAML file.") from e
if not isinstance(data, dict): 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.") raise TypeError("YAML data must be a dictionary.")
# Process self_identification # Process self_identification
try: try:
logger.debug("Processing self_identification") logger.debug("Processing self_identification")
self.self_identification = SelfIdentification(**data['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: 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 raise KeyError(f"Required field {e} is missing in self_identification data.") from e
except TypeError as 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 raise TypeError(f"Error in self_identification data: {e}") from e
except AttributeError as 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 raise AttributeError("Attribute error in self_identification processing.") from e
except Exception as 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 raise RuntimeError("An unexpected error occurred while processing self_identification.") from e
# Process legal_authorization # Process legal_authorization
try: try:
logger.debug("Processing legal_authorization") logger.debug("Processing legal_authorization")
self.legal_authorization = LegalAuthorization(**data['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: 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 raise KeyError(f"Required field {e} is missing in legal_authorization data.") from e
except TypeError as 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 raise TypeError(f"Error in legal_authorization data: {e}") from e
except AttributeError as 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 raise AttributeError("Attribute error in legal_authorization processing.") from e
except Exception as 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 raise RuntimeError("An unexpected error occurred while processing legal_authorization.") from e
# Process work_preferences # Process work_preferences
try: try:
logger.debug("Processing work_preferences") logger.debug("Processing work_preferences")
self.work_preferences = WorkPreferences(**data['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: 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 raise KeyError(f"Required field {e} is missing in work_preferences data.") from e
except TypeError as 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 raise TypeError(f"Error in work_preferences data: {e}") from e
except AttributeError as 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 raise AttributeError("Attribute error in work_preferences processing.") from e
except Exception as 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 raise RuntimeError("An unexpected error occurred while processing work_preferences.") from e
# Process availability # Process availability
try: try:
logger.debug("Processing availability") logger.debug("Processing availability")
self.availability = Availability(**data['availability']) self.availability = Availability(**data['availability'])
logger.debug("availability processed: %s", self.availability) logger.debug(f"Availability processed: {self.availability}")
except KeyError as e: 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 raise KeyError(f"Required field {e} is missing in availability data.") from e
except TypeError as 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 raise TypeError(f"Error in availability data: {e}") from e
except AttributeError as 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 raise AttributeError("Attribute error in availability processing.") from e
except Exception as 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 raise RuntimeError("An unexpected error occurred while processing availability.") from e
# Process salary_expectations # Process salary_expectations
try: try:
logger.debug("Processing salary_expectations") logger.debug("Processing salary_expectations")
self.salary_expectations = SalaryExpectations(**data['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: 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 raise KeyError(f"Required field {e} is missing in salary_expectations data.") from e
except TypeError as 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 raise TypeError(f"Error in salary_expectations data: {e}") from e
except AttributeError as 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 raise AttributeError("Attribute error in salary_expectations processing.") from e
except Exception as 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 raise RuntimeError("An unexpected error occurred while processing salary_expectations.") from e
logger.debug("JobApplicationProfile initialization completed successfully.") logger.debug("JobApplicationProfile initialization completed successfully.")
@ -173,5 +173,5 @@ class JobApplicationProfile:
f"Work Preferences:\n{format_dataclass(self.work_preferences)}\n\n" f"Work Preferences:\n{format_dataclass(self.work_preferences)}\n\n"
f"Availability: {self.availability.notice_period}\n\n" f"Availability: {self.availability.notice_period}\n\n"
f"Salary Expectations: {self.salary_expectations.salary_range_usd}\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 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 import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support.ui import WebDriverWait
from src.utils import logger from loguru import logger
class LinkedInAuthenticator: class LinkedInAuthenticator:
@ -15,12 +15,12 @@ class LinkedInAuthenticator:
self.driver = driver self.driver = driver
self.email = "" self.email = ""
self.password = "" self.password = ""
logger.debug("LinkedInAuthenticator initialized with driver: %s", driver) logger.debug(f"LinkedInAuthenticator initialized with driver: {driver}")
def set_secrets(self, email, password): def set_secrets(self, email, password):
self.email = email self.email = email
self.password = password self.password = password
logger.debug("Secrets set with email: %s", email) logger.debug(f"Secrets set with email: {email}")
def start(self): def start(self):
logger.info("Starting Chrome browser to log in to LinkedIn.") logger.info("Starting Chrome browser to log in to LinkedIn.")
@ -40,13 +40,13 @@ class LinkedInAuthenticator:
logger.info("Navigating to the LinkedIn login page...") logger.info("Navigating to the LinkedIn login page...")
self.driver.get("https://www.linkedin.com/login") self.driver.get("https://www.linkedin.com/login")
if 'feed' in self.driver.current_url: if 'feed' in self.driver.current_url:
print("User is already logged in.") logger.debug("User is already logged in.")
return return
try: try:
self.enter_credentials() self.enter_credentials()
self.submit_login_form() self.submit_login_form()
except NoSuchElementException as e: 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)) time.sleep(random.uniform(3, 5))
self.handle_security_check() self.handle_security_check()
@ -57,13 +57,13 @@ class LinkedInAuthenticator:
EC.presence_of_element_located((By.ID, "username")) EC.presence_of_element_located((By.ID, "username"))
) )
email_field.send_keys(self.email) 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 = self.driver.find_element(By.ID, "password")
password_field.send_keys(self.password) password_field.send_keys(self.password)
logger.debug("Password entered.") logger.debug("Password entered.")
except TimeoutException: except TimeoutException:
logger.error("Login form not found. Aborting login.") 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): def submit_login_form(self):
try: try:
@ -73,7 +73,6 @@ class LinkedInAuthenticator:
logger.debug("Login form submitted.") logger.debug("Login form submitted.")
except NoSuchElementException: except NoSuchElementException:
logger.error("Login button not found. Please verify the page structure.") 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): def handle_security_check(self):
try: try:
@ -82,22 +81,19 @@ class LinkedInAuthenticator:
EC.url_contains('https://www.linkedin.com/checkpoint/challengesV2/') EC.url_contains('https://www.linkedin.com/checkpoint/challengesV2/')
) )
logger.warning("Security checkpoint detected. Please complete the challenge.") logger.warning("Security checkpoint detected. Please complete the challenge.")
print("Security checkpoint detected. Please complete the challenge.")
WebDriverWait(self.driver, 300).until( WebDriverWait(self.driver, 300).until(
EC.url_contains('https://www.linkedin.com/feed/') EC.url_contains('https://www.linkedin.com/feed/')
) )
logger.info("Security check completed") logger.info("Security check completed")
print("Security check completed")
except TimeoutException: except TimeoutException:
logger.error("Security check not completed within the timeout.") logger.error("Security check not completed. Please try again later.")
print("Security check not completed. Please try again later.")
def is_logged_in(self): def is_logged_in(self):
# target_url = 'https://www.linkedin.com/feed' # target_url = 'https://www.linkedin.com/feed'
# #
# # Navigate to the target URL if not already there # # Navigate to the target URL if not already there
# if self.driver.current_url != target_url: # 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) # self.driver.get(target_url)
try: try:
@ -109,10 +105,10 @@ class LinkedInAuthenticator:
# Check for the presence of the "Start a post" button # Check for the presence of the "Start a post" button
buttons = self.driver.find_elements(By.CLASS_NAME, 'share-box-feed-entry__trigger') 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): 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): 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.") 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): def wait_for_page_load(self, timeout=10):
try: 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( WebDriverWait(self.driver, timeout).until(
lambda d: d.execute_script('return document.readyState') == 'complete' lambda d: d.execute_script('return document.readyState') == 'complete'
) )
logger.debug("Page load completed.") logger.debug("Page load completed.")
except TimeoutException: except TimeoutException:
logger.error("Page load timed out.") 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: class LinkedInBotState:
@ -16,10 +16,10 @@ class LinkedInBotState:
self.logged_in = False self.logged_in = False
def validate_state(self, required_keys): 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: for key in required_keys:
if not getattr(self, key): 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.") raise ValueError(f"{key.replace('_', ' ').capitalize()} must be set before proceeding.")
logger.debug("State validation passed") logger.debug("State validation passed")
@ -87,11 +87,11 @@ class LinkedInBotFacade:
logger.debug("Apply process started successfully") logger.debug("Apply process started successfully")
def _validate_non_empty(self, value, name): 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: 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.") 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): def _ensure_job_profile_and_resume_set(self):
logger.debug("Ensuring job profile and resume are set") logger.debug("Ensuring job profile and resume are set")

203
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 from selenium.webdriver.support.ui import Select, WebDriverWait
import src.utils as utils import src.utils as utils
from src.utils import logger from loguru import logger
class LinkedInEasyApplier: class LinkedInEasyApplier:
@ -39,7 +39,7 @@ class LinkedInEasyApplier:
def _load_questions_from_json(self) -> List[dict]: def _load_questions_from_json(self) -> List[dict]:
output_file = 'answers.json' 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: try:
with open(output_file, 'r') as f: with open(output_file, 'r') as f:
try: try:
@ -56,7 +56,7 @@ class LinkedInEasyApplier:
return [] return []
except Exception: except Exception:
tb_str = traceback.format_exc() 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}") 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): def check_for_premium_redirect(self, job: Any, max_attempts=3):
@ -73,7 +73,7 @@ class LinkedInEasyApplier:
current_url = self.driver.current_url current_url = self.driver.current_url
if "linkedin.com/premium" in 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( raise Exception(
f"Redirected to LinkedIn Premium page and failed to return after {max_attempts} attempts. Job application aborted.") f"Redirected to LinkedIn Premium page and failed to return after {max_attempts} attempts. Job application aborted.")
@ -92,13 +92,13 @@ class LinkedInEasyApplier:
raise e raise e
def job_apply(self, job: Any): 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: try:
self.driver.get(job.link) 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: 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 raise
time.sleep(random.uniform(3, 5)) time.sleep(random.uniform(3, 5))
@ -118,39 +118,29 @@ class LinkedInEasyApplier:
logger.debug("Retrieving job description") logger.debug("Retrieving job description")
job_description = self._get_job_description() job_description = self._get_job_description()
job.set_job_description(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") logger.debug("Retrieving recruiter link")
recruiter_link = self._get_job_recruiter() recruiter_link = self._get_job_recruiter()
job.set_recruiter_link(recruiter_link) job.set_recruiter_link(recruiter_link)
logger.debug("Recruiter link set: %s", recruiter_link) logger.debug(f"Recruiter link set: {recruiter_link}")
# Try clicking the "Easy Apply" button logger.debug("Attempting to click 'Easy Apply' button")
try:
logger.debug("Attempting to click 'Easy Apply' button using ActionChains")
actions = ActionChains(self.driver) actions = ActionChains(self.driver)
actions.move_to_element(easy_apply_button).click().perform() actions.move_to_element(easy_apply_button).click().perform()
logger.debug("'Easy Apply' button clicked successfully") logger.debug("'Easy Apply' button clicked successfully")
except Exception as e:
logger.warning(f"Failed to click 'Easy Apply' button using ActionChains: {e}, trying JavaScript click")
try:
self.driver.execute_script("arguments[0].click();", easy_apply_button)
logger.debug("'Easy Apply' button clicked successfully via JavaScript")
except Exception as js_error:
logger.error(f"Failed to click 'Easy Apply' button via JavaScript: {js_error}")
raise
logger.debug("Passing job information to GPT Answerer") logger.debug("Passing job information to GPT Answerer")
self.gpt_answerer.set_job(job) self.gpt_answerer.set_job(job)
logger.debug("Filling out application form") logger.debug("Filling out application form")
self._fill_application_form(job) 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: except Exception as e:
tb_str = traceback.format_exc() 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") logger.debug("Discarding application due to failure")
self._discard_application() self._discard_application()
@ -160,112 +150,73 @@ class LinkedInEasyApplier:
def _find_easy_apply_button(self, job: Any) -> WebElement: def _find_easy_apply_button(self, job: Any) -> WebElement:
logger.debug("Searching for 'Easy Apply' button") logger.debug("Searching for 'Easy Apply' button")
attempt = 0 attempt = 0
timeout = 8
search_methods = [ search_methods = [
{
'description': "'aria-label' containing 'Easy Apply to' and with data-job-id attribute",
'xpath': '//button[contains(@aria-label, "Easy Apply to") and contains(@data-job-id, "")]'
},
{ {
'description': "find all 'Easy Apply' buttons using find_elements", 'description': "find all 'Easy Apply' buttons using find_elements",
'find_elements': True, 'find_elements': True,
'xpath': '//button[contains(@class, "jobs-apply-button") and contains(., "Easy Apply") and contains(@data-job-id, "")]' 'xpath': '//button[contains(@class, "jobs-apply-button") and contains(., "Easy Apply")]'
}, },
{ {
'description': "button text search with data-job-id attribute", 'description': "'aria-label' containing 'Easy Apply to'",
'xpath': '//button[contains(text(), "Easy Apply") or contains(text(), "Apply now") and contains(@data-job-id, "")]' 'xpath': '//button[contains(@aria-label, "Easy Apply to")]'
},
{
'description': "button text search",
'xpath': '//button[contains(text(), "Easy Apply") or contains(text(), "Apply now")]'
} }
] ]
while attempt < 3: while attempt < 2:
self.check_for_premium_redirect(job) self.check_for_premium_redirect(job)
self._scroll_page() self._scroll_page()
try:
logger.info("Removing focus from the active element")
self.driver.execute_script("document.activeElement.blur();")
time.sleep(1)
logger.info("Clicking on body to reset focus via JavaScript")
try:
self.driver.execute_script("document.querySelector('body').focus();")
except Exception as e:
logger.warning(f"Failed to reset focus via body: {e}")
time.sleep(1)
logger.info("Clicking on html to reset focus via JavaScript")
try:
self.driver.execute_script("document.querySelector('html').focus();")
except Exception as e:
logger.warning(f"Failed to reset focus via html: {e}")
except Exception as e:
logger.warning(f"Failed to remove focus from the active element: {e}")
for method in search_methods: for method in search_methods:
try: try:
logger.info(f"Attempt {attempt + 1}: Searching for 'Easy Apply' button using {method['description']}") logger.debug(f"Attempting search using {method['description']}")
if method.get('find_elements'): if method.get('find_elements'):
buttons = self.driver.find_elements(By.XPATH, method['xpath']) buttons = self.driver.find_elements(By.XPATH, method['xpath'])
if buttons: if buttons:
for index, button in enumerate(buttons): for index, button in enumerate(buttons):
try: try:
WebDriverWait(self.driver, timeout).until(EC.visibility_of(button))
WebDriverWait(self.driver, timeout).until(EC.element_to_be_clickable(button))
logger.info(f"Found 'Easy Apply' button {index + 1}, attempting to click")
self.driver.execute_script("arguments[0].scrollIntoView(true);", button) WebDriverWait(self.driver, 10).until(EC.visibility_of(button))
time.sleep(1) WebDriverWait(self.driver, 10).until(EC.element_to_be_clickable(button))
if button.is_enabled() and button.is_displayed(): logger.debug(f"Found 'Easy Apply' button {index + 1}, attempting to click")
return button return button
else:
raise Exception(f"Button {index + 1} is not enabled or not displayed")
except Exception as e: except Exception as e:
logger.warning(f"Button {index + 1} found but not clickable: {e}") logger.warning(f"Button {index + 1} found but not clickable: {e}")
else: else:
raise TimeoutException("No 'Easy Apply' buttons found") raise TimeoutException("No 'Easy Apply' buttons found")
else: else:
button = WebDriverWait(self.driver, timeout).until(
button = WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.XPATH, method['xpath'])) EC.presence_of_element_located((By.XPATH, method['xpath']))
) )
WebDriverWait(self.driver, timeout).until(EC.visibility_of(button)) WebDriverWait(self.driver, 10).until(EC.visibility_of(button))
WebDriverWait(self.driver, timeout).until(EC.element_to_be_clickable(button)) WebDriverWait(self.driver, 10).until(EC.element_to_be_clickable(button))
logger.info("Found 'Easy Apply' button, attempting to click") logger.debug("Found 'Easy Apply' button, attempting to click")
self.driver.execute_script("arguments[0].scrollIntoView(true);", button)
time.sleep(1)
if button.is_enabled() and button.is_displayed():
return button return button
else:
raise Exception("Button is not enabled or not displayed")
except TimeoutException: except TimeoutException:
logger.warning(f"Timeout during search using {method['description']}") logger.warning(f"Timeout during search using {method['description']}")
except Exception as e: except Exception as e:
logger.warning(f"Failed to click 'Easy Apply' button using {method['description']} on attempt {attempt + 1}: {e}") logger.warning(
f"Failed to click 'Easy Apply' button using {method['description']} on attempt {attempt + 1}: {e}")
self.check_for_premium_redirect(job) self.check_for_premium_redirect(job)
if attempt == 0: if attempt == 0:
logger.info("Refreshing page and clicking on body to retry finding 'Easy Apply' button") logger.debug("Refreshing page to retry finding 'Easy Apply' button")
self.driver.refresh() self.driver.refresh()
time.sleep(random.randint(3, 5)) time.sleep(random.randint(3, 5))
try:
body_element = self.driver.find_element(By.TAG_NAME, 'body')
body_element.click()
logger.info("Clicked on body element to reset the page state")
except Exception as e:
logger.warning(f"Failed to click on body element: {e}")
attempt += 1 attempt += 1
logger.error("No clickable 'Easy Apply' button found after 2 attempts.") page_source = self.driver.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") raise Exception("No clickable 'Easy Apply' button found")
def _get_job_description(self) -> str: def _get_job_description(self) -> str:
@ -285,11 +236,11 @@ class LinkedInEasyApplier:
return description return description
except NoSuchElementException: except NoSuchElementException:
tb_str = traceback.format_exc() 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}") raise Exception(f"Job description not found: \nTraceback:\n{tb_str}")
except Exception: except Exception:
tb_str = traceback.format_exc() 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}") raise Exception(f"Error getting Job description: \nTraceback:\n{tb_str}")
def _get_job_recruiter(self): def _get_job_recruiter(self):
@ -306,13 +257,13 @@ class LinkedInEasyApplier:
if recruiter_elements: if recruiter_elements:
recruiter_element = recruiter_elements[0] recruiter_element = recruiter_elements[0]
recruiter_link = recruiter_element.get_attribute('href') 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 return recruiter_link
else: else:
logger.debug("No recruiter link found in the hiring team section") logger.debug("No recruiter link found in the hiring team section")
return "" return ""
except Exception as e: except Exception as e:
logger.warning("Failed to retrieve recruiter information: %s", e) logger.warning(f"Failed to retrieve recruiter information: {e}")
return "" return ""
def _scroll_page(self) -> None: def _scroll_page(self) -> None:
@ -322,7 +273,7 @@ class LinkedInEasyApplier:
utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=True) utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=True)
def _fill_application_form(self, job): 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: while True:
self.fill_up(job) self.fill_up(job)
if self._next_or_submit(): if self._next_or_submit():
@ -352,13 +303,13 @@ class LinkedInEasyApplier:
By.XPATH, "//label[contains(.,'to stay up to date with their page.')]") By.XPATH, "//label[contains(.,'to stay up to date with their page.')]")
follow_checkbox.click() follow_checkbox.click()
except Exception as e: 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: def _check_for_errors(self) -> None:
logger.debug("Checking for form errors") logger.debug("Checking for form errors")
error_elements = self.driver.find_elements(By.CLASS_NAME, 'artdeco-inline-feedback--error') error_elements = self.driver.find_elements(By.CLASS_NAME, 'artdeco-inline-feedback--error')
if error_elements: 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])}") raise Exception(f"Failed answering or file upload. {str([e.text for e in error_elements])}")
def _discard_application(self) -> None: def _discard_application(self) -> None:
@ -369,10 +320,10 @@ class LinkedInEasyApplier:
self.driver.find_elements(By.CLASS_NAME, 'artdeco-modal__confirm-dialog-btn')[0].click() self.driver.find_elements(By.CLASS_NAME, 'artdeco-modal__confirm-dialog-btn')[0].click()
time.sleep(random.uniform(3, 5)) time.sleep(random.uniform(3, 5))
except Exception as e: 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: 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: try:
easy_apply_content = WebDriverWait(self.driver, 10).until( easy_apply_content = WebDriverWait(self.driver, 10).until(
@ -435,7 +386,7 @@ class LinkedInEasyApplier:
def _is_upload_field(self, element: WebElement) -> bool: def _is_upload_field(self, element: WebElement) -> bool:
is_upload = bool(element.find_elements(By.XPATH, ".//input[@type='file']")) 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 return is_upload
def _handle_upload_fields(self, element: WebElement, job) -> None: def _handle_upload_fields(self, element: WebElement, job) -> None:
@ -801,24 +752,12 @@ class LinkedInEasyApplier:
if dropdowns: if dropdowns:
dropdown = dropdowns[0] dropdown = dropdowns[0]
select = Select(dropdown) select = Select(dropdown)
options = [option.text for option in select.options if option.text != "Select an option"] options = [option.text for option in select.options]
logger.debug(f"Dropdown options found: {options}") logger.debug(f"Dropdown options found: {options}")
try: question_text = question.find_element(By.TAG_NAME, 'label').text.lower()
question_text = question.find_element(By.TAG_NAME, 'label').text.lower().strip() logger.debug(f"Processing dropdown or combobox question: {question_text}")
except NoSuchElementException:
logger.warning("Label not found, trying to extract question text from <span> or other elements")
try:
question_text = question.find_element(By.CSS_SELECTOR,
'span[aria-hidden="true"]').text.lower().strip()
except NoSuchElementException:
question_text = section.get_attribute('data-test-text-entity-list-form-title') or "unknown question"
question_text = question_text.lower().strip()
logger.debug(f"Processing dropdown question: {question_text}")
current_selection = select.first_selected_option.text current_selection = select.first_selected_option.text
logger.debug(f"Current selection: {current_selection}") logger.debug(f"Current selection: {current_selection}")
@ -833,14 +772,14 @@ class LinkedInEasyApplier:
logger.debug(f"Found existing answer for question '{question_text}': {existing_answer}") logger.debug(f"Found existing answer for question '{question_text}': {existing_answer}")
if current_selection != existing_answer: if current_selection != existing_answer:
logger.debug(f"Updating selection to: {existing_answer}") logger.debug(f"Updating selection to: {existing_answer}")
self._select_dropdown_option(select, existing_answer) self._select_dropdown_option(dropdown, existing_answer)
return True return True
logger.debug(f"No existing answer found, querying model for: {question_text}") logger.debug(f"No existing answer found, querying model for: {question_text}")
answer = self.gpt_answerer.answer_question_from_options(question_text, options) answer = self.gpt_answerer.answer_question_from_options(question_text, options)
self._save_questions_to_json({'type': 'dropdown', 'question': question_text, 'answer': answer}) self._save_questions_to_json({'type': 'dropdown', 'question': question_text, 'answer': answer})
self._select_dropdown_option(select, answer) self._select_dropdown_option(dropdown, answer)
logger.debug(f"Selected new dropdown answer: {answer}") logger.debug(f"Selected new dropdown answer: {answer}")
return True return True
@ -855,55 +794,35 @@ class LinkedInEasyApplier:
logger.warning(f"Failed to handle dropdown or combobox question: {e}", exc_info=True) logger.warning(f"Failed to handle dropdown or combobox question: {e}", exc_info=True)
return False return False
def _select_dropdown_option(self, select: Select, text: str) -> None:
try:
select.select_by_visible_text(text)
logger.debug(f"Selected option: {text}")
except Exception as e:
logger.error(f"Failed to select option '{text}': {e}")
def _is_numeric_field(self, field: WebElement) -> bool: def _is_numeric_field(self, field: WebElement) -> bool:
field_type = field.get_attribute('type').lower() field_type = field.get_attribute('type').lower()
field_id = field.get_attribute("id").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) 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 return is_numeric
def _enter_text(self, element: WebElement, text: str) -> None: def _enter_text(self, element: WebElement, text: str) -> None:
logger.debug("Entering text: %s", text) logger.debug(f"Entering text: {text}")
element.clear() element.clear()
element.send_keys(text) element.send_keys(text)
def _select_radio(self, radios: List[WebElement], answer: str) -> None: 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: for radio in radios:
if answer in radio.text.lower(): if answer in radio.text.lower():
radio.find_element(By.TAG_NAME, 'label').click() radio.find_element(By.TAG_NAME, 'label').click()
return return
radios[-1].find_element(By.TAG_NAME, 'label').click() radios[-1].find_element(By.TAG_NAME, 'label').click()
def _select_dropdown_option(self, element: WebElement, text: str) -> None:
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: def _save_questions_to_json(self, question_data: dict) -> None:
"""
Save question data to a JSON file, with filtering to exclude company-specific or unsuitable questions.
Args:
question_data (dict): The question and answer data to be saved.
"""
output_file = 'answers.json' output_file = 'answers.json'
question_data['question'] = self._sanitize_text(question_data['question']) 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}")
# List of keywords to exclude certain questions from being saved
exclusion_keywords = ["why us", "summary"]
# Check if the question contains any exclusion keywords
if any(keyword in question_data['question'].lower() for keyword in exclusion_keywords):
logger.info(f"Skipping saving question due to company-specific keywords: {question_data['question']}")
return # Skip saving this question if it's company-specific
try: try:
try: try:
with open(output_file, 'r') as f: with open(output_file, 'r') as f:
@ -917,19 +836,17 @@ class LinkedInEasyApplier:
except FileNotFoundError: except FileNotFoundError:
logger.warning("JSON file not found, creating new file") logger.warning("JSON file not found, creating new file")
data = [] data = []
data.append(question_data) data.append(question_data)
with open(output_file, 'w') as f: with open(output_file, 'w') as f:
json.dump(data, f, indent=4) json.dump(data, f, indent=4)
logger.debug("Question data saved successfully to JSON") logger.debug("Question data saved successfully to JSON")
except Exception: except Exception:
tb_str = traceback.format_exc() 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}") raise Exception(f"Error saving questions data to JSON file: \nTraceback:\n{tb_str}")
def _sanitize_text(self, text: str) -> str: def _sanitize_text(self, text: str) -> str:
sanitized_text = text.lower().strip().replace('"', '').replace('\\', '') sanitized_text = text.lower().strip().replace('"', '').replace('\\', '')
sanitized_text = re.sub(r'[\x00-\x1F\x7F]', '', sanitized_text).replace('\n', ' ').replace('\r', '').rstrip(',') 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 return sanitized_text

181
src/linkedIn_job_manager.py
View file

@ -12,7 +12,7 @@ from selenium.webdriver.common.by import By
import src.utils as utils import src.utils as utils
from src.job import Job from src.job import Job
from src.linkedIn_easy_applier import LinkedInEasyApplier from src.linkedIn_easy_applier import LinkedInEasyApplier
from src.utils import logger from loguru import logger
class EnvironmentKeys: class EnvironmentKeys:
@ -20,19 +20,18 @@ class EnvironmentKeys:
logger.debug("Initializing EnvironmentKeys") logger.debug("Initializing EnvironmentKeys")
self.skip_apply = self._read_env_key_bool("SKIP_APPLY") self.skip_apply = self._read_env_key_bool("SKIP_APPLY")
self.disable_description_filter = self._read_env_key_bool("DISABLE_DESCRIPTION_FILTER") self.disable_description_filter = self._read_env_key_bool("DISABLE_DESCRIPTION_FILTER")
logger.debug("EnvironmentKeys initialized: skip_apply=%s, disable_description_filter=%s", logger.debug(f"EnvironmentKeys initialized: skip_apply={self.skip_apply}, disable_description_filter={self.disable_description_filter}")
self.skip_apply, self.disable_description_filter)
@staticmethod @staticmethod
def _read_env_key(key: str) -> str: def _read_env_key(key: str) -> str:
value = os.getenv(key, "") value = os.getenv(key, "")
logger.debug("Read environment key %s: %s", key, value) logger.debug(f"Read environment key {key}: {value}")
return value return value
@staticmethod @staticmethod
def _read_env_key_bool(key: str) -> bool: def _read_env_key_bool(key: str) -> bool:
value = os.getenv(key) == "True" 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 return value
@ -72,28 +71,10 @@ class LinkedInJobManager:
logger.debug("Setting resume generator manager") logger.debug("Setting resume generator manager")
self.resume_generator_manager = resume_generator_manager self.resume_generator_manager = resume_generator_manager
def wait_or_skip(self, time_left):
"""Method for waiting or skipping the sleep time based on user input"""
if time_left > 0:
try:
user_input = inputimeout(
prompt=f"Sleeping for {time_left} seconds. Press 'y' to skip waiting. Timeout 60 seconds : ",
timeout=60).strip().lower()
except TimeoutOccurred:
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)
def start_applying(self): def start_applying(self):
logger.debug("Starting job application process") logger.debug("Starting job application process")
self.easy_applier_component = LinkedInEasyApplier(self.driver, self.resume_path, self.set_old_answers, self.easy_applier_component = LinkedInEasyApplier(self.driver, self.resume_path, self.set_old_answers,
self.gpt_answerer, self.resume_generator_manager, self.gpt_answerer, self.resume_generator_manager)
self.parameters)
searches = list(product(self.positions, self.locations)) searches = list(product(self.positions, self.locations))
random.shuffle(searches) random.shuffle(searches)
page_sleep = 0 page_sleep = 0
@ -103,21 +84,21 @@ class LinkedInJobManager:
for position, location in searches: for position, location in searches:
location_url = "&location=" + location location_url = "&location=" + location
job_page_number = -1 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: try:
while True: while True:
page_sleep += 1 page_sleep += 1
job_page_number += 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) self.next_job_page(position, location_url, job_page_number)
time.sleep(random.uniform(1.5, 3.5)) 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: try:
jobs = self.get_jobs_from_page() jobs = self.get_jobs_from_page()
if not jobs: 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 break
except Exception as e: except Exception as e:
logger.error(f"Failed to retrieve jobs: {e}") logger.error(f"Failed to retrieve jobs: {e}")
@ -126,40 +107,77 @@ class LinkedInJobManager:
try: try:
self.apply_jobs() self.apply_jobs()
except Exception as e: except Exception as e:
logger.error("Error during job application: %s", e) logger.error(f"Error during job application: {e}")
utils.printred(f"Error during job application: {e}")
continue 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() time_left = minimum_page_time - time.time()
# Use the wait_or_skip function for sleeping # Ask user if they want to skip waiting, with timeout
self.wait_or_skip(time_left) if time_left > 0:
try:
user_input = inputimeout(
prompt=f"Sleeping for {time_left} seconds. Press 'y' to skip waiting. Timeout 60 seconds : ",
timeout=60).strip().lower()
except TimeoutOccurred:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
else:
logger.debug(f"Sleeping for {time_left} seconds as user chose not to skip.")
time.sleep(time_left)
minimum_page_time = time.time() + minimum_time minimum_page_time = time.time() + minimum_time
if page_sleep % 5 == 0: if page_sleep % 5 == 0:
sleep_time = random.randint(5, 34) sleep_time = random.randint(5, 34)
# Use the wait_or_skip function for extended sleep try:
self.wait_or_skip(sleep_time) user_input = inputimeout(
prompt=f"Sleeping for {sleep_time / 60} minutes. Press 'y' to skip waiting. Timeout 60 seconds : ",
timeout=60).strip().lower()
except TimeoutOccurred:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
else:
logger.debug(f"Sleeping for {sleep_time} seconds.")
time.sleep(sleep_time)
page_sleep += 1 page_sleep += 1
except Exception as e: except Exception as e:
logger.error("Unexpected error during job search: %s", e) logger.error(f"Unexpected error during job search: {e}")
utils.printred(f"Unexpected error: {e}")
continue continue
time_left = minimum_page_time - time.time() time_left = minimum_page_time - time.time()
# Use the wait_or_skip function again before moving to the next search if time_left > 0:
self.wait_or_skip(time_left) try:
user_input = inputimeout(
prompt=f"Sleeping for {time_left} seconds. Press 'y' to skip waiting. Timeout 60 seconds : ",
timeout=60).strip().lower()
except TimeoutOccurred:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
else:
logger.debug(f"Sleeping for {time_left} seconds as user chose not to skip.")
time.sleep(time_left)
minimum_page_time = time.time() + minimum_time minimum_page_time = time.time() + minimum_time
if page_sleep % 5 == 0: if page_sleep % 5 == 0:
sleep_time = random.randint(50, 90) sleep_time = random.randint(50, 90)
# Use the wait_or_skip function for a longer sleep period try:
self.wait_or_skip(sleep_time) user_input = inputimeout(
prompt=f"Sleeping for {sleep_time / 60} minutes. Press 'y' to skip waiting: ",
timeout=60).strip().lower()
except TimeoutOccurred:
user_input = '' # No input after timeout
if user_input == 'y':
logger.debug("User chose to skip waiting.")
else:
logger.debug(f"Sleeping for {sleep_time} seconds.")
time.sleep(sleep_time)
page_sleep += 1 page_sleep += 1
def get_jobs_from_page(self): def get_jobs_from_page(self):
@ -168,7 +186,6 @@ class LinkedInJobManager:
no_jobs_element = self.driver.find_element(By.CLASS_NAME, 'jobs-search-two-pane__no-results-banner--expand') 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(): 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.") logger.debug("No matching jobs found on this page, skipping.")
return [] return []
@ -178,12 +195,11 @@ class LinkedInJobManager:
try: try:
job_results = self.driver.find_element(By.CLASS_NAME, "jobs-search-results-list") job_results = self.driver.find_element(By.CLASS_NAME, "jobs-search-results-list")
utils.scroll_slow(self.driver, job_results) utils.scroll_slow(self.driver, job_results)
# utils.scroll_slow(self.driver, job_results, step=300, reverse=True) utils.scroll_slow(self.driver, job_results, step=300, reverse=True)
job_list_elements = self.driver.find_elements(By.CLASS_NAME, 'scaffold-layout__list-container')[ 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') 0].find_elements(By.CLASS_NAME, 'jobs-search-results__list-item')
if not job_list_elements: if not job_list_elements:
utils.printyellow("No job class elements found on page.")
logger.debug("No job class elements found on page, skipping.") logger.debug("No job class elements found on page, skipping.")
return [] return []
@ -201,7 +217,6 @@ class LinkedInJobManager:
try: try:
no_jobs_element = self.driver.find_element(By.CLASS_NAME, 'jobs-search-two-pane__no-results-banner--expand') 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(): 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") logger.debug("No matching jobs found on this page, skipping")
return return
except NoSuchElementException: except NoSuchElementException:
@ -215,7 +230,6 @@ class LinkedInJobManager:
0].find_elements(By.CLASS_NAME, 'jobs-search-results__list-item') 0].find_elements(By.CLASS_NAME, 'jobs-search-results__list-item')
if not job_list_elements: 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") logger.debug("No job class elements found on page, skipping")
return return
@ -236,48 +250,37 @@ class LinkedInJobManager:
# Iterate over each job insight element to find the one containing the word "applicant" # Iterate over each job insight element to find the one containing the word "applicant"
for element in job_insight_elements: for element in job_insight_elements:
applicants_text = element.text.strip().lower() logger.debug(f"Checking element text: {element.text}")
logger.debug(f"Checking element text: {applicants_text}") if "applicant" in element.text.lower():
# Found an element containing "applicant"
applicants_text = element.text.strip()
logger.debug(f"Applicants text found: {applicants_text}")
# Look for keywords indicating the presence of applicants count # Extract numeric digits from the text (e.g., "70 applicants" -> "70")
if "applicant" in applicants_text:
logger.info(f"Applicants text found: {applicants_text}")
# Try to find numeric value in the text, such as "27 applicants" or "over 100 applicants"
applicants_count = ''.join(filter(str.isdigit, applicants_text)) applicants_count = ''.join(filter(str.isdigit, applicants_text))
logger.debug(f"Extracted applicants count: {applicants_count}")
if applicants_count: if applicants_count:
if "over" in applicants_text.lower():
applicants_count = int(applicants_count) + 1 # Handle "over X applicants"
logger.debug(f"Applicants count adjusted for 'over': {applicants_count}")
else:
applicants_count = int(applicants_count) # Convert the extracted number to an integer applicants_count = int(applicants_count) # Convert the extracted number to an integer
logger.info(f"Extracted numeric applicants count: {applicants_count}") break
# Handle case with "over X applicants"
if "over" in applicants_text:
applicants_count += 1
logger.info(f"Adjusted applicants count for 'over': {applicants_count}")
logger.info(f"Final applicants count: {applicants_count}")
else:
logger.warning(f"Applicants count could not be extracted from text: {applicants_text}")
break # Stop after finding the first valid applicants count element
else:
logger.info(f"Skipping element as it does not contain 'applicant': {applicants_text}")
# Check if applicants_count is valid (not None) before performing comparisons # Check if applicants_count is valid (not None) before performing comparisons
if applicants_count is not None: if applicants_count is not None:
# Perform the threshold check for applicants count # Perform the threshold check for applicants count
if applicants_count < self.min_applicants or applicants_count > self.max_applicants: 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}") logger.debug(f"Skipping {job.title} at {job.company}, applicants count: {applicants_count}")
self.write_to_file(job, "skipped_due_to_applicants") self.write_to_file(job, "skipped_due_to_applicants")
continue # Skip this job if applicants count is outside the threshold
else: else:
logger.debug(f"Applicants count {applicants_count} is within the threshold") logger.debug(f"Applicants count {applicants_count} is within the threshold")
else: else:
# If no applicants count was found, log a warning but continue the process # If no applicants count was found, log a warning but continue the process
logger.warning( logger.warning(
f"Applicants count not found for {job.title} at {job.company}, but continuing with application.") f"Applicants count not found for {job.title} at {job.company}, continuing with application.")
except NoSuchElementException: except NoSuchElementException:
# Log a warning if the job insight elements are not found, but do not stop the job application process # Log a warning if the job insight elements are not found, but do not stop the job application process
logger.warning( logger.warning(
@ -294,8 +297,7 @@ class LinkedInJobManager:
logger.debug(f"Continuing with job application for {job.title} at {job.company}") logger.debug(f"Continuing with job application for {job.title} at {job.company}")
if self.is_blacklisted(job.title, job.company, job.link): if self.is_blacklisted(job.title, job.company, job.link):
utils.printyellow(f"Blacklisted {job.title} at {job.company}, skipping...") logger.debug(f"Job blacklisted: {job.title} at {job.company}")
logger.debug("Job blacklisted: %s at %s", job.title, job.company)
self.write_to_file(job, "skipped") self.write_to_file(job, "skipped")
continue continue
if self.is_already_applied_to_job(job.title, job.company, job.link): if self.is_already_applied_to_job(job.title, job.company, job.link):
@ -308,15 +310,14 @@ class LinkedInJobManager:
if job.apply_method not in {"Continue", "Applied", "Apply"}: if job.apply_method not in {"Continue", "Applied", "Apply"}:
self.easy_applier_component.job_apply(job) self.easy_applier_component.job_apply(job)
self.write_to_file(job, "success") 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: except Exception as e:
logger.error("Failed to apply for %s at %s: %s", job.title, job.company, e) logger.error(f"Failed to apply for {job.title} at {job.company}: {e}")
utils.printred(f"Failed to apply for {job.title} at {job.company}: {e}")
self.write_to_file(job, "failed") self.write_to_file(job, "failed")
continue continue
def write_to_file(self, job, file_name): 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 = Path(job.pdf_path).resolve()
pdf_path = pdf_path.as_uri() pdf_path = pdf_path.as_uri()
data = { data = {
@ -331,27 +332,26 @@ class LinkedInJobManager:
if not file_path.exists(): if not file_path.exists():
with open(file_path, 'w', encoding='utf-8') as f: with open(file_path, 'w', encoding='utf-8') as f:
json.dump([data], f, indent=4) 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: else:
with open(file_path, 'r+', encoding='utf-8') as f: with open(file_path, 'r+', encoding='utf-8') as f:
try: try:
existing_data = json.load(f) existing_data = json.load(f)
except json.JSONDecodeError: 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 = []
existing_data.append(data) existing_data.append(data)
f.seek(0) f.seek(0)
json.dump(existing_data, f, indent=4) json.dump(existing_data, f, indent=4)
f.truncate() 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): def get_base_search_url(self, parameters):
logger.debug("Constructing base search URL") logger.debug("Constructing base search URL")
url_parts = [] url_parts = []
if parameters['remote']: if parameters['remote']:
url_parts.append("f_CF=f_WRA") url_parts.append("f_CF=f_WRA")
experience_levels = [str(i + 1) for i, (level, v) in enumerate(parameters.get('experience_level', {}).items()) experience_levels = [str(i + 1) for i, (level, v) in enumerate(parameters.get('experience_level', {}).items()) if
if
v] v]
if experience_levels: if experience_levels:
url_parts.append(f"f_E={','.join(experience_levels)}") url_parts.append(f"f_E={','.join(experience_levels)}")
@ -369,11 +369,11 @@ class LinkedInJobManager:
url_parts.append("f_LF=f_AL") # Easy Apply url_parts.append("f_LF=f_AL") # Easy Apply
base_url = "&".join(url_parts) base_url = "&".join(url_parts)
full_url = f"?{base_url}{date_param}" 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 return full_url
def next_job_page(self, position, location, job_page): 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( self.driver.get(
f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}{location}&start={job_page * 25}") f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}{location}&start={job_page * 25}")
@ -384,39 +384,36 @@ class LinkedInJobManager:
job_title = job_tile.find_element(By.CLASS_NAME, 'job-card-list__title').text 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] 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 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: except NoSuchElementException:
utils.printyellow("Some job information (title, link, or company) is missing.")
logger.warning("Some job information (title, link, or company) is missing.") logger.warning("Some job information (title, link, or company) is missing.")
try: try:
job_location = job_tile.find_element(By.CLASS_NAME, 'job-card-container__metadata-item').text job_location = job_tile.find_element(By.CLASS_NAME, 'job-card-container__metadata-item').text
except NoSuchElementException: except NoSuchElementException:
utils.printyellow("Job location is missing.")
logger.warning("Job location is missing.") logger.warning("Job location is missing.")
try: try:
apply_method = job_tile.find_element(By.CLASS_NAME, 'job-card-container__apply-method').text apply_method = job_tile.find_element(By.CLASS_NAME, 'job-card-container__apply-method').text
except NoSuchElementException: except NoSuchElementException:
apply_method = "Applied" apply_method = "Applied"
utils.printyellow("Apply method not found, assuming 'Applied'.")
logger.warning("Apply method not found, assuming 'Applied'.") logger.warning("Apply method not found, assuming 'Applied'.")
return job_title, company, job_location, link, apply_method return job_title, company, job_location, link, apply_method
def is_blacklisted(self, job_title, company, link): 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(' ') job_title_words = job_title.lower().split(' ')
title_blacklisted = any(word in job_title_words for word in self.title_blacklist) 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) company_blacklisted = company.strip().lower() in (word.strip().lower() for word in self.company_blacklist)
link_seen = link in self.seen_jobs link_seen = link in self.seen_jobs
is_blacklisted = title_blacklisted or company_blacklisted or link_seen 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 return title_blacklisted or company_blacklisted or link_seen
def is_already_applied_to_job(self, job_title, company, link): def is_already_applied_to_job(self, job_title, company, link):
link_seen = link in self.seen_jobs link_seen = link in self.seen_jobs
if link_seen: 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 return link_seen
def is_already_applied_to_company(self, company): def is_already_applied_to_company(self, company):
@ -432,7 +429,7 @@ class LinkedInJobManager:
existing_data = json.load(f) existing_data = json.load(f)
for applied_job in existing_data: for applied_job in existing_data:
if applied_job['company'].strip().lower() == company.strip().lower(): if applied_job['company'].strip().lower() == company.strip().lower():
utils.printyellow( logger.debug(
f"Already applied at {company} (once per company policy), skipping...") f"Already applied at {company} (once per company policy), skipping...")
return True return True
except json.JSONDecodeError: except json.JSONDecodeError:

17
src/linkedin-api.py
View file

@ -2,11 +2,12 @@ from typing import Dict, List
from linkedin_api import Linkedin from linkedin_api import Linkedin
from typing import Optional, Union, Literal from typing import Optional, Union, Literal
from urllib.parse import quote, urlencode, parse_qs, urlparse from urllib.parse import quote, urlencode, parse_qs, urlparse
import logging # import logging
import json import json
from loguru import logger
# set log to all debug # set log to all debug
logging.basicConfig(level=logging.INFO) # logging.basicConfig(level=logging.INFO)
class LinkedInEvolvedAPI(Linkedin): class LinkedInEvolvedAPI(Linkedin):
already_applied_jobs: List[str] = [] already_applied_jobs: List[str] = []
@ -388,7 +389,7 @@ class LinkedInEvolvedAPI(Linkedin):
case 200: case 200:
parse_res = res.json() parse_res = res.json()
url = parse_res['data']['value'] url = parse_res['data']['value']
logging.info(url) logger.info(url)
return url return url
case _: case _:
self.logger.error("Failed to create a request PDF") self.logger.error("Failed to create a request PDF")
@ -496,22 +497,22 @@ if __name__ == "__main__":
resume: str = api.upload_linkedin_resume("resume.pdf") resume: str = api.upload_linkedin_resume("resume.pdf")
if isinstance(resume, bool): if isinstance(resume, bool):
logging.error("Failed to upload resume") logger.error("Failed to upload resume")
continue continue
elif isinstance(resume, str): elif isinstance(resume, str):
logging.info(f"Resume uploaded with hash {resume}") logger.info(f"Resume uploaded with hash {resume}")
else: else:
logging.error("Unknown error") logger.error("Unknown error")
continue continue
if job_id in api.already_applied_jobs: 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 continue
fields = api.get_fields_for_easy_apply(job_id) fields = api.get_fields_for_easy_apply(job_id)
for field in fields: for field in fields:
print(field) logger.info(field)
break break

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

@ -0,0 +1,569 @@
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.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, llm_api_url: str):
from langchain_openai import ChatOpenAI
self.model = ChatOpenAI(model_name=llm_model, openai_api_key=api_key,
temperature=0.4, base_url=llm_api_url)
def invoke(self, prompt: str) -> str:
logger.debug("Invoking OpenAI API")
response = self.model.invoke(prompt)
return response
class ClaudeModel(AIModel):
def __init__(self, api_key: str, llm_model: str, llm_api_url: str):
from langchain_anthropic import ChatAnthropic
self.model = ChatAnthropic(model=llm_model, api_key=api_key,
temperature=0.4, base_url=llm_api_url)
def invoke(self, prompt: str) -> str:
response = self.model.invoke(prompt)
return response
class OllamaModel(AIModel):
def __init__(self, api_key: str, llm_model: str, llm_api_url: str):
from langchain_ollama import ChatOllama
self.model = ChatOllama(model=llm_model, base_url=llm_api_url)
def invoke(self, prompt: str) -> str:
response = self.model.invoke(prompt)
return response
class GeminiModel(AIModel):
def __init__(self, api_key:str, llm_model: str, llm_api_url: str):
from langchain_google_genai import ChatGoogleGenerativeAI
self.model = ChatGoogleGenerativeAI(model=llm_model, google_api_key=api_key)
def invoke(self, prompt: str) -> str:
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['llm_api_url']
logger.debug('Using {0} with {1} from {2}'.format(
llm_model_type, llm_model, llm_api_url))
if llm_model_type == "openai":
return OpenAIModel(api_key, llm_model, llm_api_url)
elif llm_model_type == "claude":
return ClaudeModel(api_key, llm_model, llm_api_url)
elif llm_model_type == "ollama":
return OllamaModel(api_key, llm_model, llm_api_url)
elif llm_model_type == "gemini":
return GeminiModel(api_key, llm_model, llm_api_url)
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"

82
src/utils.py
View file

@ -1,41 +1,48 @@
import logging import logging
import os import os
import random import random
import sys
import time import time
from selenium import webdriver from selenium import webdriver
from loguru import logger
from app_config import MINIMUM_LOG_LEVEL
log_file = "app_log.log" log_file = "app_log.log"
logging.basicConfig( # TODO: REMOVE THE FOLLOWING BLOCK: No need as Loguru handles everything by default
level=logging.INFO, # logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', # level=logging.INFO,
handlers=[ # format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
logging.FileHandler(log_file, mode='a', encoding='utf-8'), # handlers=[
logging.StreamHandler() # 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 # ],
) # 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') if MINIMUM_LOG_LEVEL in ["DEBUG", "TRACE", "INFO", "WARNING", "ERROR", "CRITICAL"]:
file_handler.setFormatter(formatter) logger.remove()
logger.addHandler(file_handler) logger.add(sys.stderr, level=MINIMUM_LOG_LEVEL)
logger.setLevel(logging.INFO) 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") chromeProfilePath = os.path.join(os.getcwd(), "chrome_profile", "linkedin_profile")
def ensure_chrome_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) profile_dir = os.path.dirname(chromeProfilePath)
if not os.path.exists(profile_dir): if not os.path.exists(profile_dir):
os.makedirs(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): if not os.path.exists(chromeProfilePath):
os.makedirs(chromeProfilePath) os.makedirs(chromeProfilePath)
logger.debug("Created Chrome profile directory: %s", chromeProfilePath) logger.debug(f"Created Chrome profile directory: {chromeProfilePath}")
return chromeProfilePath return chromeProfilePath
@ -43,13 +50,12 @@ def is_scrollable(element):
scroll_height = element.get_attribute("scrollHeight") scroll_height = element.get_attribute("scrollHeight")
client_height = element.get_attribute("clientHeight") client_height = element.get_attribute("clientHeight")
scrollable = int(scroll_height) > int(client_height) scrollable = int(scroll_height) > int(client_height)
logger.debug("Element scrollable check: scrollHeight=%s, clientHeight=%s, scrollable=%s", scroll_height, logger.debug(f"Element scrollable check: scrollHeight={scroll_height}, clientHeight={client_height}, scrollable={scrollable}")
client_height, scrollable)
return scrollable return scrollable
def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse=False): 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: if reverse:
start, end = end, start start, end = end, start
@ -61,18 +67,16 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse
max_scroll_height = int(scrollable_element.get_attribute("scrollHeight")) max_scroll_height = int(scrollable_element.get_attribute("scrollHeight"))
current_scroll_position = int(scrollable_element.get_attribute("scrollTop")) current_scroll_position = int(scrollable_element.get_attribute("scrollTop"))
logger.debug("Max scroll height of the element: %d", max_scroll_height) logger.debug(f"Max scroll height of the element: {max_scroll_height}")
logger.debug("Current scroll position: %d", current_scroll_position) logger.debug(f"Current scroll position: {current_scroll_position}")
if reverse: if reverse:
if current_scroll_position < start: if current_scroll_position < start:
start = current_scroll_position 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: else:
if end > max_scroll_height: 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 end = max_scroll_height
script_scroll_to = "arguments[0].scrollTop = arguments[1];" 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 scrollable_element.is_displayed():
if not is_scrollable(scrollable_element): if not is_scrollable(scrollable_element):
logger.warning("The element is not scrollable.") logger.warning("The element is not scrollable.")
print("The element is not scrollable.")
return return
if (step > 0 and start >= end) or (step < 0 and start <= end): if (step > 0 and start >= end) or (step < 0 and start <= end):
logger.warning("No scrolling will occur due to incorrect start/end values.") logger.warning("No scrolling will occur due to incorrect start/end values.")
print("No scrolling will occur due to incorrect start/end values.")
return return
position = start 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): while (step > 0 and position < end) or (step < 0 and position > end):
if position == previous_position: if position == previous_position:
# Avoid re-scrolling to the same 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 break
try: try:
driver.execute_script(script_scroll_to, scrollable_element, position) 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: except Exception as e:
logger.error("Error during scrolling: %s", e) logger.error(f"Error during scrolling: {e}")
print(f"Error during scrolling: {e}")
previous_position = position previous_position = position
position += step 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 # Ensure the final scroll position is correct
driver.execute_script(script_scroll_to, scrollable_element, end) 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) time.sleep(0.5)
else: else:
logger.warning("The element is not visible.") logger.warning("The element is not visible.")
print("The element is not visible.")
except Exception as e: except Exception as e:
logger.error("Exception occurred during scrolling: %s", e) logger.error(f"Exception occurred during scrolling: {e}")
print(f"Exception occurred: {e}")
def chrome_browser_options(): def chrome_browser_options():
@ -159,7 +158,7 @@ def chrome_browser_options():
profile_dir = os.path.basename(chromeProfilePath) profile_dir = os.path.basename(chromeProfilePath)
options.add_argument('--user-data-dir=' + initial_path) options.add_argument('--user-data-dir=' + initial_path)
options.add_argument("--profile-directory=" + profile_dir) options.add_argument("--profile-directory=" + profile_dir)
logger.debug("Using Chrome profile directory: %s", chromeProfilePath) logger.debug(f"Using Chrome profile directory: {chromeProfilePath}")
else: else:
options.add_argument("--incognito") options.add_argument("--incognito")
logger.debug("Using Chrome in incognito mode") logger.debug("Using Chrome in incognito mode")
@ -179,8 +178,3 @@ def printyellow(text):
reset = "\033[0m" reset = "\033[0m"
logger.debug("Printing text in yellow: %s", text) logger.debug("Printing text in yellow: %s", text)
print(f"{yellow}{text}{reset}") print(f"{yellow}{text}{reset}")
def stringWidth(text, font, font_size):
bbox = font.getbbox(text)
return bbox[2] - bbox[0]

4
tests/test_linkedIn_job_manager.py
View file

@ -5,6 +5,7 @@ import os
import pytest import pytest
from src.linkedIn_job_manager import LinkedInJobManager from src.linkedIn_job_manager import LinkedInJobManager
from selenium.common.exceptions import NoSuchElementException from selenium.common.exceptions import NoSuchElementException
from loguru import logger
@pytest.fixture @pytest.fixture
@ -52,8 +53,7 @@ def test_set_parameters(mocker, job_manager):
def next_job_page(self, position, location, job_page): def next_job_page(self, position, location, job_page):
logger.debug("Navigating to next job page: %s in %s, page %d", logger.debug(f"Navigating to next job page: {position} in {location}, page {job_page}")
position, location, job_page)
self.driver.get( self.driver.get(
f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}&location={location}&start={job_page * 25}") f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}&location={location}&start={job_page * 25}")