From 0c5100c1e13465dd05164a9fce7a73360a593947 Mon Sep 17 00:00:00 2001 From: alirahmanicode Date: Mon, 19 Aug 2024 17:56:38 +0330 Subject: [PATCH] broke up utils.py into multiple files --- linkedin.py | 44 ++--- repository_wrapper.py | 33 ++-- runner.py | 2 +- utils.py | 388 ----------------------------------------- utils/__init__.py | 0 utils/logger.py | 52 ++++++ utils/sleeper.py | 20 +++ utils/url_generator.py | 227 ++++++++++++++++++++++++ utils/utils.py | 114 ++++++++++++ 9 files changed, 453 insertions(+), 427 deletions(-) delete mode 100755 utils.py create mode 100644 utils/__init__.py create mode 100644 utils/logger.py create mode 100644 utils/sleeper.py create mode 100644 utils/url_generator.py create mode 100755 utils/utils.py diff --git a/linkedin.py b/linkedin.py index 8e28e11..c3cb77e 100755 --- a/linkedin.py +++ b/linkedin.py @@ -3,15 +3,15 @@ import constants import models import repository_wrapper -import utils import warnings +from utils import sleeper, url_generator, utils, logger from selenium import webdriver from selenium.common.exceptions import NoSuchElementException, TimeoutException from selenium.webdriver.chrome.service import Service as ChromeService from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait -from utils import prGreen, prRed, prYellow +from logger.logger import prGreen, prRed, prYellow from webdriver_manager.chrome import ChromeDriverManager from typing import List @@ -53,11 +53,11 @@ def __init__(self): prYellow("🔄 Trying to log in linkedin...") try: self.driver.find_element("id","username").send_keys(config.email) - utils.sleepInBetweenActions(1,2) + sleeper.sleepInBetweenActions(1,2) self.driver.find_element("id","password").send_keys(config.password) - utils.sleepInBetweenActions(1, 2) + sleeper.sleepInBetweenActions(1, 2) self.driver.find_element("xpath",'//button[@type="submit"]').click() - utils.sleepInBetweenActions(3, 7) + sleeper.sleepInBetweenActions(3, 7) self.checkIfLoggedIn() except: prRed("❌ Couldn't log in Linkedin by using Chrome. Please check your Linkedin credentials on config files line 7 and 8. If error continue you can define Chrome profile or run the bot on Firefox") @@ -77,12 +77,12 @@ def startApplying(self): try: jobCounter = models.JobCounter() - urlData = utils.LinkedinUrlGenerator().generateSearchUrls() + urlData = url_generator.LinkedinUrlGenerator().generateSearchUrls() for url in urlData: self.goToUrl(url) - urlWords = utils.urlToKeywords(url) + urlWords = url_generator.urlToKeywords(url) try: totalJobs = self.wait.until(EC.presence_of_element_located((By.XPATH, '//small'))).text # TODO - fix finding total jobs @@ -111,21 +111,21 @@ def startApplying(self): " jobs out of " + str(jobCounter.total) + ".") except Exception as e: - utils.logDebugMessage("Unhandled exception in startApplying", utils.MessageTypes.ERROR, e, True) + logger.logDebugMessage("Unhandled exception in startApplying", logger.MessageTypes.ERROR, e, True) self.driver.save_screenshot("unhandled_exception.png") with open("page_source_at_unhandled_exception.html", "w") as file: file.write(self.driver.page_source) def goToJobsSearchPage(self): - searchUrl = utils.LinkedinUrlGenerator.getGeneralSearchUrl() + searchUrl = url_generator.LinkedinUrlGenerator.getGeneralSearchUrl() self.goToUrl(searchUrl) - utils.sleepInBetweenActions() + sleeper.sleepInBetweenActions() def goToUrl(self, url): self.driver.get(url) - utils.sleepInBetweenActions() + sleeper.sleepInBetweenActions() def goToJobPage(self, jobID): @@ -137,7 +137,7 @@ def goToJobPage(self, jobID): def processJob(self, jobID: str, jobCounter: models.JobCounter): jobPage = self.goToJobPage(jobID) jobCounter.total += 1 - utils.sleepInBetweenBatches(jobCounter.total) + sleeper.sleepInBetweenBatches(jobCounter.total) jobProperties = self.getJobProperties(jobID) repository_wrapper.update_job(jobProperties) @@ -195,7 +195,7 @@ def getJobsFromSearchPage(self) -> List[models.JobForVerification]: title=jobTitle, company=companyName)) else: - utils.logDebugMessage("Couldn't find jobID, jobTitle or companyName", utils.MessageTypes.WARNING) + logger.logDebugMessage("Couldn't find jobID, jobTitle or companyName", logger.MessageTypes.WARNING) return jobsForVerification @@ -275,7 +275,7 @@ def getJobProperties(self, jobID: str) -> models.Job: jobPostedDate = self.getJobPostedDate(primary_description_div) numberOfApplicants = self.getNumberOfApplicants(primary_description_div) else: - utils.logDebugMessage("in getting primary_description_div", utils.MessageTypes.WARNING) + logger.logDebugMessage("in getting primary_description_div", logger.MessageTypes.WARNING) return models.Job( title=jobTitle, @@ -295,7 +295,7 @@ def getJobTitle(self): try: jobTitle = self.getJobTitleMethod2() except Exception as e: - utils.logDebugMessage("in getting jobTitle", utils.MessageTypes.WARNING, e) + logger.logDebugMessage("in getting jobTitle", logger.MessageTypes.WARNING, e) return jobTitle @@ -327,7 +327,7 @@ def getJobCompanyMethod1(self, primary_description_div): jobCompanyLink = primary_description_div.find_element(By.CSS_SELECTOR, "a.app-aware-link") jobCompany = jobCompanyLink.text.strip() else: - utils.logDebugMessage("in getting jobCompany", utils.MessageTypes.WARNING) + logger.logDebugMessage("in getting jobCompany", logger.MessageTypes.WARNING) return jobCompany @@ -341,7 +341,7 @@ def getJobCompanyMethod2(self): jobCompany = jobCompanyElement.text.strip() else: - utils.logDebugMessage("in getting jobCompany card", utils.MessageTypes.WARNING) + logger.logDebugMessage("in getting jobCompany card", logger.MessageTypes.WARNING) return jobCompany @@ -353,7 +353,7 @@ def getJobLocation(self, primary_description_div): jobLocationSpan = primary_description_div.find_element(By.XPATH, ".//span[contains(@class, 'tvm__text--low-emphasis')][1]") jobLocation = jobLocationSpan.text.strip() except Exception as e: - utils.logDebugMessage("in getting jobLocation", utils.MessageTypes.WARNING, e) + logger.logDebugMessage("in getting jobLocation", logger.MessageTypes.WARNING, e) return jobLocation @@ -365,7 +365,7 @@ def getJobPostedDate(self, primary_description_div): postedDateSpan = primary_description_div.find_element(By.XPATH, ".//span[contains(@class, 'tvm__text--low-emphasis')][3]") jobPostedDate = postedDateSpan.text.strip() except Exception as e: - utils.logDebugMessage("Error in getting jobPostedDate", utils.MessageTypes.WARNING, e) + logger.logDebugMessage("Error in getting jobPostedDate", logger.MessageTypes.WARNING, e) return jobPostedDate @@ -377,7 +377,7 @@ def getNumberOfApplicants(self, primary_description_div): applicationsSpan = primary_description_div.find_element(By.XPATH, ".//span[contains(@class, 'tvm__text--low-emphasis')][5]") jobApplications = applicationsSpan.text.strip() except Exception as e: - utils.logDebugMessage("Error in getting jobApplications", utils.MessageTypes.WARNING, e) + logger.logDebugMessage("Error in getting jobApplications", logger.MessageTypes.WARNING, e) return jobApplications @@ -388,7 +388,7 @@ def getJobWorkPlaceType(self): try: jobWorkPlaceType = self.driver.find_element(By.XPATH,"//span[contains(@class, 'workplace-type')]").get_attribute("innerHTML").strip() except Exception as e: - utils.logDebugMessage("in getting jobWorkPlaceType", utils.MessageTypes.WARNING, e) + logger.logDebugMessage("in getting jobWorkPlaceType", logger.MessageTypes.WARNING, e) return jobWorkPlaceType @@ -400,7 +400,7 @@ def getJobDescription(self): try: jobDescription = self.driver.find_element(By.XPATH, "//div[contains(@class, 'job-details-jobs')]//div").text.replace("¡", "|") except Exception as e: - utils.logDebugMessage("in getting jobDescription: ", utils.MessageTypes.WARNING, e) + logger.logDebugMessage("in getting jobDescription: ", logger.MessageTypes.WARNING, e) return jobDescription diff --git a/repository_wrapper.py b/repository_wrapper.py index 8b1bb58..c042542 100644 --- a/repository_wrapper.py +++ b/repository_wrapper.py @@ -1,4 +1,5 @@ -import utils, models +import models +from utils import logger from dotenv import load_dotenv initialized = False @@ -7,7 +8,7 @@ def init(): global initialized, backend_api - utils.logDebugMessage("Initializing repository wrapper...") + logger.logDebugMessage("Initializing repository wrapper...") initialized, backend_api = import_backend_module() @@ -19,22 +20,22 @@ def import_backend_module(): from frontend.utils import ( api as backend_api, # Change this line with your backend module ) - utils.logDebugMessage(f"Successfully imported backend module", utils.MessageTypes.SUCCESS) + logger.logDebugMessage(f"Successfully imported backend module", logger.MessageTypes.SUCCESS) return True, backend_api except ImportError as e: - utils.logDebugMessage(f"Could not import backend module: {e}", utils.MessageTypes.WARNING) + logger.logDebugMessage(f"Could not import backend module: {e}", logger.MessageTypes.WARNING) return False, None def verify_jobs(jobs): if initialized: try: - utils.logDebugMessage(f"Verifying jobs: {jobs}") + logger.logDebugMessage(f"Verifying jobs: {jobs}") jobs = backend_api.verify_jobs(jobs) except Exception as e: - utils.logDebugMessage(f"Error verifying jobs: {e}", utils.MessageTypes.ERROR) + logger.logDebugMessage(f"Error verifying jobs: {e}", logger.MessageTypes.ERROR) return jobs @@ -42,43 +43,43 @@ def verify_jobs(jobs): def update_job(job: models.Job): if initialized: try: - utils.logDebugMessage(f"Updating job: {job}") + logger.logDebugMessage(f"Updating job: {job}") job = backend_api.update_job_with_job_properties(job) except Exception as e: - utils.logDebugMessage(f"Error updating job: {e}", utils.MessageTypes.ERROR) + logger.logDebugMessage(f"Error updating job: {e}", logger.MessageTypes.ERROR) return job def attached_resume_to_job(job: models.Job, resume: str): if initialized: try: - utils.logDebugMessage(f"Attaching resume to job: {job}") + logger.logDebugMessage(f"Attaching resume to job: {job}") backend_api.attached_resume_to_job(job.linkedin_job_id, resume) except Exception as e: - utils.logDebugMessage(f"Error attaching resume to job: {e}", utils.MessageTypes.ERROR) + logger.logDebugMessage(f"Error attaching resume to job: {e}", logger.MessageTypes.ERROR) def get_answer_by_question(question): if initialized: try: - utils.logDebugMessage(f"Getting answer for question: {question}") + logger.logDebugMessage(f"Getting answer for question: {question}") # TODO: Implement this except Exception as e: - utils.logDebugMessage(f"Error getting answer for question: {e}", utils.MessageTypes.ERROR) + logger.logDebugMessage(f"Error getting answer for question: {e}", logger.MessageTypes.ERROR) def post_question(question): if initialized: try: - utils.logDebugMessage(f"Posting question: {question} with answer:") + logger.logDebugMessage(f"Posting question: {question} with answer:") # TODO: Implement this except Exception as e: - utils.logDebugMessage(f"Error posting question: {e}", utils.MessageTypes.ERROR) + logger.logDebugMessage(f"Error posting question: {e}", logger.MessageTypes.ERROR) def applied_to_job(job: models.Job): if initialized: try: - utils.logDebugMessage(f"Marking job as applied: {job}") + logger.logDebugMessage(f"Marking job as applied: {job}") backend_api.applied_to_job(job.linkedin_job_id) except Exception as e: - utils.logDebugMessage(f"Error marking job as applied: {e}", utils.MessageTypes.ERROR) + logger.logDebugMessage(f"Error marking job as applied: {e}", logger.MessageTypes.ERROR) diff --git a/runner.py b/runner.py index 7bffc11..3ab54c1 100644 --- a/runner.py +++ b/runner.py @@ -1,6 +1,6 @@ import time from linkedin import Linkedin -from utils import prYellow +from utils.logger import prYellow start = time.time() diff --git a/utils.py b/utils.py deleted file mode 100755 index c8f7c5b..0000000 --- a/utils.py +++ /dev/null @@ -1,388 +0,0 @@ -import math -import os -import random -import time -import traceback -from enum import Enum -from typing import List - -import config -import constants -from selenium import webdriver - - -def chromeBrowserOptions(): - options = webdriver.ChromeOptions() - options.add_argument('--no-sandbox') - options.add_argument("--ignore-certificate-errors") - options.add_argument("--disable-extensions") - options.add_argument('--disable-gpu') - options.add_argument('--disable-dev-shm-usage') - if(config.headless): - options.add_argument("--headless") - options.add_argument("--start-maximized") - options.add_argument("--disable-blink-features") - options.add_argument("--disable-blink-features=AutomationControlled") - options.add_experimental_option('useAutomationExtension', False) - options.add_experimental_option("excludeSwitches", ["enable-automation"]) - if(len(config.chromeProfilePath)>0): - initialPath = config.chromeProfilePath[0:config.chromeProfilePath.rfind("/")] - profileDir = config.chromeProfilePath[config.chromeProfilePath.rfind("/")+1:] - options.add_argument('--user-data-dir=' + initialPath) - options.add_argument("--profile-directory=" + profileDir) - else: - # options.add_argument("--incognito") - # this is for running in a docker container - user_data_dir = os.environ.get('CHROME_USER_DATA_DIR', '/home/user/chrome_data') - options.add_argument(f'--user-data-dir={user_data_dir}') - return options - - -def prRed(prt): - print(f"\033[91m{prt}\033[00m") - - -def prGreen(prt): - print(f"\033[92m{prt}\033[00m") - - -def prYellow(prt): - print(f"\033[93m{prt}\033[00m") - - -def prBlue(prt): - print(f"\033[94m{prt}\033[00m") - - -class MessageTypes(Enum): - INFO = 1 - WARNING = 2 - ERROR = 3 - SUCCESS = 4 - - -def printInfoMes(bot:str): - prYellow("ℹī¸ " +bot+ " is starting soon... ") - - -def logDebugMessage(message, messageType=MessageTypes.INFO, exception=Exception(), displayTraceback = False): - if (config.displayWarnings): - match messageType: - case MessageTypes.INFO: - prBlue(f"ℹī¸ {message}") - case MessageTypes.WARNING: - prYellow(f"⚠ī¸ Warning ⚠ī¸ {message}: {str(exception)[0:100]}") - case MessageTypes.ERROR: - prRed(f"❌ Error ❌ {message}: {str(exception)[0:100]}") - case MessageTypes.SUCCESS: - prGreen(f"✅ {message}") - - if (displayTraceback): - traceback.print_exc() - - -def jobsToPages(numOfJobs: str) -> int: - number_of_pages = 1 - - if (' ' in numOfJobs): - spaceIndex = numOfJobs.index(' ') - totalJobs = (numOfJobs[0:spaceIndex]) - totalJobs_int = int(totalJobs.replace(',', '')) - number_of_pages = math.ceil(totalJobs_int/constants.jobsPerPage) - if (number_of_pages > 40 ): number_of_pages = 40 - - else: - number_of_pages = int(numOfJobs) - - return number_of_pages - - -def urlToKeywords(url: str) -> List[str]: - keywordUrl = url[url.index("keywords=")+9:] - keyword = keywordUrl[0:keywordUrl.index("&") ] - locationUrl = url[url.index("location=")+9:] - location = locationUrl[0:locationUrl.index("&") ] - return [keyword,location] - - -def writeResults(text: str): - timeStr = time.strftime("%Y%m%d") - directory = "data" - fileName = "Applied Jobs DATA - " + timeStr + ".txt" - filePath = os.path.join(directory, fileName) - - try: - os.makedirs(directory, exist_ok=True) # Ensure the 'data' directory exists. - - # Open the file for reading and writing ('r+' opens the file for both) - with open(filePath, 'r+', encoding="utf-8") as file: - lines = [] - for line in file: - if "----" not in line: - lines.append(line) - file.seek(0) # Go back to the start of the file - file.truncate() # Clear the file - file.write("---- Applied Jobs Data ---- created at: " + timeStr + "\n") - file.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " + "\n") - for line in lines: - file.write(line) - file.write(text + "\n") - except FileNotFoundError: - with open(filePath, 'w', encoding="utf-8") as f: - f.write("---- Applied Jobs Data ---- created at: " + timeStr + "\n") - f.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " + "\n") - f.write(text + "\n") - except Exception as e: - prRed(f"❌ Error in writeResults: {e}") # Assuming prRed is a function to print errors in red color - - -# def writeResults(text: str): -# timeStr = time.strftime("%Y%m%d") -# fileName = "Applied Jobs DATA - " +timeStr + ".txt" -# try: -# with open("data/" +fileName, encoding="utf-8" ) as file: -# lines = [] -# for line in file: -# if "----" not in line: -# lines.append(line) - -# with open("data/" +fileName, 'w' ,encoding="utf-8") as f: -# f.write("---- Applied Jobs Data ---- created at: " +timeStr+ "\n" ) -# f.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " +"\n" ) -# for line in lines: -# f.write(line) -# f.write(text+ "\n") - -# except: -# with open("data/" +fileName, 'w', encoding="utf-8") as f: -# f.write("---- Applied Jobs Data ---- created at: " +timeStr+ "\n" ) -# f.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " +"\n" ) - -# f.write(text+ "\n") - - -def interact(action): - action() - sleepInBetweenActions() - - -def sleepInBetweenActions(bottom: int = constants.botSleepInBetweenActionsBottom, top: int = constants.botSleepInBetweenActionsTop): - time.sleep(random.uniform(bottom, top)) - - -def sleepInBetweenBatches(currentBatch: int, bottom: int = constants.botSleepInBetweenBatchesBottom, top: int = constants.botSleepInBetweenBatchesTop): - if (currentBatch % constants.batchSize == 0): - time.sleep(random.uniform(bottom, top)) - - -class LinkedinUrlGenerator: - @staticmethod - def getGeneralSearchUrl(): - return constants.searchJobsUrl - - - def generateSearchUrls(self): - urls = [] - for location in config.location: - for keyword in config.keywords: - url = constants.searchJobsUrl + "?f_AL=true&keywords=" + keyword + self.jobType() + self.remote() + self.checkJobLocation(location) + self.jobExp() + self.datePosted() + self.jobTitle() + self.salary() + self.sortBy() - urls.append(url) - return urls - - - def checkJobLocation(self, job): - jobLoc = "&location=" + job - match job.casefold(): - case "asia": - jobLoc += "&geoId=102393603" - case "europe": - jobLoc += "&geoId=100506914" - case "northamerica": - jobLoc += "&geoId=102221843&" - case "southamerica": - jobLoc += "&geoId=104514572" - case "australia": - jobLoc += "&geoId=101452733" - case "africa": - jobLoc += "&geoId=103537801" - case "sweden": - jobLoc += "&geoId=105117694" - case "norway": - jobLoc += "&geoId=103819153" - case "germany": - jobLoc += "&geoId=101282230" - case "switzerland": - jobLoc += "&geoId=106693272" - case "new york": - jobLoc += "&geoId=105080838" - - return jobLoc - - - def jobExp(self): - jobtExpArray = config.experienceLevels - firstJobExp = jobtExpArray[0] - jobExp = "" - match firstJobExp: - case "Internship": - jobExp = "&f_E=1" - case "Entry level": - jobExp = "&f_E=2" - case "Associate": - jobExp = "&f_E=3" - case "Mid-Senior level": - jobExp = "&f_E=4" - case "Director": - jobExp = "&f_E=5" - case "Executive": - jobExp = "&f_E=6" - for index in range (1,len(jobtExpArray)): - match jobtExpArray[index]: - case "Internship": - jobExp += "%2C1" - case "Entry level": - jobExp +="%2C2" - case "Associate": - jobExp +="%2C3" - case "Mid-Senior level": - jobExp += "%2C4" - case "Director": - jobExp += "%2C5" - case "Executive": - jobExp +="%2C6" - - return jobExp - - - def datePosted(self): - datePosted = "" - match config.datePosted[0]: - case "Any Time": - datePosted = "" - case "Past Month": - datePosted = "&f_TPR=r2592000&" - case "Past Week": - datePosted = "&f_TPR=r604800&" - case "Past 24 hours": - datePosted = "&f_TPR=r86400&" - return datePosted - - - def jobType(self): - jobTypeArray = config.jobType - firstjobType = jobTypeArray[0] - jobType = "" - match firstjobType: - case "Full-time": - jobType = "&f_JT=F" - case "Part-time": - jobType = "&f_JT=P" - case "Contract": - jobType = "&f_JT=C" - case "Temporary": - jobType = "&f_JT=T" - case "Volunteer": - jobType = "&f_JT=V" - case "Intership": - jobType = "&f_JT=I" - case "Other": - jobType = "&f_JT=O" - for index in range (1,len(jobTypeArray)): - match jobTypeArray[index]: - case "Full-time": - jobType += "%2CF" - case "Part-time": - jobType +="%2CP" - case "Contract": - jobType +="%2CC" - case "Temporary": - jobType += "%2CT" - case "Volunteer": - jobType += "%2CV" - case "Intership": - jobType +="%2CI" - case "Other": - jobType +="%2CO" - jobType += "&" - return jobType - - - def remote(self): - remoteArray = config.remote - firstJobRemote = remoteArray[0] - jobRemote = "" - match firstJobRemote: - case "On-site": - jobRemote = "f_WT=1" - case "Remote": - jobRemote = "f_WT=2" - case "Hybrid": - jobRemote = "f_WT=3" - for index in range (1,len(remoteArray)): - match remoteArray[index]: - case "On-site": - jobRemote += "%2C1" - case "Remote": - jobRemote += "%2C2" - case "Hybrid": - jobRemote += "%2C3" - - return jobRemote - - - def jobTitle(self): - jobTitleArray = config.jobTitles - - # Ensure we have at least one job title to process - if not jobTitleArray: - return "" - - # Use the first job title for the initial job title parameter - initial_code = constants.job_title_codes.get(jobTitleArray[0]) - if initial_code: - jobTitle = f"f_T={initial_code}" - else: - return "" # If the first job title isn't recognized, return an empty string or handle error appropriately - - # Process subsequent job titles - for title in jobTitleArray[1:]: - code = constants.job_title_codes.get(title) - if code: - jobTitle += f"%2C{code}" - - jobTitle += "&" - return jobTitle - - - def salary(self): - salary = "" - match config.salary: - case "$40,000+": - salary = "f_SB2=1&" - case "$60,000+": - salary = "f_SB2=2&" - case "$80,000+": - salary = "f_SB2=3&" - case "$100,000+": - salary = "f_SB2=4&" - case "$120,000+": - salary = "f_SB2=5&" - case "$140,000+": - salary = "f_SB2=6&" - case "$160,000+": - salary = "f_SB2=7&" - case "$180,000+": - salary = "f_SB2=8&" - case "$200,000+": - salary = "f_SB2=9&" - return salary - - - def sortBy(self): - sortBy = "" - match config.sort[0]: - case "Recent": - sortBy = "sortBy=DD" - case "Relevent": - sortBy = "sortBy=R" - return sortBy diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils/logger.py b/utils/logger.py new file mode 100644 index 0000000..6e97b12 --- /dev/null +++ b/utils/logger.py @@ -0,0 +1,52 @@ +import traceback +from enum import Enum + +import config + + +class MessageTypes(Enum): + INFO = 1 + WARNING = 2 + ERROR = 3 + SUCCESS = 4 + + +def prRed(prt): + print(f"\033[91m{prt}\033[00m") + + +def prGreen(prt): + print(f"\033[92m{prt}\033[00m") + + +def prYellow(prt): + print(f"\033[93m{prt}\033[00m") + + +def prBlue(prt): + print(f"\033[94m{prt}\033[00m") + + +def printInfoMes(bot: str): + prYellow("ℹī¸ " + bot + " is starting soon... ") + + +def logDebugMessage( + message, + messageType=MessageTypes.INFO, + exception=Exception(), + displayTraceback=False, +): + if config.displayWarnings: + match messageType: + case MessageTypes.INFO: + prBlue(f"ℹī¸ {message}") + case MessageTypes.WARNING: + prYellow(f"⚠ī¸ Warning ⚠ī¸ {message}: {str(exception)[0:100]}") + case MessageTypes.ERROR: + prRed(f"❌ Error ❌ {message}: {str(exception)[0:100]}") + case MessageTypes.SUCCESS: + prGreen(f"✅ {message}") + + if displayTraceback: + traceback.print_exc() diff --git a/utils/sleeper.py b/utils/sleeper.py new file mode 100644 index 0000000..8f66cf9 --- /dev/null +++ b/utils/sleeper.py @@ -0,0 +1,20 @@ +import random +import time + +import constants + + +def sleepInBetweenActions( + bottom: int = constants.botSleepInBetweenActionsBottom, + top: int = constants.botSleepInBetweenActionsTop, +): + time.sleep(random.uniform(bottom, top)) + + +def sleepInBetweenBatches( + currentBatch: int, + bottom: int = constants.botSleepInBetweenBatchesBottom, + top: int = constants.botSleepInBetweenBatchesTop, +): + if currentBatch % constants.batchSize == 0: + time.sleep(random.uniform(bottom, top)) diff --git a/utils/url_generator.py b/utils/url_generator.py new file mode 100644 index 0000000..fd4376f --- /dev/null +++ b/utils/url_generator.py @@ -0,0 +1,227 @@ +from typing import List +import config +import constants + + +class LinkedinUrlGenerator: + @staticmethod + def getGeneralSearchUrl(): + return constants.searchJobsUrl + + def generateSearchUrls(self): + urls = [] + for location in config.location: + for keyword in config.keywords: + url = ( + constants.searchJobsUrl + + "?f_AL=true&keywords=" + + keyword + + self.jobType() + + self.remote() + + self.checkJobLocation(location) + + self.jobExp() + + self.datePosted() + + self.jobTitle() + + self.salary() + + self.sortBy() + ) + urls.append(url) + return urls + + def checkJobLocation(self, job): + jobLoc = "&location=" + job + match job.casefold(): + case "asia": + jobLoc += "&geoId=102393603" + case "europe": + jobLoc += "&geoId=100506914" + case "northamerica": + jobLoc += "&geoId=102221843&" + case "southamerica": + jobLoc += "&geoId=104514572" + case "australia": + jobLoc += "&geoId=101452733" + case "africa": + jobLoc += "&geoId=103537801" + case "sweden": + jobLoc += "&geoId=105117694" + case "norway": + jobLoc += "&geoId=103819153" + case "germany": + jobLoc += "&geoId=101282230" + case "switzerland": + jobLoc += "&geoId=106693272" + case "new york": + jobLoc += "&geoId=105080838" + + return jobLoc + + def jobExp(self): + jobtExpArray = config.experienceLevels + firstJobExp = jobtExpArray[0] + jobExp = "" + match firstJobExp: + case "Internship": + jobExp = "&f_E=1" + case "Entry level": + jobExp = "&f_E=2" + case "Associate": + jobExp = "&f_E=3" + case "Mid-Senior level": + jobExp = "&f_E=4" + case "Director": + jobExp = "&f_E=5" + case "Executive": + jobExp = "&f_E=6" + for index in range(1, len(jobtExpArray)): + match jobtExpArray[index]: + case "Internship": + jobExp += "%2C1" + case "Entry level": + jobExp += "%2C2" + case "Associate": + jobExp += "%2C3" + case "Mid-Senior level": + jobExp += "%2C4" + case "Director": + jobExp += "%2C5" + case "Executive": + jobExp += "%2C6" + + return jobExp + + def datePosted(self): + datePosted = "" + match config.datePosted[0]: + case "Any Time": + datePosted = "" + case "Past Month": + datePosted = "&f_TPR=r2592000&" + case "Past Week": + datePosted = "&f_TPR=r604800&" + case "Past 24 hours": + datePosted = "&f_TPR=r86400&" + return datePosted + + def jobType(self): + jobTypeArray = config.jobType + firstjobType = jobTypeArray[0] + jobType = "" + match firstjobType: + case "Full-time": + jobType = "&f_JT=F" + case "Part-time": + jobType = "&f_JT=P" + case "Contract": + jobType = "&f_JT=C" + case "Temporary": + jobType = "&f_JT=T" + case "Volunteer": + jobType = "&f_JT=V" + case "Intership": + jobType = "&f_JT=I" + case "Other": + jobType = "&f_JT=O" + for index in range(1, len(jobTypeArray)): + match jobTypeArray[index]: + case "Full-time": + jobType += "%2CF" + case "Part-time": + jobType += "%2CP" + case "Contract": + jobType += "%2CC" + case "Temporary": + jobType += "%2CT" + case "Volunteer": + jobType += "%2CV" + case "Intership": + jobType += "%2CI" + case "Other": + jobType += "%2CO" + jobType += "&" + return jobType + + def remote(self): + remoteArray = config.remote + firstJobRemote = remoteArray[0] + jobRemote = "" + match firstJobRemote: + case "On-site": + jobRemote = "f_WT=1" + case "Remote": + jobRemote = "f_WT=2" + case "Hybrid": + jobRemote = "f_WT=3" + for index in range(1, len(remoteArray)): + match remoteArray[index]: + case "On-site": + jobRemote += "%2C1" + case "Remote": + jobRemote += "%2C2" + case "Hybrid": + jobRemote += "%2C3" + + return jobRemote + + def jobTitle(self): + jobTitleArray = config.jobTitles + + # Ensure we have at least one job title to process + if not jobTitleArray: + return "" + + # Use the first job title for the initial job title parameter + initial_code = constants.job_title_codes.get(jobTitleArray[0]) + if initial_code: + jobTitle = f"f_T={initial_code}" + else: + return "" # If the first job title isn't recognized, return an empty string or handle error appropriately + + # Process subsequent job titles + for title in jobTitleArray[1:]: + code = constants.job_title_codes.get(title) + if code: + jobTitle += f"%2C{code}" + + jobTitle += "&" + return jobTitle + + def salary(self): + salary = "" + match config.salary: + case "$40,000+": + salary = "f_SB2=1&" + case "$60,000+": + salary = "f_SB2=2&" + case "$80,000+": + salary = "f_SB2=3&" + case "$100,000+": + salary = "f_SB2=4&" + case "$120,000+": + salary = "f_SB2=5&" + case "$140,000+": + salary = "f_SB2=6&" + case "$160,000+": + salary = "f_SB2=7&" + case "$180,000+": + salary = "f_SB2=8&" + case "$200,000+": + salary = "f_SB2=9&" + return salary + + def sortBy(self): + sortBy = "" + match config.sort[0]: + case "Recent": + sortBy = "sortBy=DD" + case "Relevent": + sortBy = "sortBy=R" + return sortBy + + +def urlToKeywords(url: str) -> List[str]: + keywordUrl = url[url.index("keywords=")+9:] + keyword = keywordUrl[0:keywordUrl.index("&") ] + locationUrl = url[url.index("location=")+9:] + location = locationUrl[0:locationUrl.index("&") ] + return [keyword,location] \ No newline at end of file diff --git a/utils/utils.py b/utils/utils.py new file mode 100755 index 0000000..2add101 --- /dev/null +++ b/utils/utils.py @@ -0,0 +1,114 @@ +import math +import os +import time + +import config +import constants +from selenium import webdriver + +from utils.sleeper import sleepInBetweenActions +from utils.logger import prRed + + +def chromeBrowserOptions(): + options = webdriver.ChromeOptions() + options.add_argument('--no-sandbox') + options.add_argument("--ignore-certificate-errors") + options.add_argument("--disable-extensions") + options.add_argument('--disable-gpu') + options.add_argument('--disable-dev-shm-usage') + if(config.headless): + options.add_argument("--headless") + options.add_argument("--start-maximized") + options.add_argument("--disable-blink-features") + options.add_argument("--disable-blink-features=AutomationControlled") + options.add_experimental_option('useAutomationExtension', False) + options.add_experimental_option("excludeSwitches", ["enable-automation"]) + if(len(config.chromeProfilePath)>0): + initialPath = config.chromeProfilePath[0:config.chromeProfilePath.rfind("/")] + profileDir = config.chromeProfilePath[config.chromeProfilePath.rfind("/")+1:] + options.add_argument('--user-data-dir=' + initialPath) + options.add_argument("--profile-directory=" + profileDir) + else: + # options.add_argument("--incognito") + # this is for running in a docker container + user_data_dir = os.environ.get('CHROME_USER_DATA_DIR', '/home/user/chrome_data') + options.add_argument(f'--user-data-dir={user_data_dir}') + return options + + + +def jobsToPages(numOfJobs: str) -> int: + number_of_pages = 1 + + if (' ' in numOfJobs): + spaceIndex = numOfJobs.index(' ') + totalJobs = (numOfJobs[0:spaceIndex]) + totalJobs_int = int(totalJobs.replace(',', '')) + number_of_pages = math.ceil(totalJobs_int/constants.jobsPerPage) + if (number_of_pages > 40 ): number_of_pages = 40 + + else: + number_of_pages = int(numOfJobs) + + return number_of_pages + +def writeResults(text: str): + timeStr = time.strftime("%Y%m%d") + directory = "data" + fileName = "Applied Jobs DATA - " + timeStr + ".txt" + filePath = os.path.join(directory, fileName) + + try: + os.makedirs(directory, exist_ok=True) # Ensure the 'data' directory exists. + + # Open the file for reading and writing ('r+' opens the file for both) + with open(filePath, 'r+', encoding="utf-8") as file: + lines = [] + for line in file: + if "----" not in line: + lines.append(line) + file.seek(0) # Go back to the start of the file + file.truncate() # Clear the file + file.write("---- Applied Jobs Data ---- created at: " + timeStr + "\n") + file.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " + "\n") + for line in lines: + file.write(line) + file.write(text + "\n") + except FileNotFoundError: + with open(filePath, 'w', encoding="utf-8") as f: + f.write("---- Applied Jobs Data ---- created at: " + timeStr + "\n") + f.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " + "\n") + f.write(text + "\n") + except Exception as e: + prRed(f"❌ Error in writeResults: {e}") # Assuming prRed is a function to print errors in red color + + +# def writeResults(text: str): +# timeStr = time.strftime("%Y%m%d") +# fileName = "Applied Jobs DATA - " +timeStr + ".txt" +# try: +# with open("data/" +fileName, encoding="utf-8" ) as file: +# lines = [] +# for line in file: +# if "----" not in line: +# lines.append(line) + +# with open("data/" +fileName, 'w' ,encoding="utf-8") as f: +# f.write("---- Applied Jobs Data ---- created at: " +timeStr+ "\n" ) +# f.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " +"\n" ) +# for line in lines: +# f.write(line) +# f.write(text+ "\n") + +# except: +# with open("data/" +fileName, 'w', encoding="utf-8") as f: +# f.write("---- Applied Jobs Data ---- created at: " +timeStr+ "\n" ) +# f.write("---- Number | Job Title | Company | Location | Work Place | Posted Date | Applications | Result " +"\n" ) + +# f.write(text+ "\n") + + +def interact(action): + action() + sleepInBetweenActions() \ No newline at end of file