diff --git a/data_folder/config.yaml b/data_folder/config.yaml index 53b71f1..1cbe9ed 100644 --- a/data_folder/config.yaml +++ b/data_folder/config.yaml @@ -35,13 +35,17 @@ applyOnceAtCompany: [true/false] distance: 100 -companyBlacklist: +company_blacklist: - Company1 - Company2 titleBlacklist: - word1 - word2 + +job_applicants_threshold: + min_applicants: 0 + max_applicants: 100 llm_model_type: openai llm_model: gpt-4o diff --git a/src/gpt.py b/src/gpt.py index c22f123..4107797 100644 --- a/src/gpt.py +++ b/src/gpt.py @@ -3,11 +3,11 @@ import os import re import textwrap import time -from datetime import datetime from abc import ABC, abstractmethod -from typing import Dict, List, Union +from datetime import datetime from pathlib import Path from typing import Dict, List +from typing import Union import httpx from Levenshtein import distance @@ -16,39 +16,42 @@ from langchain_core.messages.ai import AIMessage from langchain_core.output_parsers import StrOutputParser from langchain_core.prompt_values import StringPromptValue from langchain_core.prompts import ChatPromptTemplate -from langchain_openai import ChatOpenAI import src.strings as strings from src.utils import logger load_dotenv() + class AIModel(ABC): @abstractmethod def invoke(self, prompt: str) -> str: pass + class OpenAIModel(AIModel): def __init__(self, api_key: str, llm_model: str, llm_api_url: str): from langchain_openai import ChatOpenAI self.model = ChatOpenAI(model_name=llm_model, openai_api_key=api_key, temperature=0.4, base_url=llm_api_url) - + def invoke(self, prompt: str) -> str: print("invoke in openai") response = self.model.invoke(prompt) return response + class ClaudeModel(AIModel): def __init__(self, api_key: str, llm_model: str, llm_api_url: str): from langchain_anthropic import ChatAnthropic self.model = ChatAnthropic(model=llm_model, api_key=api_key, - temperature=0.4, base_url=llm_api_url) + temperature=0.4, base_url=llm_api_url) def invoke(self, prompt: str) -> str: response = self.model.invoke(prompt) return response + class OllamaModel(AIModel): def __init__(self, api_key: str, llm_model: str, llm_api_url: str): from langchain_ollama import ChatOllama @@ -58,6 +61,7 @@ class OllamaModel(AIModel): response = self.model.invoke(prompt) return response + class AIAdapter: def __init__(self, config: dict, api_key: str): self.model = self._create_model(config, api_key) @@ -67,7 +71,7 @@ class AIAdapter: llm_model = config['llm_model'] llm_api_url = config['llm_api_url'] print('Using {0} with {1} from {2}'.format(llm_model_type, llm_model, llm_api_url)) - + if llm_model_type == "openai": return OpenAIModel(api_key, llm_model, llm_api_url) elif llm_model_type == "claude": @@ -80,9 +84,9 @@ class AIAdapter: def invoke(self, prompt: str) -> str: return self.model.invoke(prompt) + class LLMLogger: - def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]): self.llm = llm @@ -189,7 +193,6 @@ class LLMLogger: class LoggerChatModel: - def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]): self.llm = llm @@ -247,7 +250,6 @@ class LoggerChatModel: time.sleep(30) continue - def parse_llmresult(self, llmresult: AIMessage) -> Dict[str, Dict]: logger.debug("Parsing LLM result: %s", llmresult) @@ -454,12 +456,14 @@ class GPTAnswerer: chain = prompt | self.llm_cheap | StrOutputParser() output = chain.invoke({"question": question}) - match = re.search(r"(Personal information|Self Identification|Legal Authorization|Work Preferences|Education Details|Experience Details|Projects|Availability|Salary Expectations|Certifications|Languages|Interests|Cover letter)", output, re.IGNORECASE) + match = re.search( + r"(Personal information|Self Identification|Legal Authorization|Work Preferences|Education Details|Experience Details|Projects|Availability|Salary Expectations|Certifications|Languages|Interests|Cover letter)", + output, re.IGNORECASE) if not match: raise ValueError("Could not extract section name from the response.") section_name = match.group(1).lower().replace(" ", "_") - + if section_name == "cover_letter": chain = chains.get(section_name) output = chain.invoke({"resume": self.resume, "job_description": self.job_description}) diff --git a/src/linkedIn_easy_applier.py b/src/linkedIn_easy_applier.py index 951fea8..cf64245 100644 --- a/src/linkedIn_easy_applier.py +++ b/src/linkedIn_easy_applier.py @@ -8,7 +8,7 @@ import traceback from typing import List, Optional, Any, Tuple from httpx import HTTPStatusError -from reportlab.lib.pagesizes import letter +from reportlab.lib.pagesizes import A4 from reportlab.pdfgen import canvas from selenium.common.exceptions import NoSuchElementException, TimeoutException from selenium.webdriver import ActionChains @@ -37,7 +37,6 @@ class LinkedInEasyApplier: logger.debug("LinkedInEasyApplier initialized successfully") - def _load_questions_from_json(self) -> List[dict]: output_file = 'answers.json' logger.debug("Loading questions from JSON file: %s", output_file) @@ -60,10 +59,8 @@ class LinkedInEasyApplier: logger.error("Error loading questions data from JSON file: %s", tb_str) raise Exception(f"Error loading questions data from JSON file: \nTraceback:\n{tb_str}") - def check_for_premium_redirect(self, job: Any, max_attempts=3): - """Проверяет, был ли выполнен редирект на страницу LinkedIn Premium. - В случае редиректа возвращает пользователя на исходную страницу вакансии.""" + current_url = self.driver.current_url attempts = 0 @@ -80,7 +77,6 @@ class LinkedInEasyApplier: raise Exception( f"Redirected to LinkedIn Premium page and failed to return after {max_attempts} attempts. Job application aborted.") - def job_apply(self, job: Any): logger.debug("Starting job application for job: %s", job) @@ -167,7 +163,7 @@ class LinkedInEasyApplier: logger.debug(f"Attempting search using {method['description']}") if method.get('find_elements'): - # Поиск всех кнопок "Easy Apply" + buttons = self.driver.find_elements(By.XPATH, method['xpath']) if buttons: for index, button in enumerate(buttons): @@ -209,7 +205,6 @@ class LinkedInEasyApplier: logger.error("No clickable 'Easy Apply' button found after 2 attempts. Page source:\n%s", page_source) raise Exception("No clickable 'Easy Apply' button found") - def _get_job_description(self) -> str: logger.debug("Getting job description") try: @@ -514,11 +509,47 @@ class LinkedInEasyApplier: file_path_pdf = os.path.join(folder_path, f"Cover_Letter_{timestamp}.pdf") logger.debug(f"Generated file path for cover letter: {file_path_pdf}") - c = canvas.Canvas(file_path_pdf, pagesize=letter) - _, height = letter - text_object = c.beginText(100, height - 100) + c = canvas.Canvas(file_path_pdf, pagesize=A4) + page_width, page_height = A4 + text_object = c.beginText(50, page_height - 50) text_object.setFont("Helvetica", 12) - text_object.textLines(cover_letter_text) + + max_width = page_width - 100 + bottom_margin = 50 + available_height = page_height - bottom_margin - 50 + + def split_text_by_width(text, font, font_size, max_width): + wrapped_lines = [] + for line in text.splitlines(): + + if utils.stringWidth(line, font, font_size) > max_width: + words = line.split() + new_line = "" + for word in words: + if utils.stringWidth(new_line + word + " ", font, font_size) <= max_width: + new_line += word + " " + else: + wrapped_lines.append(new_line.strip()) + new_line = word + " " + wrapped_lines.append(new_line.strip()) + else: + wrapped_lines.append(line) + return wrapped_lines + + lines = split_text_by_width(cover_letter_text, "Helvetica", 12, max_width) + + for line in lines: + text_height = text_object.getY() + if text_height > bottom_margin: + text_object.textLine(line) + else: + + c.drawText(text_object) + c.showPage() + text_object = c.beginText(50, page_height - 50) + text_object.setFont("Helvetica", 12) + text_object.textLine(line) + c.drawText(text_object) c.save() logger.debug(f"Cover letter successfully generated and saved to: {file_path_pdf}") @@ -632,7 +663,6 @@ class LinkedInEasyApplier: for item in self.all_data: - logger.debug( f"Comparing sanitized stored question: '{self._sanitize_text(item['question'])}' and type: '{item.get('type')}' with current question: '{self._sanitize_text(question_text)}' and type: '{question_type}'") @@ -659,7 +689,6 @@ class LinkedInEasyApplier: answer = self.gpt_answerer.answer_question_textual_wide_range(question_text) logger.debug(f"Generated textual answer: {answer}") - self._save_questions_to_json({'type': question_type, 'question': question_text, 'answer': answer}) self._enter_text(text_field, answer) logger.debug("Entered new answer into the textbox and saved it to JSON.") @@ -692,7 +721,6 @@ class LinkedInEasyApplier: logger.debug("Entered existing date answer") return True - self._save_questions_to_json({'type': 'date', 'question': question_text, 'answer': answer_text}) self._enter_text(date_field, answer_text) logger.debug("Entered new date answer") @@ -701,12 +729,12 @@ class LinkedInEasyApplier: def _find_and_handle_dropdown_question(self, section: WebElement) -> bool: try: - question = section.find_element(By.CLASS_NAME, 'jobs-easy-apply-form-element') - question_text = question.find_element(By.TAG_NAME, 'label').text.lower() - logger.debug(f"Processing dropdown or combobox question: {question_text}") dropdowns = question.find_elements(By.TAG_NAME, 'select') + if not dropdowns: + dropdowns = section.find_elements(By.CSS_SELECTOR, '[data-test-text-entity-list-form-select]') + if dropdowns: dropdown = dropdowns[0] select = Select(dropdown) @@ -714,6 +742,9 @@ class LinkedInEasyApplier: logger.debug(f"Dropdown options found: {options}") + question_text = question.find_element(By.TAG_NAME, 'label').text.lower() + logger.debug(f"Processing dropdown or combobox question: {question_text}") + current_selection = select.first_selected_option.text logger.debug(f"Current selection: {current_selection}") @@ -738,9 +769,15 @@ class LinkedInEasyApplier: logger.debug(f"Selected new dropdown answer: {answer}") return True - return False + else: + + logger.debug(f"No dropdown found. Logging elements for debugging.") + elements = section.find_elements(By.XPATH, ".//*") + logger.debug(f"Elements found: {[element.tag_name for element in elements]}") + return False + except Exception as e: - logger.warning(f"Failed to handle dropdown or combobox question: {e}") + logger.warning(f"Failed to handle dropdown or combobox question: {e}", exc_info=True) return False def _is_numeric_field(self, field: WebElement) -> bool: diff --git a/src/linkedIn_job_manager.py b/src/linkedIn_job_manager.py index 5b6e53b..9308708 100644 --- a/src/linkedIn_job_manager.py +++ b/src/linkedIn_job_manager.py @@ -5,6 +5,7 @@ import time from itertools import product from pathlib import Path +from inputimeout import inputimeout, TimeoutOccurred from selenium.common.exceptions import NoSuchElementException from selenium.webdriver.common.by import By @@ -45,13 +46,18 @@ class LinkedInJobManager: def set_parameters(self, parameters): logger.debug("Setting parameters for LinkedInJobManager") - self.company_blacklist = parameters.get('companyBlacklist', []) or [] + self.company_blacklist = parameters.get('company_blacklist', []) or [] self.title_blacklist = parameters.get('titleBlacklist', []) or [] self.positions = parameters.get('positions', []) self.locations = parameters.get('locations', []) self.apply_once_at_company = parameters.get('applyOnceAtCompany', False) self.base_search_url = self.get_base_search_url(parameters) self.seen_jobs = [] + + job_applicants_threshold = parameters.get('job_applicants_threshold', {}) + self.min_applicants = job_applicants_threshold.get('min_applicants', 0) + self.max_applicants = job_applicants_threshold.get('max_applicants', float('inf')) + resume_path = parameters.get('uploads', {}).get('resume', None) self.resume_path = Path(resume_path) if resume_path and Path(resume_path).exists() else None self.output_file_directory = Path(parameters['outputFileDirectory']) @@ -109,32 +115,80 @@ class LinkedInJobManager: utils.printyellow("Applying to jobs on this page has been completed!") time_left = minimum_page_time - time.time() + + # Ask user if they want to skip waiting, with timeout if time_left > 0: - utils.printyellow(f"Sleeping for {time_left} seconds.") - logger.debug("Sleeping for %d seconds", time_left) - time.sleep(time_left) - minimum_page_time = time.time() + minimum_time + try: + user_input = inputimeout( + prompt=f"Sleeping for {time_left} seconds. Press 'y' to skip waiting. Timeout 10 seconds : ", + timeout=10).strip().lower() + except TimeoutOccurred: + user_input = '' # No input after timeout + if user_input == 'y': + logger.debug("User chose to skip waiting.") + utils.printyellow("User skipped waiting.") + else: + logger.debug(f"Sleeping for {time_left} seconds as user chose not to skip.") + 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.") - logger.debug("Sleeping for %d seconds", sleep_time) - time.sleep(sleep_time) + try: + user_input = inputimeout( + prompt=f"Sleeping for {sleep_time / 60} minutes. Press 'y' to skip waiting. Timeout 10 seconds : ", + timeout=10).strip().lower() + except TimeoutOccurred: + user_input = '' # No input after timeout + if user_input == 'y': + logger.debug("User chose to skip waiting.") + utils.printyellow("User skipped waiting.") + else: + logger.debug(f"Sleeping for {sleep_time} seconds.") + utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.") + time.sleep(sleep_time) page_sleep += 1 except Exception as e: logger.error("Unexpected error during job search: %s", e) utils.printred(f"Unexpected error: {e}") continue + time_left = minimum_page_time - time.time() + if time_left > 0: - utils.printyellow(f"Sleeping for {time_left} seconds.") - logger.debug("Sleeping for %d seconds", time_left) - time.sleep(time_left) - minimum_page_time = time.time() + minimum_time + try: + user_input = inputimeout( + prompt=f"Sleeping for {time_left} seconds. Press 'y' to skip waiting. Timeout 10 seconds : ", + timeout=10).strip().lower() + except TimeoutOccurred: + user_input = '' # No input after timeout + if user_input == 'y': + logger.debug("User chose to skip waiting.") + utils.printyellow("User skipped waiting.") + else: + logger.debug(f"Sleeping for {time_left} seconds as user chose not to skip.") + 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.") - logger.debug("Sleeping for %d seconds", sleep_time) - time.sleep(sleep_time) + try: + user_input = inputimeout( + prompt=f"Sleeping for {sleep_time / 60} minutes. Press 'y' to skip waiting: ", + timeout=10).strip().lower() + except TimeoutOccurred: + user_input = '' # No input after timeout + if user_input == 'y': + logger.debug("User chose to skip waiting.") + utils.printyellow("User skipped waiting.") + else: + logger.debug(f"Sleeping for {sleep_time} seconds.") + utils.printyellow(f"Sleeping for {sleep_time / 60} minutes.") + time.sleep(sleep_time) page_sleep += 1 def get_jobs_from_page(self): @@ -183,16 +237,82 @@ class LinkedInJobManager: 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) + # 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: utils.printyellow("No job class elements found on page, moving to next page.") logger.debug("No job class elements found on page, skipping") return + job_list = [Job(*self.extract_job_information_from_tile(job_element)) for job_element in job_list_elements] + for job in job_list: + + try: + logger.debug(f"Starting applicant count search for job: {job.title} at {job.company}") + + # Find all job insight elements + job_insight_elements = self.driver.find_elements(By.CLASS_NAME, + "job-details-jobs-unified-top-card__job-insight") + logger.debug(f"Found {len(job_insight_elements)} job insight elements") + + # Initialize applicants_count as None + applicants_count = None + + # Iterate over each job insight element to find the one containing the word "applicant" + for element in job_insight_elements: + logger.debug(f"Checking element text: {element.text}") + if "applicant" in element.text.lower(): + # Found an element containing "applicant" + applicants_text = element.text.strip() + logger.debug(f"Applicants text found: {applicants_text}") + + # Extract numeric digits from the text (e.g., "70 applicants" -> "70") + applicants_count = ''.join(filter(str.isdigit, applicants_text)) + logger.debug(f"Extracted applicants count: {applicants_count}") + + if applicants_count: + if "over" in applicants_text.lower(): + applicants_count = int(applicants_count) + 1 # Handle "over X applicants" + logger.debug(f"Applicants count adjusted for 'over': {applicants_count}") + else: + applicants_count = int(applicants_count) # Convert the extracted number to an integer + break + + # Check if applicants_count is valid (not None) before performing comparisons + if applicants_count is not None: + # Perform the threshold check for applicants count + if applicants_count < self.min_applicants or applicants_count > self.max_applicants: + utils.printyellow( + f"Skipping {job.title} at {job.company} due to applicants count: {applicants_count}") + logger.debug(f"Skipping {job.title} at {job.company}, applicants count: {applicants_count}") + self.write_to_file(job, "skipped_due_to_applicants") + continue # Skip this job if applicants count is outside the threshold + else: + logger.debug(f"Applicants count {applicants_count} is within the threshold") + else: + # If no applicants count was found, log a warning but continue the process + logger.warning( + f"Applicants count not found for {job.title} at {job.company}, continuing with application.") + except NoSuchElementException: + # Log a warning if the job insight elements are not found, but do not stop the job application process + logger.warning( + f"Applicants count elements not found for {job.title} at {job.company}, continuing with application.") + except ValueError as e: + # Handle errors when parsing the applicants count + logger.error(f"Error parsing applicants count for {job.title} at {job.company}: {e}") + except Exception as e: + # Catch any other exceptions to ensure the process continues + logger.error( + f"Unexpected error during applicants count processing for {job.title} at {job.company}: {e}") + + # Continue with the job application process regardless of the applicants count check + logger.debug(f"Continuing with job application for {job.title} at {job.company}") + if self.is_blacklisted(job.title, job.company, job.link): utils.printyellow(f"Blacklisted {job.title} at {job.company}, skipping...") logger.debug("Job blacklisted: %s at %s", job.title, job.company) @@ -200,7 +320,7 @@ class LinkedInJobManager: continue if self.is_already_applied_to_job(job.title, job.company, job.link): self.write_to_file(job, "skipped") - continue + continue if self.is_already_applied_to_company(job.company): self.write_to_file(job, "skipped") continue @@ -307,7 +427,6 @@ class LinkedInJobManager: 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 - is_blacklisted = title_blacklisted or company_blacklisted or link_seen logger.debug("Job blacklisted status: %s", is_blacklisted) return is_blacklisted @@ -322,8 +441,8 @@ class LinkedInJobManager: def is_already_applied_to_company(self, company): if not self.apply_once_at_company: - return False - + return False + output_files = ["success.json"] for file_name in output_files: file_path = self.output_file_directory / file_name @@ -333,9 +452,9 @@ class LinkedInJobManager: existing_data = json.load(f) for applied_job in existing_data: if applied_job['company'].strip().lower() == company.strip().lower(): - utils.printyellow(f"Already applied at {company} (once per company policy), skipping...") + utils.printyellow( + f"Already applied at {company} (once per company policy), skipping...") return True except json.JSONDecodeError: continue return False - diff --git a/src/linkedin-api.py b/src/linkedin-api.py index 37f727d..c061493 100644 --- a/src/linkedin-api.py +++ b/src/linkedin-api.py @@ -1,58 +1,59 @@ -from typing import Dict, List -from linkedin_api import Linkedin -from typing import Optional, Union, Literal -from urllib.parse import quote, urlencode import logging -import json +from typing import Dict, List +from typing import Optional, Union, Literal +from urllib.parse import urlencode + +from linkedin_api import Linkedin # set log to all debug logging.basicConfig(level=logging.INFO) + class LinkedInEvolvedAPI(Linkedin): already_applied_jobs: List[str] = [] - + def __init__(self, username, password): super().__init__(username, password) def search_jobs( - self, - keywords: Optional[str] = None, - companies: Optional[List[str]] = None, - experience: Optional[ - List[ - Union[ - Literal["1"], - Literal["2"], - Literal["3"], - Literal["4"], - Literal["5"], - Literal["6"], + self, + keywords: Optional[str] = None, + companies: Optional[List[str]] = None, + experience: Optional[ + List[ + Union[ + Literal["1"], + Literal["2"], + Literal["3"], + Literal["4"], + Literal["5"], + Literal["6"], + ] ] - ] - ] = None, - job_type: Optional[ - List[ - Union[ - Literal["F"], - Literal["C"], - Literal["P"], - Literal["T"], - Literal["I"], - Literal["V"], - Literal["O"], + ] = None, + job_type: Optional[ + List[ + Union[ + Literal["F"], + Literal["C"], + Literal["P"], + Literal["T"], + Literal["I"], + Literal["V"], + Literal["O"], + ] ] - ] - ] = None, - job_title: Optional[List[str]] = None, - industries: Optional[List[str]] = None, - location_name: Optional[str] = None, - remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None, - listed_at: None | int = None, - distance: Optional[int] = None, - easy_apply: Optional[bool] = True, - limit=-1, - offset=0, - **kwargs, + ] = None, + job_title: Optional[List[str]] = None, + industries: Optional[List[str]] = None, + location_name: Optional[str] = None, + remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None, + listed_at: None | int = None, + distance: Optional[int] = None, + easy_apply: Optional[bool] = True, + limit=-1, + offset=0, + **kwargs, ) -> List[Dict]: """Perform a LinkedIn search for jobs. @@ -154,21 +155,21 @@ class LinkedInEvolvedAPI(Linkedin): e["job_id"] = trackingUrn if e.get("$type") == "com.linkedin.voyager.dash.jobs.JobPosting": new_data.append(e) - + if not new_data: break results.extend(new_data) if ( - (-1 < limit <= len(results)) - or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS + (-1 < limit <= len(results)) + or len(results) / count >= Linkedin._MAX_REPEATED_REQUESTS ) or len(elements) == 0: break self.logger.debug(f"results grew to {len(results)}") return results - - def get_fields_for_easy_apply(self,job_id: str) -> List[Dict]: + + def get_fields_for_easy_apply(self, job_id: str) -> List[Dict]: """Get fields needed for easy apply jobs. :param job_id: Job ID @@ -181,14 +182,12 @@ class LinkedInEvolvedAPI(Linkedin): cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()]) headers: Dict[str, str] = self._headers() - headers["Accept"] = "application/vnd.linkedin.normalized+json+2.1" headers["csrf-token"] = cookies["JSESSIONID"].replace('"', "") headers["Cookie"] = cookie_str headers["Connection"] = "keep-alive" - default_params = { "decorationId": "com.linkedin.voyager.dash.deco.jobs.OnsiteApplyApplication-67", "jobPostingUrn": f"urn:li:fsd_jobPosting:{job_id}", @@ -217,26 +216,26 @@ class LinkedInEvolvedAPI(Linkedin): except ValueError: self.logger.error("Failed to parse JSON response") return [] - + form_components = [] for item in data.get("included", []): - if 'formComponent' in item: + if 'formComponent' in item: urn = item['urn'] try: title = item['title']['text'] except TypeError: title = urn - + form_component_type = list(item['formComponent'].keys())[0] form_component_details = item['formComponent'][form_component_type] - + component_info = { 'title': title, 'urn': urn, 'formComponentType': form_component_type, } - + if 'textSelectableOptions' in form_component_details: options = [ opt['optionText']['text'] for opt in form_component_details['textSelectableOptions'] @@ -244,18 +243,18 @@ class LinkedInEvolvedAPI(Linkedin): component_info['selectableOptions'] = options elif 'selectableOptions' in form_component_details: options = [ - opt['textSelectableOption']['optionText']['text'] + opt['textSelectableOption']['optionText']['text'] for opt in form_component_details['selectableOptions'] ] component_info['selectableOptions'] = options - + form_components.append(component_info) return form_components - - def apply_to_job(self,job_id: str, fields: dict, followCompany: bool = True) -> bool: + + def apply_to_job(self, job_id: str, fields: dict, followCompany: bool = True) -> bool: return False - + # ToDo: Implement apply to job parser first # How need to be implemented: # 1. Get fields for easy apply job from the previous method (get_fields_for_easy_apply) @@ -265,11 +264,11 @@ class LinkedInEvolvedAPI(Linkedin): # {'title': 'Quanti anni di esperienza di lavoro hai con Router?', 'urn': 'urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4013860791,9478711764,numeric)', 'formComponentType': 'singleLineTextFormComponent', 'response': '5'} # To fill, you can temporary use input() function to get the data from the user manually for testing purposes (for the further implementation, the question will be asked to AI implementation and automatically filled) # Build a working payload. - + # EXAMPLE OF WORKING PAYLOAD # 4005350454 is job_id, so need to be replaced with the job_id - #{ + # { # "followCompany": true, # "responses": [ # { @@ -349,23 +348,19 @@ class LinkedInEvolvedAPI(Linkedin): # } # ], # "trackingId": "" - #} + # } # Push the commit to the repository and create a pull request to the v3 branch. - + def set_job_as_applied(self, job_id: str) -> None: self.already_applied_jobs.append(job_id) - - - - - ## EXAMPLE USAGE if __name__ == "__main__": - api: LinkedInEvolvedAPI = LinkedInEvolvedAPI(username="", password="") - jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=100, easy_apply=True, offset=1, listed_at=None) + api: LinkedInEvolvedAPI = LinkedInEvolvedAPI(username="", password="") + jobs = api.search_jobs(keywords="Frontend Developer", location_name="Italia", limit=100, easy_apply=True, offset=1, + listed_at=None) for job in jobs: job_id: str = job["job_id"] print(f"Job ID: {job_id}") @@ -379,7 +374,3 @@ if __name__ == "__main__": for field in fields: print(field) break - - - - \ No newline at end of file diff --git a/src/utils.py b/src/utils.py index 44d022f..f4e4d4a 100644 --- a/src/utils.py +++ b/src/utils.py @@ -90,7 +90,13 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse return position = start + previous_position = None # Tracking the previous position to avoid duplicate scrolls while (step > 0 and position < end) or (step < 0 and position > end): + if position == previous_position: + # Avoid re-scrolling to the same position + logger.debug("Stopping scroll as position hasn't changed: %d", position) + break + try: driver.execute_script(script_scroll_to, scrollable_element, position) logger.debug("Scrolled to position: %d", position) @@ -98,11 +104,15 @@ def scroll_slow(driver, scrollable_element, start=0, end=3600, step=300, reverse logger.error("Error during scrolling: %s", e) print(f"Error during scrolling: {e}") + previous_position = position position += step + + # Decrease the step but ensure it doesn't reverse direction step = max(10, abs(step) - 10) * (-1 if reverse else 1) time.sleep(random.uniform(0.6, 1.5)) + # Ensure the final scroll position is correct driver.execute_script(script_scroll_to, scrollable_element, end) logger.debug("Scrolled to final position: %d", end) time.sleep(0.5)