Merge branch 'v3' into v3

src/gpt.py

@@ -4,6 +4,8 @@ import re
 import textwrap
 import time
 from datetime import datetime
+from abc import ABC, abstractmethod
+from typing import Dict, List, Union
 from pathlib import Path
 from typing import Dict, List
 
@@ -21,11 +23,68 @@ 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)
+
+    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
+        self.model = ChatOllama(model=llm_model, base_url=llm_api_url)
+
+    def invoke(self, prompt: str) -> str:
+        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)
+
+    def _create_model(self, config: dict, api_key: str) -> AIModel:
+        llm_model_type = config['llm_model_type']
+        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":
+            return ClaudeModel(api_key, llm_model, llm_api_url)
+        elif llm_model_type == "ollama":
+            return OllamaModel(api_key, llm_model, llm_api_url)
+        else:
+            raise ValueError(f"Unsupported model type: {model_type}")
+
+    def invoke(self, prompt: str) -> str:
+        return self.model.invoke(prompt)
 
 class LLMLogger:
 
-    def __init__(self, llm: ChatOpenAI):
-        logger.debug("Initializing LLMLogger with LLM: %s", llm)
+    
+    def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]):
+
         self.llm = llm
         logger.debug("LLMLogger successfully initialized with LLM: %s", llm)
 
@@ -129,12 +188,15 @@ class LLMLogger:
 
 
 class LoggerChatModel:
-    def __init__(self, llm: ChatOpenAI):
-        logger.debug("Initializing LoggerChatModel with LLM: %s", llm)
+
+
+    def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]):
+
         self.llm = llm
         logger.debug("LoggerChatModel successfully initialized with LLM: %s", llm)
 
     def __call__(self, messages: List[Dict[str, str]]) -> str:
+
         logger.debug("Entering __call__ method with messages: %s", messages)
         while True:
             try:
@@ -185,6 +247,7 @@ class LoggerChatModel:
                 time.sleep(30)
                 continue
 
+
     def parse_llmresult(self, llmresult: AIMessage) -> Dict[str, Dict]:
         logger.debug("Parsing LLM result: %s", llmresult)
 
@@ -223,11 +286,10 @@ class LoggerChatModel:
 
 
 class GPTAnswerer:
-    def __init__(self, openai_api_key):
-        self.llm_cheap = LoggerChatModel(
-            ChatOpenAI(model_name="gpt-4o-mini", openai_api_key=openai_api_key, temperature=0.4)
-        )
-        logger.debug("GPTAnswerer initialized with API key")
+
+    def __init__(self, config, llm_api_key):
+        self.ai_adapter = AIAdapter(config, llm_api_key)
+        self.llm_cheap = LoggerChatModel(self.ai_adapter)
 
     @property
     def job_description(self):
@@ -391,8 +453,13 @@ class GPTAnswerer:
         prompt = ChatPromptTemplate.from_template(section_prompt)
         chain = prompt | self.llm_cheap | StrOutputParser()
         output = chain.invoke({"question": question})
-        logger.debug("Section determined from question: %s", output)
-        section_name = output.lower().replace(" ", "_")
+
+        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})

src/linkedIn_authenticator.py

@@ -39,6 +39,9 @@ class LinkedInAuthenticator:
     def handle_login(self):
         logger.info("Navigating to the LinkedIn login page...")
         self.driver.get("https://www.linkedin.com/login")
+        if 'feed' in self.driver.current_url:
+            print("User is already logged in.")
+            return
         try:
             self.enter_credentials()
             self.submit_login_form()

src/linkedIn_easy_applier.py

@@ -34,8 +34,10 @@ class LinkedInEasyApplier:
         self.gpt_answerer = gpt_answerer
         self.resume_generator_manager = resume_generator_manager
         self.all_data = self._load_questions_from_json()
+
         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)
@@ -58,6 +60,7 @@ 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.
         В случае редиректа возвращает пользователя на исходную страницу вакансии."""
@@ -77,6 +80,7 @@ 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)
 
@@ -205,6 +209,7 @@ 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:
@@ -596,6 +601,7 @@ class LinkedInEasyApplier:
             for item in self.all_data:
                 if self._sanitize_text(question_text) in item['question'] and item['type'] == 'radio':
                     existing_answer = item
+
                     break
             if existing_answer:
                 self._select_radio(radios, existing_answer['answer'])
@@ -626,6 +632,7 @@ 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}'")
 
@@ -652,6 +659,7 @@ 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.")
@@ -677,12 +685,14 @@ class LinkedInEasyApplier:
             for item in self.all_data:
                 if self._sanitize_text(question_text) in item['question'] and item['type'] == 'date':
                     existing_answer = item
+
                     break
             if existing_answer:
                 self._enter_text(date_field, existing_answer['answer'])
                 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,6 +711,7 @@ class LinkedInEasyApplier:
                 dropdown = dropdowns[0]
                 select = Select(dropdown)
                 options = [option.text for option in select.options]
+
                 logger.debug(f"Dropdown options found: {options}")
 
                 current_selection = select.first_selected_option.text
@@ -720,6 +731,7 @@ class LinkedInEasyApplier:
                     return True
 
                 logger.debug(f"No existing answer found, querying model for: {question_text}")
+
                 answer = self.gpt_answerer.answer_question_from_options(question_text, options)
                 self._save_questions_to_json({'type': 'dropdown', 'question': question_text, 'answer': answer})
                 self._select_dropdown_option(dropdown, answer)

src/linkedIn_job_manager.py

@@ -49,6 +49,7 @@ class LinkedInJobManager:
         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 = []
         resume_path = parameters.get('uploads', {}).get('resume', None)
@@ -200,6 +201,12 @@ class LinkedInJobManager:
                 logger.debug("Job blacklisted: %s at %s", job.title, job.company)
                 self.write_to_file(job, "skipped")
                 continue
+            if self.is_already_applied_to_job(job.title, job.company, job.link):
+                self.write_to_file(job, "skipped")
+                continue            
+            if self.is_already_applied_to_company(job.company):
+                self.write_to_file(job, "skipped")
+                continue
             try:
                 if job.apply_method not in {"Continue", "Applied", "Apply"}:
                     self.easy_applier_component.job_apply(job)
@@ -303,6 +310,35 @@ 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
+
+        return title_blacklisted or company_blacklisted or link_seen
+
+    def is_already_applied_to_job(self, job_title, company, link):
+        link_seen = link in self.seen_jobs
+        if link_seen:
+            utils.printyellow(f"Already applied to job: {job_title} at {company}, skipping...")
+        return link_seen
+
+    def is_already_applied_to_company(self, company):
+        if not self.apply_once_at_company:
+            return False  
+        
+        output_files = ["success.json"]
+        for file_name in output_files:
+            file_path = self.output_file_directory / file_name
+            if file_path.exists():
+                with open(file_path, 'r', encoding='utf-8') as f:
+                    try:
+                        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...")
+                                return True
+                    except json.JSONDecodeError:
+                        continue
+        return False
+

src/linkedin-api.py

@@ -1,9 +1,16 @@
 from typing import Dict, List
 from linkedin_api import Linkedin
 from typing import Optional, Union, Literal
-from urllib.parse import urlencode
+from urllib.parse import quote, urlencode
+import logging
+import json
+
+# 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)
 
@@ -40,7 +47,7 @@ class LinkedInEvolvedAPI(Linkedin):
         industries: Optional[List[str]] = None,
         location_name: Optional[str] = None,
         remote: Optional[List[Union[Literal["1"], Literal["2"], Literal["3"]]]] = None,
-        listed_at=24 * 60 * 60,
+        listed_at: None | int = None,
         distance: Optional[int] = None,
         easy_apply: Optional[bool] = True,
         limit=-1,
@@ -65,8 +72,8 @@ class LinkedInEvolvedAPI(Linkedin):
         :type location_name: str, optional
         :param remote: Filter for remote jobs, onsite or hybrid. onsite:"1", remote:"2", hybrid:"3"
         :type remote: list, optional
-        :param listed_at: maximum number of seconds passed since job posting. 86400 will filter job postings posted in last 24 hours.
-        :type listed_at: int/str, optional. Default value is equal to 24 hours.
+        :param listed_at: maximum number of seconds passed since job posting. 86400 will filter job postings posted in last 24 hours, default is None
+        :type listed_at: int or none, if none, no filter applied, otherwise, filter applied in seconds
         :param distance: maximum distance from location in miles
         :type distance: int/str, optional. If not specified, None or 0, the default value of 25 miles applied.
         :param easy_apply: filter for jobs that are easy to apply to
@@ -106,9 +113,10 @@ class LinkedInEvolvedAPI(Linkedin):
         if remote:
             query["selectedFilters"]["workplaceType"] = f"List({','.join(remote)})"
         if easy_apply:
-            query["selectedFilters"]["easyApply"] = "List(true)"
+            query["selectedFilters"]["applyWithLinkedin"] = "List(true)"
 
-        query["selectedFilters"]["timePostedRange"] = f"List(r{listed_at})"
+        if listed_at:
+            query["selectedFilters"]["timePostedRange"] = f"List(r{listed_at})"
         query["spellCorrectionEnabled"] = "true"
 
         query_string = (
@@ -137,7 +145,6 @@ class LinkedInEvolvedAPI(Linkedin):
                 headers={"accept": "application/vnd.linkedin.normalized+json+2.1"},
             )
             data = res.json()
-
             elements = data.get("included", [])
             new_data = []
             for e in elements:
@@ -160,9 +167,219 @@ class LinkedInEvolvedAPI(Linkedin):
             self.logger.debug(f"results grew to {len(results)}")
 
         return results
-
     
+    def get_fields_for_easy_apply(self,job_id: str) -> List[Dict]:
+        """Get fields needed for easy apply jobs.
+
+        :param job_id: Job ID
+        :type job_id: str
+        :return: Fields
+        :rtype: dict
+        """
+
+        cookies = self.client.session.cookies.get_dict()
+        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}",
+            "q": "jobPosting",
+        }
+
+        default_params = urlencode(default_params)
+        res = self._fetch(
+            f"/voyagerJobsDashOnsiteApplyApplication?{default_params}",
+            headers=headers,
+            cookies=cookies,
+        )
+
+        match res.status_code:
+            case 200:
+                pass
+            case 409:
+                self.logger.error("Failed to fetch fields for easy apply job because already applied to this job!")
+                return []
+            case _:
+                self.logger.error("Failed to fetch fields for easy apply job")
+                return []
+
+        try:
+            data = res.json()
+        except ValueError:
+            self.logger.error("Failed to parse JSON response")
+            return []
+        
+        form_components = []
+
+        for item in data.get("included", []):
+            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']
+                    ]
+                    component_info['selectableOptions'] = options
+                elif 'selectableOptions' in form_component_details:
+                    options = [
+                        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:
+        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)
+        # 2. Fill the fields with the data adding a response parameter in the specific field in the dict object, for example:
+        # {'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'}
+        # Became:
+        # {'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": [
+        #        {
+        #            "formElementUrn": "urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3497278561,multipleChoice)",
+        #            "formElementInputValues": [
+        #                {
+        #                    "entityInputValue": {
+        #                        "inputEntityName": "email@gmail.com"
+        #                    }
+        #                }
+        #            ]
+        #        },
+        #        {
+        #            "formElementUrn": "urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3497278545,phoneNumber~country)",
+        #            "formElementInputValues": [
+        #                {
+        #                    "entityInputValue": {
+        #                        "inputEntityName": "Italy (+39)",
+        #                        "inputEntityUrn": "urn:li:country:it"
+        #                    }
+        #                }
+        #            ]
+        #        },
+        #        {
+        #            "formElementUrn": "urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3497278545,phoneNumber~nationalNumber)",
+        #            "formElementInputValues": [
+        #                {
+        #                    "textInputValue": "3333333"
+        #                }
+        #            ]
+        #        },
+        #        {
+        #            "formElementUrn": "urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3497278529,multipleChoice)",
+        #            "formElementInputValues": [
+        #                {
+        #                    "entityInputValue": {
+        #                        "inputEntityName": "Native or bilingual"
+        #                    }
+        #                }
+        #            ]
+        #        },
+        #        {
+        #            "formElementUrn": "urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3497278537,numeric)",
+        #            "formElementInputValues": [
+        #                {
+        #                    "textInputValue": "0"
+        #                }
+        #            ]
+        #        },
+        #        {
+        #            "formElementUrn": "urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3498546713,multipleChoice)",
+        #            "formElementInputValues": [
+        #                {
+        #                    "entityInputValue": {
+        #                        "inputEntityName": "No"
+        #                    }
+        #                }
+        #            ]
+        #        },
+        #        {
+        #            "formElementUrn": "urn:li:fsd_formElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3497278521,multipleChoice)",
+        #            "formElementInputValues": [
+        #                {
+        #                    "entityInputValue": {
+        #                        "inputEntityName": "No"
+        #                    }
+        #                }
+        #            ]
+        #        }
+        #    ],
+        #    "referenceId": "",
+        #    "trackingCode": "d_flagship3_search_srp_jobs",
+        #    "fileUploadResponses": [
+        #        {
+        #            "inputUrn": "urn:li:fsd_resume:/##todo##",
+        #            "formElementUrn": "urn:li:fsu_jobApplicationFileUploadFormElement:urn:li:jobs_applyformcommon_easyApplyFormElement:(4005350454,3497278553,document)"
+        #        }
+        #    ],
+        #    "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)
+    for job in jobs:
+        job_id: str = job["job_id"]
+        print(f"Job ID: {job_id}")
+        continue
+
+        if job_id in api.already_applied_jobs:
+            logging.info(f"Already applied to job {job_id}, skipping it")
+            continue
+
+        fields = api.get_fields_for_easy_apply(job_id)
+        for field in fields:
+            print(field)
+        break
+
+