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

Merge pull request #318 from queukat/v3

Browse source
This commit is contained in:
Federico 2024-09-09 00:44:33 +02:00 • committed by GitHub
commit 02e0185d75
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 294 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

6
data_folder/config.yaml
View file

@ -35,13 +35,17 @@ applyOnceAtCompany: [true/false]
distance: 100 distance: 100
companyBlacklist: company_blacklist:
- Company1 - Company1
- Company2 - Company2
titleBlacklist: titleBlacklist:
- word1 - word1
- word2 - word2
job_applicants_threshold:
min_applicants: 0
max_applicants: 100
llm_model_type: openai llm_model_type: openai
llm_model: gpt-4o llm_model: gpt-4o

26
src/gpt.py
View file

@ -3,11 +3,11 @@ import os
import re import re
import textwrap import textwrap
import time import time
from datetime import datetime
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import Dict, List, Union from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Dict, List from typing import Dict, List
from typing import Union
import httpx import httpx
from Levenshtein import distance from Levenshtein import distance
@ -16,39 +16,42 @@ from langchain_core.messages.ai import AIMessage
from langchain_core.output_parsers import StrOutputParser from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompt_values import StringPromptValue from langchain_core.prompt_values import StringPromptValue
from langchain_core.prompts import ChatPromptTemplate from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import src.strings as strings import src.strings as strings
from src.utils import logger from src.utils import logger
load_dotenv() load_dotenv()
class AIModel(ABC): class AIModel(ABC):
@abstractmethod @abstractmethod
def invoke(self, prompt: str) -> str: def invoke(self, prompt: str) -> str:
pass pass
class OpenAIModel(AIModel): class OpenAIModel(AIModel):
def __init__(self, api_key: str, llm_model: str, llm_api_url: str): def __init__(self, api_key: str, llm_model: str, llm_api_url: str):
from langchain_openai import ChatOpenAI from langchain_openai import ChatOpenAI
self.model = ChatOpenAI(model_name=llm_model, openai_api_key=api_key, self.model = ChatOpenAI(model_name=llm_model, openai_api_key=api_key,
temperature=0.4, base_url=llm_api_url) temperature=0.4, base_url=llm_api_url)
def invoke(self, prompt: str) -> str: def invoke(self, prompt: str) -> str:
print("invoke in openai") print("invoke in openai")
response = self.model.invoke(prompt) response = self.model.invoke(prompt)
return response return response
class ClaudeModel(AIModel): class ClaudeModel(AIModel):
def __init__(self, api_key: str, llm_model: str, llm_api_url: str): def __init__(self, api_key: str, llm_model: str, llm_api_url: str):
from langchain_anthropic import ChatAnthropic from langchain_anthropic import ChatAnthropic
self.model = ChatAnthropic(model=llm_model, api_key=api_key, self.model = ChatAnthropic(model=llm_model, api_key=api_key,
temperature=0.4, base_url=llm_api_url) temperature=0.4, base_url=llm_api_url)
def invoke(self, prompt: str) -> str: def invoke(self, prompt: str) -> str:
response = self.model.invoke(prompt) response = self.model.invoke(prompt)
return response return response
class OllamaModel(AIModel): class OllamaModel(AIModel):
def __init__(self, api_key: str, llm_model: str, llm_api_url: str): def __init__(self, api_key: str, llm_model: str, llm_api_url: str):
from langchain_ollama import ChatOllama from langchain_ollama import ChatOllama
@ -58,6 +61,7 @@ class OllamaModel(AIModel):
response = self.model.invoke(prompt) response = self.model.invoke(prompt)
return response return response
class AIAdapter: class AIAdapter:
def __init__(self, config: dict, api_key: str): def __init__(self, config: dict, api_key: str):
self.model = self._create_model(config, api_key) self.model = self._create_model(config, api_key)
@ -67,7 +71,7 @@ class AIAdapter:
llm_model = config['llm_model'] llm_model = config['llm_model']
llm_api_url = config['llm_api_url'] llm_api_url = config['llm_api_url']
print('Using {0} with {1} from {2}'.format(llm_model_type, llm_model, llm_api_url)) print('Using {0} with {1} from {2}'.format(llm_model_type, llm_model, llm_api_url))
if llm_model_type == "openai": if llm_model_type == "openai":
return OpenAIModel(api_key, llm_model, llm_api_url) return OpenAIModel(api_key, llm_model, llm_api_url)
elif llm_model_type == "claude": elif llm_model_type == "claude":
@ -80,9 +84,9 @@ class AIAdapter:
def invoke(self, prompt: str) -> str: def invoke(self, prompt: str) -> str:
return self.model.invoke(prompt) return self.model.invoke(prompt)
class LLMLogger: class LLMLogger:
def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]): def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]):
self.llm = llm self.llm = llm
@ -189,7 +193,6 @@ class LLMLogger:
class LoggerChatModel: class LoggerChatModel:
def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]): def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]):
self.llm = llm self.llm = llm
@ -247,7 +250,6 @@ class LoggerChatModel:
time.sleep(30) time.sleep(30)
continue continue
def parse_llmresult(self, llmresult: AIMessage) -> Dict[str, Dict]: def parse_llmresult(self, llmresult: AIMessage) -> Dict[str, Dict]:
logger.debug("Parsing LLM result: %s", llmresult) logger.debug("Parsing LLM result: %s", llmresult)
@ -454,12 +456,14 @@ class GPTAnswerer:
chain = prompt | self.llm_cheap | StrOutputParser() chain = prompt | self.llm_cheap | StrOutputParser()
output = chain.invoke({"question": question}) output = chain.invoke({"question": question})
match = re.search(r"(Personal information|Self Identification|Legal Authorization|Work Preferences|Education Details|Experience Details|Projects|Availability|Salary Expectations|Certifications|Languages|Interests|Cover letter)", output, re.IGNORECASE) match = re.search(
r"(Personal information|Self Identification|Legal Authorization|Work Preferences|Education Details|Experience Details|Projects|Availability|Salary Expectations|Certifications|Languages|Interests|Cover letter)",
output, re.IGNORECASE)
if not match: if not match:
raise ValueError("Could not extract section name from the response.") raise ValueError("Could not extract section name from the response.")
section_name = match.group(1).lower().replace(" ", "_") section_name = match.group(1).lower().replace(" ", "_")
if section_name == "cover_letter": if section_name == "cover_letter":
chain = chains.get(section_name) chain = chains.get(section_name)
output = chain.invoke({"resume": self.resume, "job_description": self.job_description}) output = chain.invoke({"resume": self.resume, "job_description": self.job_description})

