first commit v2
This commit is contained in:
feder-cr
parent
2dcd7943e8
commit
d8b7e7fda6
18 changed files with 663 additions and 1398 deletions
|
|
@ -1,12 +1,12 @@
|
|||
import base64
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import tempfile
|
||||
import time
|
||||
import traceback
|
||||
from datetime import date
|
||||
from typing import List, Optional, Any, Tuple
|
||||
import uuid
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from reportlab.pdfgen import canvas
|
||||
from selenium.common.exceptions import NoSuchElementException
|
||||
|
|
@ -15,27 +15,18 @@ from selenium.webdriver.common.keys import Keys
|
|||
from selenium.webdriver.remote.webelement import WebElement
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
from selenium.webdriver.support.ui import Select, WebDriverWait
|
||||
import tempfile
|
||||
import time
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from reportlab.pdfgen import canvas
|
||||
import io
|
||||
import time
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from reportlab.pdfgen import canvas
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from xhtml2pdf import pisa
|
||||
|
||||
import utils
|
||||
from selenium.webdriver import ActionChains
|
||||
import utils
|
||||
|
||||
class LinkedInEasyApplier:
|
||||
def __init__(self, driver: Any, resume_dir: Optional[str], set_old_answers: List[Tuple[str, str, str]], gpt_answerer: Any):
|
||||
def __init__(self, driver: Any, resume_dir: Optional[str], set_old_answers: List[Tuple[str, str, str]], gpt_answerer: Any, resume_generator_manager):
|
||||
if resume_dir is None or not os.path.exists(resume_dir):
|
||||
resume_dir = None
|
||||
self.driver = driver
|
||||
self.resume_dir = resume_dir
|
||||
self.set_old_answers = set_old_answers
|
||||
self.gpt_answerer = gpt_answerer
|
||||
self.resume_generator_manager = resume_generator_manager
|
||||
|
||||
def job_apply(self, job: Any):
|
||||
self.driver.get(job.link)
|
||||
|
|
@ -44,55 +35,64 @@ class LinkedInEasyApplier:
|
|||
easy_apply_button = self._find_easy_apply_button()
|
||||
job_description = self._get_job_description()
|
||||
job.set_job_description(job_description)
|
||||
easy_apply_button.click()
|
||||
actions = ActionChains(self.driver)
|
||||
actions.move_to_element(easy_apply_button).click().perform()
|
||||
self.gpt_answerer.set_job(job)
|
||||
self._fill_application_form()
|
||||
self._fill_application_form(job)
|
||||
except Exception:
|
||||
tb_str = traceback.format_exc()
|
||||
self._discard_application()
|
||||
raise Exception(f"Failed to apply to job! Original exception: \nTraceback:\n{tb_str}")
|
||||
|
||||
|
||||
def _find_easy_apply_button(self) -> WebElement:
|
||||
buttons = WebDriverWait(self.driver, 10).until(
|
||||
EC.presence_of_all_elements_located(
|
||||
(By.XPATH, '//button[contains(@class, "jobs-apply-button") and contains(., "Easy Apply")]')
|
||||
)
|
||||
)
|
||||
for index, button in enumerate(buttons):
|
||||
try:
|
||||
return WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.XPATH, f'(//button[contains(@class, "jobs-apply-button") and contains(., "Easy Apply")])[{index + 1}]')
|
||||
)
|
||||
attempt = 0
|
||||
while attempt < 2:
|
||||
self._scroll_page()
|
||||
buttons = WebDriverWait(self.driver, 10).until(
|
||||
EC.presence_of_all_elements_located(
|
||||
(By.XPATH, '//button[contains(@class, "jobs-apply-button") and contains(., "Easy Apply")]')
|
||||
)
|
||||
except Exception as e:
|
||||
pass
|
||||
)
|
||||
for index, _ in enumerate(buttons):
|
||||
try:
|
||||
button = WebDriverWait(self.driver, 10).until(
|
||||
EC.element_to_be_clickable(
|
||||
(By.XPATH, f'(//button[contains(@class, "jobs-apply-button") and contains(., "Easy Apply")])[{index + 1}]')
|
||||
)
|
||||
)
|
||||
return button
|
||||
except Exception as e:
|
||||
pass
|
||||
if attempt == 0:
|
||||
self.driver.refresh()
|
||||
time.sleep(3)
|
||||
attempt += 1
|
||||
raise Exception("No clickable 'Easy Apply' button found")
|
||||
|
||||
|
||||
def _get_job_description(self) -> str:
|
||||
try:
|
||||
see_more_button = self.driver.find_element(By.XPATH, '//button[@aria-label="Click to see more description"]')
|
||||
see_more_button.click()
|
||||
actions = ActionChains(self.driver)
|
||||
actions.move_to_element(see_more_button).click().perform()
|
||||
time.sleep(2)
|
||||
description = self.driver.find_element(By.CLASS_NAME, 'jobs-description-content__text').text
|
||||
self._scroll_page()
|
||||
return description
|
||||
except NoSuchElementException:
|
||||
tb_str = traceback.format_exc()
|
||||
raise Exception("Job description 'See more' button not found: \nTraceback:\n{tb_str}")
|
||||
except Exception :
|
||||
except Exception:
|
||||
tb_str = traceback.format_exc()
|
||||
raise Exception(f"Error getting Job description: \nTraceback:\n{tb_str}")
|
||||
|
||||
def _scroll_page(self) -> None:
|
||||
scrollable_element = self.driver.find_element(By.TAG_NAME, 'html')
|
||||
utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=False)
|
||||
utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=True)
|
||||
#utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=False)
|
||||
#utils.scroll_slow(self.driver, scrollable_element, step=300, reverse=True)
|
||||
|
||||
def _fill_application_form(self):
|
||||
def _fill_application_form(self, job):
|
||||
while True:
|
||||
self.fill_up()
|
||||
self.fill_up(job)
|
||||
if self._next_or_submit():
|
||||
break
|
||||
|
||||
|
|
@ -110,7 +110,6 @@ class LinkedInEasyApplier:
|
|||
time.sleep(random.uniform(3.0, 5.0))
|
||||
self._check_for_errors()
|
||||
|
||||
|
||||
def _unfollow_company(self) -> None:
|
||||
try:
|
||||
follow_checkbox = self.driver.find_element(
|
||||
|
|
@ -133,88 +132,57 @@ class LinkedInEasyApplier:
|
|||
except Exception as e:
|
||||
pass
|
||||
|
||||
def fill_up(self) -> None:
|
||||
try:
|
||||
easy_apply_content = self.driver.find_element(By.CLASS_NAME, 'jobs-easy-apply-content')
|
||||
pb4_elements = easy_apply_content.find_elements(By.CLASS_NAME, 'pb4')
|
||||
for element in pb4_elements:
|
||||
self._process_form_element(element)
|
||||
except Exception as e:
|
||||
pass
|
||||
def fill_up(self, job) -> None:
|
||||
easy_apply_content = self.driver.find_element(By.CLASS_NAME, 'jobs-easy-apply-content')
|
||||
pb4_elements = easy_apply_content.find_elements(By.CLASS_NAME, 'pb4')
|
||||
for element in pb4_elements:
|
||||
self._process_form_element(element, job)
|
||||
|
||||
|
||||
|
||||
def _process_form_element(self, element: WebElement) -> None:
|
||||
try:
|
||||
if self._is_upload_field(element):
|
||||
self._handle_upload_fields(element)
|
||||
else:
|
||||
self._fill_additional_questions()
|
||||
except Exception as e:
|
||||
pass
|
||||
def _process_form_element(self, element: WebElement, job) -> None:
|
||||
if self._is_upload_field(element):
|
||||
self._handle_upload_fields(element, job)
|
||||
else:
|
||||
self._fill_additional_questions()
|
||||
|
||||
def _is_upload_field(self, element: WebElement) -> bool:
|
||||
try:
|
||||
element.find_element(By.XPATH, ".//input[@type='file']")
|
||||
return True
|
||||
except NoSuchElementException:
|
||||
return False
|
||||
return bool(element.find_elements(By.XPATH, ".//input[@type='file']"))
|
||||
|
||||
def _handle_upload_fields(self, element: WebElement) -> None:
|
||||
def _handle_upload_fields(self, element: WebElement, job) -> None:
|
||||
file_upload_elements = self.driver.find_elements(By.XPATH, "//input[@type='file']")
|
||||
for element in file_upload_elements:
|
||||
parent = element.find_element(By.XPATH, "..")
|
||||
self.driver.execute_script("arguments[0].classList.remove('hidden')", element)
|
||||
if 'resume' in parent.text.lower():
|
||||
if self.resume_dir != None:
|
||||
output = self.gpt_answerer.resume_or_cover(parent.text.lower())
|
||||
if 'resume' in output:
|
||||
if self.resume_dir:
|
||||
resume_path = self.resume_dir.resolve()
|
||||
if self.resume_dir != None and resume_path.exists() and resume_path.is_file():
|
||||
element.send_keys(str(resume_path))
|
||||
else:
|
||||
self._create_and_upload_resume(element)
|
||||
elif 'cover' in parent.text.lower():
|
||||
if resume_path.exists() and resume_path.is_file():
|
||||
element.send_keys(str(resume_path))
|
||||
else:
|
||||
self._create_and_upload_resume(element, job)
|
||||
elif 'cover' in output:
|
||||
self._create_and_upload_cover_letter(element)
|
||||
|
||||
def _create_and_upload_resume(self, element):
|
||||
max_retries = 3
|
||||
retry_delay = 1
|
||||
def _create_and_upload_resume(self, element, job):
|
||||
folder_path = 'generated_cv'
|
||||
|
||||
if not os.path.exists(folder_path):
|
||||
os.makedirs(folder_path)
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
html_string = self.gpt_answerer.get_resume_html()
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.html', mode='w', encoding='utf-8') as temp_html_file:
|
||||
temp_html_file.write(html_string)
|
||||
file_name_HTML = temp_html_file.name
|
||||
|
||||
file_name_pdf = f"resume_{uuid.uuid4().hex}.pdf"
|
||||
file_path_pdf = os.path.join(folder_path, file_name_pdf)
|
||||
|
||||
with open(file_path_pdf, "wb") as f:
|
||||
f.write(base64.b64decode(utils.HTML_to_PDF(file_name_HTML)))
|
||||
|
||||
element.send_keys(os.path.abspath(file_path_pdf))
|
||||
time.sleep(2) # Give some time for the upload process
|
||||
os.remove(file_name_HTML)
|
||||
return True
|
||||
except Exception:
|
||||
if attempt < max_retries - 1:
|
||||
time.sleep(retry_delay)
|
||||
else:
|
||||
tb_str = traceback.format_exc()
|
||||
raise Exception(f"Max retries reached. Upload failed: \nTraceback:\n{tb_str}")
|
||||
|
||||
def _upload_resume(self, element: WebElement) -> None:
|
||||
element.send_keys(str(self.resume_dir))
|
||||
os.makedirs(folder_path, exist_ok=True)
|
||||
try:
|
||||
file_path_pdf = os.path.join(folder_path, f"CV_{random.randint(0, 9999)}.pdf")
|
||||
with open(file_path_pdf, "xb") as f:
|
||||
f.write(base64.b64decode(self.resume_generator_manager.pdf_base64(job_description_text=job.description)))
|
||||
element.send_keys(os.path.abspath(file_path_pdf))
|
||||
job.pdf_path = os.path.abspath(file_path_pdf)
|
||||
time.sleep(2)
|
||||
except Exception:
|
||||
tb_str = traceback.format_exc()
|
||||
raise Exception(f"Upload failed: \nTraceback:\n{tb_str}")
|
||||
|
||||
def _create_and_upload_cover_letter(self, element: WebElement) -> None:
|
||||
cover_letter = self.gpt_answerer.answer_question_textual_wide_range("Write a cover letter")
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as temp_pdf_file:
|
||||
letter_path = temp_pdf_file.name
|
||||
c = canvas.Canvas(letter_path, pagesize=letter)
|
||||
width, height = letter
|
||||
_, height = letter
|
||||
text_object = c.beginText(100, height - 100)
|
||||
text_object.setFont("Helvetica", 12)
|
||||
text_object.textLines(cover_letter)
|
||||
|
|
@ -225,131 +193,84 @@ class LinkedInEasyApplier:
|
|||
def _fill_additional_questions(self) -> None:
|
||||
form_sections = self.driver.find_elements(By.CLASS_NAME, 'jobs-easy-apply-form-section__grouping')
|
||||
for section in form_sections:
|
||||
self._process_question(section)
|
||||
self._process_form_section(section)
|
||||
|
||||
def _process_question(self, section: WebElement) -> None:
|
||||
def _process_form_section(self, section: WebElement) -> None:
|
||||
if self._handle_terms_of_service(section):
|
||||
return
|
||||
self._handle_radio_question(section)
|
||||
self._handle_textbox_question(section)
|
||||
self._handle_date_question(section)
|
||||
self._handle_dropdown_question(section)
|
||||
if self._find_and_handle_radio_question(section):
|
||||
return
|
||||
if self._find_and_handle_textbox_question(section):
|
||||
return
|
||||
if self._find_and_handle_date_question(section):
|
||||
return
|
||||
if self._find_and_handle_dropdown_question(section):
|
||||
return
|
||||
|
||||
def _handle_terms_of_service(self, element: WebElement) -> bool:
|
||||
try:
|
||||
question = element.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element')
|
||||
checkbox = question.find_element(By.TAG_NAME, 'label')
|
||||
question_text = question.text.lower()
|
||||
if 'terms of service' in question_text or 'privacy policy' in question_text or 'terms of use' in question_text:
|
||||
checkbox.click()
|
||||
return True
|
||||
except NoSuchElementException:
|
||||
pass
|
||||
checkbox = element.find_elements(By.TAG_NAME, 'label')
|
||||
if checkbox and any(term in checkbox[0].text.lower() for term in ['terms of service', 'privacy policy', 'terms of use']):
|
||||
checkbox[0].click()
|
||||
return True
|
||||
return False
|
||||
|
||||
def _handle_radio_question(self, element: WebElement) -> None:
|
||||
try:
|
||||
question = element.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element')
|
||||
radios = question.find_elements(By.CLASS_NAME, 'fb-text-selectable__option')
|
||||
if not radios:
|
||||
return
|
||||
|
||||
question_text = element.text.lower()
|
||||
def _find_and_handle_radio_question(self, section: WebElement) -> bool:
|
||||
radios = section.find_elements(By.CLASS_NAME, 'fb-text-selectable__option')
|
||||
if radios:
|
||||
question_text = section.text.lower()
|
||||
options = [radio.text.lower() for radio in radios]
|
||||
|
||||
answer = self._get_answer_from_set('radio', question_text, options)
|
||||
if not answer:
|
||||
answer = self.gpt_answerer.answer_question_from_options(question_text, options)
|
||||
|
||||
answer = self.gpt_answerer.answer_question_from_options(question_text, options)
|
||||
self._select_radio(radios, answer)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _handle_textbox_question(self, element: WebElement) -> None:
|
||||
try:
|
||||
question = element.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element')
|
||||
question_text = question.find_element(By.TAG_NAME, 'label').text.lower()
|
||||
text_field = self._find_text_field(question)
|
||||
|
||||
is_numeric = self._is_numeric_field(text_field)
|
||||
answer = self._get_answer_from_set('numeric' if is_numeric else 'text', question_text)
|
||||
|
||||
if not answer:
|
||||
answer = self.gpt_answerer.answer_question_numeric(question_text) if is_numeric else self.gpt_answerer.answer_question_textual_wide_range(question_text)
|
||||
|
||||
self._enter_text(text_field, answer)
|
||||
self._handle_form_errors(element, question_text, answer, text_field)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _handle_date_question(self, element: WebElement) -> None:
|
||||
try:
|
||||
date_picker = element.find_element(By.CLASS_NAME, 'artdeco-datepicker__input')
|
||||
date_picker.clear()
|
||||
date_picker.send_keys(date.today().strftime("%m/%d/%y"))
|
||||
time.sleep(3)
|
||||
date_picker.send_keys(Keys.RETURN)
|
||||
time.sleep(2)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _handle_dropdown_question(self, element: WebElement) -> None:
|
||||
try:
|
||||
question = element.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element')
|
||||
question_text = question.find_element(By.TAG_NAME, 'label').text.lower()
|
||||
dropdown = question.find_element(By.TAG_NAME, 'select')
|
||||
select = Select(dropdown)
|
||||
options = [option.text for option in select.options]
|
||||
|
||||
answer = self._get_answer_from_set('dropdown', question_text, options)
|
||||
if not answer:
|
||||
answer = self.gpt_answerer.answer_question_from_options(question_text, options)
|
||||
|
||||
self._select_dropdown(dropdown, answer)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _get_answer_from_set(self, question_type: str, question_text: str, options: Optional[List[str]] = None) -> Optional[str]:
|
||||
for entry in self.set_old_answers:
|
||||
if isinstance(entry, tuple) and len(entry) == 3:
|
||||
if entry[0] == question_type and question_text in entry[1].lower():
|
||||
answer = entry[2]
|
||||
return answer if options is None or answer in options else None
|
||||
return None
|
||||
|
||||
def _find_text_field(self, question: WebElement) -> WebElement:
|
||||
try:
|
||||
return question.find_element(By.TAG_NAME, 'input')
|
||||
except NoSuchElementException:
|
||||
return question.find_element(By.TAG_NAME, 'textarea')
|
||||
|
||||
def _is_numeric_field(self, field: WebElement) -> bool:
|
||||
field_type = field.get_attribute('type').lower()
|
||||
if 'numeric' in field_type:
|
||||
return True
|
||||
class_attribute = field.get_attribute("id")
|
||||
return class_attribute and 'numeric' in class_attribute
|
||||
return False
|
||||
|
||||
def _find_and_handle_textbox_question(self, section: WebElement) -> bool:
|
||||
text_fields = section.find_elements(By.TAG_NAME, 'input') + section.find_elements(By.TAG_NAME, 'textarea')
|
||||
if text_fields:
|
||||
text_field = text_fields[0]
|
||||
question_text = section.find_element(By.TAG_NAME, 'label').text.lower()
|
||||
is_numeric = self._is_numeric_field(text_field)
|
||||
answer = self.gpt_answerer.answer_question_numeric(question_text) if is_numeric else self.gpt_answerer.answer_question_textual_wide_range(question_text)
|
||||
self._enter_text(text_field, answer)
|
||||
return True
|
||||
return False
|
||||
|
||||
def _find_and_handle_date_question(self, section: WebElement) -> bool:
|
||||
date_fields = section.find_elements(By.CLASS_NAME, 'artdeco-datepicker__input ')
|
||||
if date_fields:
|
||||
date_field = date_fields[0]
|
||||
answer_date = self.gpt_answerer.answer_question_date()
|
||||
self._enter_text(date_field, answer_date.strftime("%Y-%m-%d"))
|
||||
return True
|
||||
return False
|
||||
|
||||
def _find_and_handle_dropdown_question(self, section: WebElement) -> bool:
|
||||
dropdowns = section.find_elements(By.CLASS_NAME, 'fb-dropdown__select')
|
||||
if dropdowns:
|
||||
dropdown = dropdowns[0]
|
||||
question_text = section.text.lower()
|
||||
select = Select(dropdown)
|
||||
answer = self.gpt_answerer.answer_question_from_options(question_text, [option.text.lower() for option in select.options])
|
||||
self._select_dropdown_option(select, answer)
|
||||
return True
|
||||
return False
|
||||
|
||||
def _is_numeric_field(self, element: WebElement) -> bool:
|
||||
input_type = element.get_attribute('type')
|
||||
return input_type == 'number'
|
||||
|
||||
def _enter_text(self, element: WebElement, text: str) -> None:
|
||||
element.clear()
|
||||
element.send_keys(text)
|
||||
|
||||
def _select_dropdown(self, element: WebElement, text: str) -> None:
|
||||
select = Select(element)
|
||||
select.select_by_visible_text(text)
|
||||
|
||||
def _select_radio(self, radios: List[WebElement], answer: str) -> None:
|
||||
for radio in radios:
|
||||
if answer in radio.text.lower():
|
||||
radio.find_element(By.TAG_NAME, 'label').click()
|
||||
return
|
||||
radios[-1].find_element(By.TAG_NAME, 'label').click()
|
||||
if radio.text.lower() == answer.lower():
|
||||
radio.click()
|
||||
break
|
||||
|
||||
def _handle_form_errors(self, element: WebElement, question_text: str, answer: str, text_field: WebElement) -> None:
|
||||
try:
|
||||
error = element.find_element(By.CLASS_NAME, 'artdeco-inline-feedback--error')
|
||||
error_text = error.text.lower()
|
||||
new_answer = self.gpt_answerer.try_fix_answer(question_text, answer, error_text)
|
||||
self._enter_text(text_field, new_answer)
|
||||
except NoSuchElementException:
|
||||
pass
|
||||
def _select_dropdown_option(self, select: Select, answer: str) -> None:
|
||||
for option in select.options:
|
||||
if option.text.lower() == answer.lower():
|
||||
select.select_by_visible_text(option.text)
|
||||
break
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue