afiqzudinhadi/linkedIn_auto_jobs_applier_with_AI
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

reformat code

Browse source
This commit is contained in:
queukat 2024-09-08 17:46:00 +03:00
parent ad85b9587f
commit 5efe6c3048
4 changed files with 82 additions and 98 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/gpt.py
View file

@ -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,18 +16,19 @@ 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
@ -39,6 +40,7 @@ class OpenAIModel(AIModel):
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
@ -49,6 +51,7 @@ class ClaudeModel(AIModel):
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)
@ -80,8 +84,8 @@ class AIAdapter:
def invoke(self, prompt: str) -> str:
return self.model.invoke(prompt)
class LLMLogger:
class LLMLogger:
def __init__(self, llm: Union[OpenAIModel, OllamaModel, ClaudeModel]):
@ -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,7 +456,9 @@ 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.")

11
src/linkedIn_easy_applier.py
View file

@ -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,7 +59,6 @@ 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):
current_url = self.driver.current_url
@ -79,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 +164,6 @@ class LinkedInEasyApplier:
if method.get('find_elements'):
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:
@ -541,7 +536,6 @@ class LinkedInEasyApplier:
wrapped_lines.append(line)
return wrapped_lines
lines = split_text_by_width(cover_letter_text, "Helvetica", 12, max_width)
for line in lines:
@ -567,7 +561,6 @@ class LinkedInEasyApplier:
logger.error(f"Traceback: {tb_str}")
raise
file_size = os.path.getsize(file_path_pdf)
max_file_size = 2 * 1024 * 1024 # 2 MB
logger.debug(f"Cover letter file size: {file_size} bytes")
@ -670,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}'")
@ -697,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.")
@ -730,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")
@ -752,7 +742,6 @@ 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}")

4
src/linkedIn_job_manager.py
View file

@ -452,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

25
src/linkedin-api.py
View file

@ -1,13 +1,14 @@
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] = []
@ -182,13 +183,11 @@ class LinkedInEvolvedAPI(Linkedin):
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}",
@ -357,15 +356,11 @@ class LinkedInEvolvedAPI(Linkedin):
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)
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