linkedIn_auto_jobs_applier_.../linkedIn_job_manager.py
2024-08-22 19:33:25 +01:00

220 lines
9.7 KiB
Python

import os
import random
import time
import traceback
from itertools import product
from pathlib import Path
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
import utils
from job import Job
from linkedIn_easy_applier import LinkedInEasyApplier
import json
class EnvironmentKeys:
def __init__(self):
self.skip_apply = self._read_env_key_bool("SKIP_APPLY")
self.disable_description_filter = self._read_env_key_bool("DISABLE_DESCRIPTION_FILTER")
@staticmethod
def _read_env_key(key: str) -> str:
return os.getenv(key, "")
@staticmethod
def _read_env_key_bool(key: str) -> bool:
return os.getenv(key) == "True"
class LinkedInJobManager:
def __init__(self, driver):
self.driver = driver
self.set_old_answers = set()
self.easy_applier_component = None
def set_parameters(self, parameters):
self.company_blacklist = parameters.get('companyBlacklist', []) or []
self.title_blacklist = parameters.get('titleBlacklist', []) or []
self.positions = parameters.get('positions', [])
self.locations = parameters.get('locations', [])
self.base_search_url = self.get_base_search_url(parameters)
self.seen_jobs = []
resume_path = parameters.get('uploads', {}).get('resume', None)
if resume_path is not None and Path(resume_path).exists():
self.resume_path = Path(resume_path)
else:
self.resume_path = None
self.output_file_directory = Path(parameters['outputFileDirectory'])
self.env_config = EnvironmentKeys()
#self.old_question()
def set_gpt_answerer(self, gpt_answerer):
self.gpt_answerer = gpt_answerer
def set_resume_generator_manager(self, resume_generator_manager):
self.resume_generator_manager = resume_generator_manager
""" def old_question(self):
self.set_old_answers = {}
file_path = 'data_folder/output/old_Questions.csv'
if os.path.exists(file_path):
with open(file_path, 'r', newline='', encoding='utf-8', errors='ignore') as file:
csv_reader = csv.reader(file, delimiter=',', quotechar='"')
for row in csv_reader:
if len(row) == 3:
answer_type, question_text, answer = row
self.set_old_answers[(answer_type.lower(), question_text.lower())] = answer"""
def start_applying(self):
self.easy_applier_component = LinkedInEasyApplier(self.driver, self.resume_path, self.set_old_answers, self.gpt_answerer, self.resume_generator_manager)
searches = list(product(self.positions, self.locations))
random.shuffle(searches)
page_sleep = 0
minimum_time = 60 * 15
minimum_page_time = time.time() + minimum_time
for position, location in searches:
location_url = "&location=" + location
job_page_number = -1
utils.printyellow(f"Starting the search for {position} in {location}.")
try:
while True:
page_sleep += 1
job_page_number += 1
utils.printyellow(f"Going to job page {job_page_number}")
self.next_job_page(position, location_url, job_page_number)
time.sleep(random.uniform(1.5, 3.5))
utils.printyellow("Starting the application process for this page...")
self.apply_jobs()
utils.printyellow("Applying to jobs on this page has been completed!")
time_left = minimum_page_time - time.time()
if time_left > 0:
utils.printyellow(f"Sleeping for {time_left} seconds.")
time.sleep(time_left)
minimum_page_time = time.time() + minimum_time
if page_sleep % 5 == 0:
sleep_time = random.randint(5, 34)
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.")
time.sleep(sleep_time)
page_sleep += 1
except Exception:
traceback.format_exc()
pass
time_left = minimum_page_time - time.time()
if time_left > 0:
utils.printyellow(f"Sleeping for {time_left} seconds.")
time.sleep(time_left)
minimum_page_time = time.time() + minimum_time
if page_sleep % 5 == 0:
sleep_time = random.randint(50, 90)
utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.")
time.sleep(sleep_time)
page_sleep += 1
def apply_jobs(self):
try:
no_jobs_element = self.driver.find_element(By.CLASS_NAME, 'jobs-search-two-pane__no-results-banner--expand')
if 'No matching jobs found' in no_jobs_element.text or 'unfortunately, things aren' in self.driver.page_source.lower():
raise Exception("No more jobs on this page")
except NoSuchElementException:
pass
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, step=300, reverse=True)
job_list_elements = self.driver.find_elements(By.CLASS_NAME, 'scaffold-layout__list-container')[0].find_elements(By.CLASS_NAME, 'jobs-search-results__list-item')
if not job_list_elements:
raise Exception("No job class elements found on page")
job_list = [Job(*self.extract_job_information_from_tile(job_element)) for job_element in job_list_elements]
for job in job_list:
if self.is_blacklisted(job.title, job.company, job.link):
utils.printyellow(f"Blacklisted {job.title} at {job.company}, skipping...")
self.write_to_file(job, "skipped")
continue
try:
if job.apply_method not in {"Continue", "Applied", "Apply"}:
self.easy_applier_component.job_apply(job)
self.write_to_file(job, "success")
except Exception as e:
utils.printred(traceback.format_exc())
self.write_to_file(job, "failed")
continue
def write_to_file(self, job, file_name):
pdf_path = Path(job.pdf_path).as_uri()
data = {
"company": job.company,
"job_title": job.title,
"link": job.link,
"job_location": job.location,
"pdf_path": pdf_path
}
file_path = self.output_file_directory / f"{file_name}.json"
if not file_path.exists():
with open(file_path, 'w', encoding='utf-8') as f:
json.dump([data], f, indent=4)
else:
with open(file_path, 'r+', encoding='utf-8') as f:
try:
existing_data = json.load(f)
except json.JSONDecodeError:
existing_data = []
existing_data.append(data)
f.seek(0)
json.dump(existing_data, f, indent=4)
f.truncate()
def get_base_search_url(self, parameters):
url_parts = []
if parameters['remote']:
url_parts.append("f_CF=f_WRA")
experience_levels = [str(i+1) for i, v in enumerate(parameters.get('experienceLevel', [])) if v]
if experience_levels:
url_parts.append(f"f_E={','.join(experience_levels)}")
url_parts.append(f"distance={parameters['distance']}")
job_types = [key[0].upper() for key, value in parameters.get('jobTypes', {}).items() if value]
if job_types:
url_parts.append(f"f_JT={','.join(job_types)}")
date_mapping = {
"all time": "",
"month": "&f_TPR=r2592000",
"week": "&f_TPR=r604800",
"24 hours": "&f_TPR=r86400"
}
date_param = next((v for k, v in date_mapping.items() if parameters.get('date', {}).get(k)), "")
url_parts.append("f_LF=f_AL") # Easy Apply
base_url = "&".join(url_parts)
return f"?{base_url}{date_param}"
def next_job_page(self, position, location, job_page):
self.driver.get(f"https://www.linkedin.com/jobs/search/{self.base_search_url}&keywords={position}{location}&start={job_page * 25}")
def extract_job_information_from_tile(self, job_tile):
job_title, company, job_location, apply_method, link = "", "", "", "", ""
try:
job_title = job_tile.find_element(By.CLASS_NAME, 'job-card-list__title').text
link = job_tile.find_element(By.CLASS_NAME, 'job-card-list__title').get_attribute('href').split('?')[0]
company = job_tile.find_element(By.CLASS_NAME, 'job-card-container__primary-description').text
except:
pass
try:
job_location = job_tile.find_element(By.CLASS_NAME, 'job-card-container__metadata-item').text
except:
pass
try:
apply_method = job_tile.find_element(By.CLASS_NAME, 'job-card-container__apply-method').text
except:
apply_method = "Applied"
return job_title, company, job_location, link, apply_method
def is_blacklisted(self, job_title, company, link):
job_title_words = job_title.lower().split(' ')
title_blacklisted = any(word in job_title_words for word in self.title_blacklist)
company_blacklisted = company.strip().lower() in (word.strip().lower() for word in self.company_blacklist)
link_seen = link in self.seen_jobs
return title_blacklisted or company_blacklisted or link_seen