utils.py

import math
import os
import random
import time
import traceback
import re
from enum import Enum
from typing import List
from selenium import webdriver
from selenium.common.exceptions import WebDriverException

import config
import constants


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))


def extractTextWithinParentheses(text):
    # Pattern to match text within parentheses
    pattern = r"\((.*?)\)"
    match = re.search(pattern, text)
    
    if match:
        # Return the content within the first set of parentheses
        return match.group(1)  # `group(1)` returns the content within the parentheses
    else:
        return ""


class LinkedinUrlGenerator:


    @staticmethod
    def getGeneralSearchUrl():
        return constants.searchJobsUrl
    

    @staticmethod
    def getEasyApplySearchUrl():
        return constants.searchEasyApplyJobsUrl


    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