Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

broke up utils.py into multiple files #39

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

broke up utils.py into multiple files #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 22 additions & 22 deletions linkedin.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OPTIONAL: Remove direct usage of prGreen, prRed, prYellow outside of logger.

This suggestion is optional and can be done in a separate PR

from webdriver_manager.chrome import ChromeDriverManager
from typing import List

Expand Down Expand Up @@ -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")
Expand All @@ -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
Expand Down Expand Up @@ -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()
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
searchUrl = url_generator.LinkedinUrlGenerator.getGeneralSearchUrl()
searchUrl = LinkedinUrlGenerator.getGeneralSearchUrl()

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adjust the import so you can apply this suggestion

self.goToUrl(searchUrl)
utils.sleepInBetweenActions()
sleeper.sleepInBetweenActions()


def goToUrl(self, url):
self.driver.get(url)
utils.sleepInBetweenActions()
sleeper.sleepInBetweenActions()


def goToJobPage(self, jobID):
Expand All @@ -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)
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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,
Expand All @@ -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

Expand Down Expand Up @@ -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

Expand All @@ -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

Expand All @@ -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

Expand All @@ -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

Expand All @@ -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

Expand All @@ -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

Expand All @@ -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

Expand Down
33 changes: 17 additions & 16 deletions repository_wrapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import utils, models
import models
from utils import logger
from dotenv import load_dotenv

initialized = False
Expand All @@ -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()


Expand All @@ -19,66 +20,66 @@ 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


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)
2 changes: 1 addition & 1 deletion runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import time
from linkedin import Linkedin
from utils import prYellow
from utils.logger import prYellow


start = time.time()
Expand Down
Loading