77
src/linkedIn_easy_applier.py
View file

@ -8,7 +8,7 @@ import traceback
from typing import List, Optional, Any, Tuple from typing import List, Optional, Any, Tuple
from httpx import HTTPStatusError from httpx import HTTPStatusError
from reportlab.lib.pagesizes import letter from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas from reportlab.pdfgen import canvas
from selenium.common.exceptions import NoSuchElementException, TimeoutException from selenium.common.exceptions import NoSuchElementException, TimeoutException
from selenium.webdriver import ActionChains from selenium.webdriver import ActionChains
@ -37,7 +37,6 @@ class LinkedInEasyApplier:
logger.debug("LinkedInEasyApplier initialized successfully") logger.debug("LinkedInEasyApplier initialized successfully")
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("Loading questions from JSON file: %s", output_file)
@ -60,10 +59,8 @@ class LinkedInEasyApplier:
logger.error("Error loading questions data from JSON file: %s", tb_str) logger.error("Error loading questions data from JSON file: %s", 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):
"""Проверяет, был ли выполнен редирект на страницу LinkedIn Premium.
В случае редиректа возвращает пользователя на исходную страницу вакансии."""
current_url = self.driver.current_url current_url = self.driver.current_url
attempts = 0 attempts = 0
@ -80,7 +77,6 @@ class LinkedInEasyApplier:
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.")
def job_apply(self, job: Any): def job_apply(self, job: Any):
logger.debug("Starting job application for job: %s", job) logger.debug("Starting job application for job: %s", job)
@ -167,7 +163,7 @@ class LinkedInEasyApplier:
logger.debug(f"Attempting search using {method['description']}") logger.debug(f"Attempting search using {method['description']}")
if method.get('find_elements'): if method.get('find_elements'):
# Поиск всех кнопок "Easy Apply"
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):
@ -209,7 +205,6 @@ class LinkedInEasyApplier:
logger.error("No clickable 'Easy Apply' button found after 2 attempts. Page source:\n%s", page_source) logger.error("No clickable 'Easy Apply' button found after 2 attempts. Page source:\n%s", 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:
logger.debug("Getting job description") logger.debug("Getting job description")
try: try:
@ -514,11 +509,47 @@ class LinkedInEasyApplier:
file_path_pdf = os.path.join(folder_path, f"Cover_Letter_{timestamp}.pdf") file_path_pdf = os.path.join(folder_path, f"Cover_Letter_{timestamp}.pdf")
logger.debug(f"Generated file path for cover letter: {file_path_pdf}") logger.debug(f"Generated file path for cover letter: {file_path_pdf}")
c = canvas.Canvas(file_path_pdf, pagesize=letter) c = canvas.Canvas(file_path_pdf, pagesize=A4)
_, height = letter page_width, page_height = A4
text_object = c.beginText(100, height - 100) text_object = c.beginText(50, page_height - 50)
text_object.setFont("Helvetica", 12) text_object.setFont("Helvetica", 12)
text_object.textLines(cover_letter_text)
max_width = page_width - 100
bottom_margin = 50
available_height = page_height - bottom_margin - 50
def split_text_by_width(text, font, font_size, max_width):
wrapped_lines = []
for line in text.splitlines():
if utils.stringWidth(line, font, font_size) > max_width:
words = line.split()
new_line = ""
for word in words:
if utils.stringWidth(new_line + word + " ", font, font_size) <= max_width:
new_line += word + " "
else:
wrapped_lines.append(new_line.strip())
new_line = word + " "
wrapped_lines.append(new_line.strip())
else:
wrapped_lines.append(line)
return wrapped_lines
lines = split_text_by_width(cover_letter_text, "Helvetica", 12, max_width)
for line in lines:
text_height = text_object.getY()
if text_height > bottom_margin:
text_object.textLine(line)
else:
c.drawText(text_object)
c.showPage()
text_object = c.beginText(50, page_height - 50)
text_object.setFont("Helvetica", 12)
text_object.textLine(line)
c.drawText(text_object) c.drawText(text_object)
c.save() c.save()
logger.debug(f"Cover letter successfully generated and saved to: {file_path_pdf}") logger.debug(f"Cover letter successfully generated and saved to: {file_path_pdf}")
@ -632,7 +663,6 @@ class LinkedInEasyApplier:
for item in self.all_data: for item in self.all_data:
logger.debug( logger.debug(
f"Comparing sanitized stored question: '{self._sanitize_text(item['question'])}' and type: '{item.get('type')}' with current question: '{self._sanitize_text(question_text)}' and type: '{question_type}'") f"Comparing sanitized stored question: '{self._sanitize_text(item['question'])}' and type: '{item.get('type')}' with current question: '{self._sanitize_text(question_text)}' and type: '{question_type}'")
@ -659,7 +689,6 @@ class LinkedInEasyApplier:
answer = self.gpt_answerer.answer_question_textual_wide_range(question_text) answer = self.gpt_answerer.answer_question_textual_wide_range(question_text)
logger.debug(f"Generated textual answer: {answer}") logger.debug(f"Generated textual answer: {answer}")
self._save_questions_to_json({'type': question_type, 'question': question_text, 'answer': answer}) self._save_questions_to_json({'type': question_type, 'question': question_text, 'answer': answer})
self._enter_text(text_field, answer) self._enter_text(text_field, answer)
logger.debug("Entered new answer into the textbox and saved it to JSON.") logger.debug("Entered new answer into the textbox and saved it to JSON.")
@ -692,7 +721,6 @@ class LinkedInEasyApplier:
logger.debug("Entered existing date answer") logger.debug("Entered existing date answer")
return True return True
self._save_questions_to_json({'type': 'date', 'question': question_text, 'answer': answer_text}) self._save_questions_to_json({'type': 'date', 'question': question_text, 'answer': answer_text})
self._enter_text(date_field, answer_text) self._enter_text(date_field, answer_text)
logger.debug("Entered new date answer") logger.debug("Entered new date answer")
@ -701,12 +729,12 @@ class LinkedInEasyApplier:
def _find_and_handle_dropdown_question(self, section: WebElement) -> bool: def _find_and_handle_dropdown_question(self, section: WebElement) -> bool:
try: try:
question = section.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element') question = section.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element')
question_text = question.find_element(By.TAG_NAME, 'label').text.lower()
logger.debug(f"Processing dropdown or combobox question: {question_text}")
dropdowns = question.find_elements(By.TAG_NAME, 'select') dropdowns = question.find_elements(By.TAG_NAME, 'select')
if not dropdowns:
dropdowns = section.find_elements(By.CSS_SELECTOR, '[data-test-text-entity-list-form-select]')
if dropdowns: if dropdowns:
dropdown = dropdowns[0] dropdown = dropdowns[0]
select = Select(dropdown) select = Select(dropdown)
@ -714,6 +742,9 @@ class LinkedInEasyApplier:
logger.debug(f"Dropdown options found: {options}") logger.debug(f"Dropdown options found: {options}")
question_text = question.find_element(By.TAG_NAME, 'label').text.lower()
logger.debug(f"Processing dropdown or combobox 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}")
@ -738,9 +769,15 @@ class LinkedInEasyApplier:
logger.debug(f"Selected new dropdown answer: {answer}") logger.debug(f"Selected new dropdown answer: {answer}")
return True return True
return False else:
logger.debug(f"No dropdown found. Logging elements for debugging.")
elements = section.find_elements(By.XPATH, ".//*")
logger.debug(f"Elements found: {[element.tag_name for element in elements]}")
return False
except Exception as e: except Exception as e:
logger.warning(f"Failed to handle dropdown or combobox question: {e}") logger.warning(f"Failed to handle dropdown or combobox question: {e}", exc_info=True)
return False return False
def _is_numeric_field(self, field: WebElement) -> bool: def _is_numeric_field(self, field: WebElement) -> bool:

165
src/linkedIn_job_manager.py
View file

@ -5,6 +5,7 @@ import time
from itertools import product from itertools import product
from pathlib import Path from pathlib import Path
from inputimeout import inputimeout, TimeoutOccurred
from selenium.common.exceptions import NoSuchElementException from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By from selenium.webdriver.common.by import By
@ -45,13 +46,18 @@ class LinkedInJobManager:
def set_parameters(self, parameters): def set_parameters(self, parameters):
logger.debug("Setting parameters for LinkedInJobManager") logger.debug("Setting parameters for LinkedInJobManager")
self.company_blacklist = parameters.get('companyBlacklist', []) or [] self.company_blacklist = parameters.get('company_blacklist', []) or []
self.title_blacklist = parameters.get('titleBlacklist', []) or [] self.title_blacklist = parameters.get('titleBlacklist', []) or []
self.positions = parameters.get('positions', []) self.positions = parameters.get('positions', [])
self.locations = parameters.get('locations', []) self.locations = parameters.get('locations', [])
self.apply_once_at_company = parameters.get('applyOnceAtCompany', False) self.apply_once_at_company = parameters.get('applyOnceAtCompany', False)
self.base_search_url = self.get_base_search_url(parameters) self.base_search_url = self.get_base_search_url(parameters)
self.seen_jobs = [] self.seen_jobs = []
job_applicants_threshold = parameters.get('job_applicants_threshold', {})
self.min_applicants = job_applicants_threshold.get('min_applicants', 0)
self.max_applicants = job_applicants_threshold.get('max_applicants', float('inf'))
resume_path = parameters.get('uploads', {}).get('resume', None) resume_path = parameters.get('uploads', {}).get('resume', None)
self.resume_path = Path(resume_path) if resume_path and Path(resume_path).exists() else None self.resume_path = Path(resume_path) if resume_path and Path(resume_path).exists() else None
self.output_file_directory = Path(parameters['outputFileDirectory']) self.output_file_directory = Path(parameters['outputFileDirectory'])
@ -109,32 +115,80 @@ class LinkedInJobManager:
utils.printyellow("Applying to jobs on this page has been completed!") utils.printyellow("Applying to jobs on this page has been completed!")
time_left = minimum_page_time - time.time() time_left = minimum_page_time - time.time()
# Ask user if they want to skip waiting, with timeout
if time_left > 0: if time_left > 0:
utils.printyellow(f"Sleeping for {time_left} seconds.") try:
logger.debug("Sleeping for %d seconds", time_left) user_input = inputimeout(
time.sleep(time_left) prompt=f"Sleeping for {time_left} seconds. Press 'y' to skip waiting. Timeout 10 seconds : ",
minimum_page_time = time.time() + minimum_time timeout=10).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)
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)
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.") try:
logger.debug("Sleeping for %d seconds", sleep_time) user_input = inputimeout(
time.sleep(sleep_time) prompt=f"Sleeping for {sleep_time / 60} minutes. Press 'y' to skip waiting. Timeout 10 seconds : ",
timeout=10).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 {sleep_time} seconds.")
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.")
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("Unexpected error during job search: %s", e)
utils.printred(f"Unexpected error: {e}") utils.printred(f"Unexpected error: {e}")
continue continue
time_left = minimum_page_time - time.time() time_left = minimum_page_time - time.time()
if time_left > 0: if time_left > 0:
utils.printyellow(f"Sleeping for {time_left} seconds.") try:
logger.debug("Sleeping for %d seconds", time_left) user_input = inputimeout(
time.sleep(time_left) prompt=f"Sleeping for {time_left} seconds. Press 'y' to skip waiting. Timeout 10 seconds : ",
minimum_page_time = time.time() + minimum_time timeout=10).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)
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)
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.") try:
logger.debug("Sleeping for %d seconds", sleep_time) user_input = inputimeout(
time.sleep(sleep_time) prompt=f"Sleeping for {sleep_time / 60} minutes. Press 'y' to skip waiting: ",
timeout=10).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 {sleep_time} seconds.")
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.")
time.sleep(sleep_time)
page_sleep += 1 page_sleep += 1
def get_jobs_from_page(self): def get_jobs_from_page(self):
@ -183,16 +237,82 @@ class LinkedInJobManager:
pass pass
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, moving to next page.") 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
job_list = [Job(*self.extract_job_information_from_tile(job_element)) for job_element in job_list_elements] job_list = [Job(*self.extract_job_information_from_tile(job_element)) for job_element in job_list_elements]
for job in job_list: for job in job_list:
try:
logger.debug(f"Starting applicant count search for job: {job.title} at {job.company}")
# Find all job insight elements
job_insight_elements = self.driver.find_elements(By.CLASS_NAME,
"job-details-jobs-unified-top-card__job-insight")
logger.debug(f"Found {len(job_insight_elements)} job insight elements")
# Initialize applicants_count as None
applicants_count = None
# Iterate over each job insight element to find the one containing the word "applicant"
for element in job_insight_elements:
logger.debug(f"Checking element text: {element.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}")
# Extract numeric digits from the text (e.g., "70 applicants" -> "70")
applicants_count = ''.join(filter(str.isdigit, applicants_text))
logger.debug(f"Extracted applicants count: {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
break
# Check if applicants_count is valid (not None) before performing comparisons
if applicants_count is not None:
# Perform the threshold check for applicants count
if applicants_count < self.min_applicants or applicants_count > self.max_applicants:
utils.printyellow(
f"Skipping {job.title} at {job.company} due to applicants count: {applicants_count}")
logger.debug(f"Skipping {job.title} at {job.company}, applicants count: {applicants_count}")
self.write_to_file(job, "skipped_due_to_applicants")
continue # Skip this job if applicants count is outside the threshold
else:
logger.debug(f"Applicants count {applicants_count} is within the threshold")
else:
# If no applicants count was found, log a warning but continue the process
logger.warning(
f"Applicants count not found for {job.title} at {job.company}, continuing with application.")
except NoSuchElementException:
# Log a warning if the job insight elements are not found, but do not stop the job application process
logger.warning(
f"Applicants count elements not found for {job.title} at {job.company}, continuing with application.")
except ValueError as e:
# Handle errors when parsing the applicants count
logger.error(f"Error parsing applicants count for {job.title} at {job.company}: {e}")
except Exception as e:
# Catch any other exceptions to ensure the process continues
logger.error(
f"Unexpected error during applicants count processing for {job.title} at {job.company}: {e}")
# Continue with the job application process regardless of the applicants count check
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...") utils.printyellow(f"Blacklisted {job.title} at {job.company}, skipping...")
logger.debug("Job blacklisted: %s at %s", job.title, job.company) logger.debug("Job blacklisted: %s at %s", job.title, job.company)
@ -200,7 +320,7 @@ class LinkedInJobManager:
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):
self.write_to_file(job, "skipped") self.write_to_file(job, "skipped")
continue continue
if self.is_already_applied_to_company(job.company): if self.is_already_applied_to_company(job.company):
self.write_to_file(job, "skipped") self.write_to_file(job, "skipped")
continue continue
@ -307,7 +427,6 @@ class LinkedInJobManager:
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("Job blacklisted status: %s", is_blacklisted)
return is_blacklisted return is_blacklisted
@ -322,8 +441,8 @@ class LinkedInJobManager:
def is_already_applied_to_company(self, company): def is_already_applied_to_company(self, company):
if not self.apply_once_at_company: if not self.apply_once_at_company:
return False return False
output_files = ["success.json"] output_files = ["success.json"]
for file_name in output_files: for file_name in output_files:
file_path = self.output_file_directory / file_name file_path = self.output_file_directory / file_name
@ -333,9 +452,9 @@ 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(f"Already applied at {company} (once per company policy), skipping...") utils.printyellow(
f"Already applied at {company} (once per company policy), skipping...")
return True return True
except json.JSONDecodeError: except json.JSONDecodeError:
continue continue
return False return False

139
src/linkedin-api.py
View file

@ -1,58 +1,59 @@
from typing import Dict, List
from linkedin_api import Linkedin
from typing import Optional, Union, Literal
from urllib.parse import quote, urlencode
import logging import logging
import json from typing import Dict, List
from typing import Optional, Union, Literal
from urllib.parse import urlencode
from linkedin_api import Linkedin
# 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] = []
def __init__(self, username, password): def __init__(self, username, password):
super().__init__(username, password) super().__init__(username, password)
def search_jobs( def search_jobs(
self, self,
keywords: Optional[str] = None, keywords: Optional[str] = None,
companies: Optional[List[str]] = None, companies: Optional[List[str]] = None,
experience: Optional[ experience: Optional[
List[ List[
Union[ Union[
Literal["1"], Literal["1"],
Literal["2"], Literal["2"],
Literal["3"], Literal["3"],
Literal["4"], Literal["4"],
Literal["5"], Literal["5"],
Literal["6"], Literal["6"],
]
] ]
] ] = None,
] = None, job_type: Optional[
job_type: Optional[ List[
List[ Union[
Union[ Literal["F"],
Literal["F"], Literal["C"],
Literal["C"], Literal["P"],
Literal["P"], Literal["T"],
Literal["T"], Literal["I"],
Literal["I"], Literal["V"],
Literal["V"], Literal["O"],
Literal["O"], ]
] ]
] ] = None,
] = None, job_title: Optional[List[str]] = None,
job_title: Optional[List[str]] = None, industries: Optional[List[str]] = None,
industries: Optional[List[str]] = None, location_name: Optional[str] = None,
location_name: Optional[str] = None, remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None,
remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None, listed_at: None | int = None,
listed_at: None | int = None, distance: Optional[int] = None,
distance: Optional[int] = None, easy_apply: Optional[bool] = True,
easy_apply: Optional[bool] = True, limit=-1,
limit=-1, offset=0,
offset=0, **kwargs,
**kwargs,
) -> List[Dict]: ) -> List[Dict]:
"""Perform a LinkedIn search for jobs. """Perform a LinkedIn search for jobs.
@ -154,21 +155,21 @@ class LinkedInEvolvedAPI(Linkedin):
e["job_id"] = trackingUrn e["job_id"] = trackingUrn
if e.get("$type") == "com.linkedin.voyager.dash.jobs.JobPosting": if e.get("$type") == "com.linkedin.voyager.dash.jobs.JobPosting":
new_data.append(e) new_data.append(e)
if not new_data: if not new_data:
break break
results.extend(new_data) results.extend(new_data)
if ( if (
(-1 < limit <= len(results)) (-1 < limit <= len(results))
or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS
) or len(elements) == 0: ) or len(elements) == 0:
break break
self.logger.debug(f"results grew to {len(results)}") self.logger.debug(f"results grew to {len(results)}")
return results return results
def get_fields_for_easy_apply(self,job_id: str) -> List[Dict]: def get_fields_for_easy_apply(self, job_id: str) -> List[Dict]:
"""Get fields needed for easy apply jobs. """Get fields needed for easy apply jobs.
:param job_id: Job ID :param job_id: Job ID
@ -181,14 +182,12 @@ class LinkedInEvolvedAPI(Linkedin):
cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()]) cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
headers: Dict[str, str] = self._headers() headers: Dict[str, str] = self._headers()
headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1" headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1"
headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "") headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "")
headers["Cookie"] = cookie_str headers["Cookie"] = cookie_str
headers["Connection"] = "keep-alive" headers["Connection"] = "keep-alive"
default_params = { default_params = {
"decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67", "decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67",
"jobPostingUrn": f"urn:li:fsd_jobPosting:{job_id}", "jobPostingUrn": f"urn:li:fsd_jobPosting:{job_id}",
@ -217,26 +216,26 @@ class LinkedInEvolvedAPI(Linkedin):
except ValueError: except ValueError:
self.logger.error("Failed to parse JSON response") self.logger.error("Failed to parse JSON response")
return [] return []
form_components = [] form_components = []
for item in data.get("included", []): for item in data.get("included", []):
if 'formComponent' in item: if 'formComponent' in item:
urn = item['urn'] urn = item['urn']
try: try:
title = item['title']['text'] title = item['title']['text']
except TypeError: except TypeError:
title = urn title = urn
form_component_type = list(item['formComponent'].keys())[0] form_component_type = list(item['formComponent'].keys())[0]
form_component_details = item['formComponent'][form_component_type] form_component_details = item['formComponent'][form_component_type]
component_info = { component_info = {
'title': title, 'title': title,
'urn': urn, 'urn': urn,
'formComponentType': form_component_type, 'formComponentType': form_component_type,
} }
if 'textSelectableOptions' in form_component_details: if 'textSelectableOptions' in form_component_details:
options = [ options = [
opt['optionText']['text'] for opt in form_component_details['textSelectableOptions'] opt['optionText']['text'] for opt in form_component_details['textSelectableOptions']
@ -244,18 +243,18 @@ class LinkedInEvolvedAPI(Linkedin):
component_info['selectableOptions'] = options component_info['selectableOptions'] = options
elif 'selectableOptions' in form_component_details: elif 'selectableOptions' in form_component_details:
options = [ options = [
opt['textSelectableOption']['optionText']['text'] opt['textSelectableOption']['optionText']['text']
for opt in form_component_details['selectableOptions'] for opt in form_component_details['selectableOptions']
] ]
component_info['selectableOptions'] = options component_info['selectableOptions'] = options
form_components.append(component_info) form_components.append(component_info)
return form_components return form_components
def apply_to_job(self,job_id: str, fields: dict, followCompany: bool = True) -> bool: def apply_to_job(self, job_id: str, fields: dict, followCompany: bool = True) -> bool:
return False return False
# ToDo: Implement apply to job parser first # ToDo: Implement apply to job parser first
# How need to be implemented: # How need to be implemented:
# 1. Get fields for easy apply job from the previous method (get_fields_for_easy_apply) # 1. Get fields for easy apply job from the previous method (get_fields_for_easy_apply)
@ -265,11 +264,11 @@ class LinkedInEvolvedAPI(Linkedin):
# {'title': 'Quanti anni di esperienza di lavoro hai con Router?', 'urn': 'urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4013860791,9478711764,numeric)', 'formComponentType': 'singleLineTextFormComponent', 'response': '5'} # {'title': 'Quanti anni di esperienza di lavoro hai con Router?', 'urn': 'urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4013860791,9478711764,numeric)', 'formComponentType': 'singleLineTextFormComponent', 'response': '5'}
# To fill, you can temporary use input() function to get the data from the user manually for testing purposes (for the further implementation, the question will be asked to AI implementation and automatically filled) # To fill, you can temporary use input() function to get the data from the user manually for testing purposes (for the further implementation, the question will be asked to AI implementation and automatically filled)
# Build a working payload. # Build a working payload.
# EXAMPLE OF WORKING PAYLOAD # EXAMPLE OF WORKING PAYLOAD
# 4005350454 is job_id, so need to be replaced with the job_id # 4005350454 is job_id, so need to be replaced with the job_id
#{ # {
# "followCompany": true, # "followCompany": true,
# "responses": [ # "responses": [
# { # {
@ -349,23 +348,19 @@ class LinkedInEvolvedAPI(Linkedin):
# } # }
# ], # ],
# "trackingId": "" # "trackingId": ""
#} # }
# Push the commit to the repository and create a pull request to the v3 branch. # Push the commit to the repository and create a pull request to the v3 branch.
def set_job_as_applied(self, job_id: str) -> None: def set_job_as_applied(self, job_id: str) -> None:
self.already_applied_jobs.append(job_id) self.already_applied_jobs.append(job_id)
## EXAMPLE USAGE ## EXAMPLE USAGE
if __name__ == "__main__": if __name__ == "__main__":
api: LinkedInEvolvedAPI = LinkedInEvolvedAPI(username="", password="") api: LinkedInEvolvedAPI = LinkedInEvolvedAPI(username="", password="")
jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=100, easy_apply=True, offset=1, listed_at=None) jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=100, easy_apply=True, offset=1,
listed_at=None)
for job in jobs: for job in jobs:
job_id: str = job["job_id"] job_id: str = job["job_id"]
print(f"Job ID: {job_id}") print(f"Job ID: {job_id}")
@ -379,7 +374,3 @@ if __name__ == "__main__":
for field in fields: for field in fields:
print(field) print(field)
break break

10
src/utils.py
View file

@ -90,7 +90,13 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse
return return
position = start position = start
previous_position = None # Tracking the previous position to avoid duplicate scrolls
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:
# Avoid re-scrolling to the same position
logger.debug("Stopping scroll as position hasn't changed: %d", position)
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("Scrolled to position: %d", position)
@ -98,11 +104,15 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse
logger.error("Error during scrolling: %s", e) logger.error("Error during scrolling: %s", e)
print(f"Error during scrolling: {e}") print(f"Error during scrolling: {e}")
previous_position = position
position += step position += step
# Decrease the step but ensure it doesn't reverse direction
step = max(10, abs(step) - 10) * (-1 if reverse else 1) step = max(10, abs(step) - 10) * (-1 if reverse else 1)
time.sleep(random.uniform(0.6, 1.5)) time.sleep(random.uniform(0.6, 1.5))
# 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("Scrolled to final position: %d", end)
time.sleep(0.5) time.sleep(0.5)